agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

app-server: Replay pending item requests on thread/resume (#12560)

Browse Source 
Replay pending client requests after `thread/resume` and emit resolved
notifications when those requests clear so approval/input UI state stays
in sync after reconnects and across subscribed clients.

Affected RPCs:
- `item/commandExecution/requestApproval`
- `item/fileChange/requestApproval`
- `item/tool/requestUserInput`

Motivation:
- Resumed clients need to see pending approval/input requests that were
already outstanding before the reconnect.
- Clients also need an explicit signal when a pending request resolves
or is cleared so stale UI can be removed on turn start, completion, or
interruption.

Implementation notes:
- Use pending client requests from `OutgoingMessageSender` in order to
replay them after `thread/resume` attaches the connection, using
original request ids.
- Emit `serverRequest/resolved` when pending requests are answered
or cleared by lifecycle cleanup.
- Update the app-server protocol schema, generated TypeScript bindings,
and README docs for the replay/resolution flow.

High-level test plan:
- Added automated coverage for replaying pending command execution and
file change approval requests on `thread/resume`.
- Added automated coverage for resolved notifications in command
approval, file change approval, request_user_input, turn start, and turn
interrupt flows.
- Verified schema/docs updates in the relevant protocol and app-server
tests.

Manual testing:
- Tested reconnect/resume with multiple connections.
- Confirmed state stayed in sync between connections.
This commit is contained in:
Ruslan Nigmatullin
2026-02-27 12:45:59 -08:00
committed by GitHub
Unverified
parent 66b0adb34c
commit 69d7a456bb
22 changed files with 1416 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/bespoke_event_handling.rs
+65 -7
View File
@@ -6,6 +6,8 @@ use crate::error_code::INTERNAL_ERROR_CODE;
use crate::error_code::INVALID_REQUEST_ERROR_CODE;
use crate::outgoing_message::ClientRequestResult;
use crate::outgoing_message::ThreadScopedOutgoingMessageSender;
use crate::server_request_error::is_turn_transition_server_request_error;
use crate::thread_state::ThreadListenerCommand;
use crate::thread_state::ThreadState;
use crate::thread_state::TurnSummary;
use crate::thread_status::ThreadWatchActiveGuard;
@@ -56,6 +58,7 @@ use codex_app_server_protocol::RawResponseItemCompletedNotification;
use codex_app_server_protocol::ReasoningSummaryPartAddedNotification;
use codex_app_server_protocol::ReasoningSummaryTextDeltaNotification;
use codex_app_server_protocol::ReasoningTextDeltaNotification;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequestPayload;
use codex_app_server_protocol::TerminalInteractionNotification;
@@ -132,6 +135,38 @@ struct CommandExecutionCompletionItem {
command_actions: Vec<V2ParsedCommand>,
}
async fn resolve_server_request_on_thread_listener(
thread_state: &Arc<Mutex<ThreadState>>,
request_id: RequestId,
) {
let (completion_tx, completion_rx) = oneshot::channel();
let listener_command_tx = {
let state = thread_state.lock().await;
state.listener_command_tx()
};
let Some(listener_command_tx) = listener_command_tx else {
error!("failed to remove pending client request: thread listener is not running");
return;
};
if listener_command_tx
.send(ThreadListenerCommand::ResolveServerRequest {
request_id,
completion_tx,
})
.is_err()
{
error!(
"failed to remove pending client request: thread listener command channel is closed"
);
return;
}
if let Err(err) = completion_rx.await {
error!("failed to remove pending client request: {err}");
}
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn apply_bespoke_event_handling(
event: Event,
@@ -151,11 +186,15 @@ pub(crate) async fn apply_bespoke_event_handling(
} = event;
match msg {
EventMsg::TurnStarted(_) => {
// While not technically necessary as it was already done on TurnComplete, be extra cautios and abort any pending server requests.
outgoing.abort_pending_server_requests().await;
thread_watch_manager
.note_turn_started(&conversation_id.to_string())
.await;
}
EventMsg::TurnComplete(_ev) => {
// All per-thread requests are bound to a turn, so abort them.
outgoing.abort_pending_server_requests().await;
let turn_failed = thread_state.lock().await.turn_summary.last_error.is_some();
thread_watch_manager
.note_turn_completed(&conversation_id.to_string(), turn_failed)
@@ -263,7 +302,7 @@ pub(crate) async fn apply_bespoke_event_handling(
reason,
grant_root,
};
let rx = outgoing
let (_pending_request_id, rx) = outgoing
.send_request(ServerRequestPayload::ApplyPatchApproval(params))
.await;
tokio::spawn(async move {
@@ -307,7 +346,7 @@ pub(crate) async fn apply_bespoke_event_handling(
reason,
grant_root,
};
let rx = outgoing
let (pending_request_id, rx) = outgoing
.send_request(ServerRequestPayload::FileChangeRequestApproval(params))
.await;
tokio::spawn(async move {
@@ -316,6 +355,7 @@ pub(crate) async fn apply_bespoke_event_handling(
conversation_id,
item_id,
patch_changes,
pending_request_id,
rx,
conversation,
outgoing,
@@ -362,7 +402,7 @@ pub(crate) async fn apply_bespoke_event_handling(
reason,
parsed_cmd,
};
let rx = outgoing
let (_pending_request_id, rx) = outgoing
.send_request(ServerRequestPayload::ExecCommandApproval(params))
.await;
tokio::spawn(async move {
@@ -435,7 +475,7 @@ pub(crate) async fn apply_bespoke_event_handling(
proposed_network_policy_amendments: proposed_network_policy_amendments_v2,
available_decisions: Some(available_decisions),
};
let rx = outgoing
let (pending_request_id, rx) = outgoing
.send_request(ServerRequestPayload::CommandExecutionRequestApproval(
params,
))
@@ -447,6 +487,7 @@ pub(crate) async fn apply_bespoke_event_handling(
approval_id,
call_id,
completion_item,
pending_request_id,
rx,
conversation,
outgoing,
@@ -489,14 +530,16 @@ pub(crate) async fn apply_bespoke_event_handling(
item_id: request.call_id,
questions,
};
let rx = outgoing
let (pending_request_id, rx) = outgoing
.send_request(ServerRequestPayload::ToolRequestUserInput(params))
.await;
tokio::spawn(async move {
on_request_user_input_response(
event_turn_id,
pending_request_id,
rx,
conversation,
thread_state,
user_input_guard,
)
.await;
@@ -550,7 +593,7 @@ pub(crate) async fn apply_bespoke_event_handling(
tool: tool.clone(),
arguments: arguments.clone(),
};
let rx = outgoing
let (_pending_request_id, rx) = outgoing
.send_request(ServerRequestPayload::DynamicToolCall(params))
.await;
tokio::spawn(async move {
@@ -1136,6 +1179,7 @@ pub(crate) async fn apply_bespoke_event_handling(
// Until we migrate the core to be aware of a first class FileChangeItem
// and emit the corresponding EventMsg, we repurpose the call_id as the item_id.
let item_id = patch_begin_event.call_id.clone();
let changes = convert_patch_changes(&patch_begin_event.changes);
let first_start = {
let mut state = thread_state.lock().await;
@@ -1147,7 +1191,7 @@ pub(crate) async fn apply_bespoke_event_handling(
if first_start {
let item = ThreadItem::FileChange {
id: item_id.clone(),
changes: convert_patch_changes(&patch_begin_event.changes),
changes,
status: PatchApplyStatus::InProgress,
};
let notification = ItemStartedNotification {
@@ -1329,6 +1373,8 @@ pub(crate) async fn apply_bespoke_event_handling(
}
// If this is a TurnAborted, reply to any pending interrupt requests.
EventMsg::TurnAborted(turn_aborted_event) => {
// All per-thread requests are bound to a turn, so abort them.
outgoing.abort_pending_server_requests().await;
let pending = {
let mut state = thread_state.lock().await;
std::mem::take(&mut state.pending_interrupts)
@@ -1725,6 +1771,7 @@ async fn on_patch_approval_response(
let response = receiver.await;
let value = match response {
Ok(Ok(value)) => value,
Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
Ok(Err(err)) => {
error!("request failed with client error: {err:?}");
if let Err(submit_err) = codex
@@ -1781,6 +1828,7 @@ async fn on_exec_approval_response(
let response = receiver.await;
let value = match response {
Ok(Ok(value)) => value,
Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
Ok(Err(err)) => {
error!("request failed with client error: {err:?}");
return;
@@ -1816,14 +1864,18 @@ async fn on_exec_approval_response(
async fn on_request_user_input_response(
event_turn_id: String,
pending_request_id: RequestId,
receiver: oneshot::Receiver<ClientRequestResult>,
conversation: Arc<CodexThread>,
thread_state: Arc<Mutex<ThreadState>>,
user_input_guard: ThreadWatchActiveGuard,
) {
let response = receiver.await;
resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await;
drop(user_input_guard);
let value = match response {
Ok(Ok(value)) => value,
Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
Ok(Err(err)) => {
error!("request failed with client error: {err:?}");
let empty = CoreRequestUserInputResponse {
@@ -1934,6 +1986,7 @@ async fn on_file_change_request_approval_response(
conversation_id: ThreadId,
item_id: String,
changes: Vec<FileUpdateChange>,
pending_request_id: RequestId,
receiver: oneshot::Receiver<ClientRequestResult>,
codex: Arc<CodexThread>,
outgoing: ThreadScopedOutgoingMessageSender,
@@ -1941,6 +1994,7 @@ async fn on_file_change_request_approval_response(
permission_guard: ThreadWatchActiveGuard,
) {
let response = receiver.await;
resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await;
drop(permission_guard);
let (decision, completion_status) = match response {
Ok(Ok(value)) => {
@@ -1958,6 +2012,7 @@ async fn on_file_change_request_approval_response(
// Only short-circuit on declines/cancels/failures.
(decision, completion_status)
}
Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
Ok(Err(err)) => {
error!("request failed with client error: {err:?}");
(ReviewDecision::Denied, Some(PatchApplyStatus::Failed))
@@ -1999,6 +2054,7 @@ async fn on_command_execution_request_approval_response(
approval_id: Option<String>,
item_id: String,
completion_item: Option<CommandExecutionCompletionItem>,
pending_request_id: RequestId,
receiver: oneshot::Receiver<ClientRequestResult>,
conversation: Arc<CodexThread>,
outgoing: ThreadScopedOutgoingMessageSender,
@@ -2006,6 +2062,7 @@ async fn on_command_execution_request_approval_response(
permission_guard: ThreadWatchActiveGuard,
) {
let response = receiver.await;
resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await;
drop(permission_guard);
let (decision, completion_status) = match response {
Ok(Ok(value)) => {
@@ -2057,6 +2114,7 @@ async fn on_command_execution_request_approval_response(
};
(decision, completion_status)
}
Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
Ok(Err(err)) => {
error!("request failed with client error: {err:?}");
(ReviewDecision::Denied, Some(CommandExecutionStatus::Failed))
codex-rs/app-server/src/codex_message_processor.rs
+146 -19
View File
@@ -100,6 +100,7 @@ use codex_app_server_protocol::NewConversationResponse;
use codex_app_server_protocol::ProductSurface as ApiProductSurface;
use codex_app_server_protocol::RemoveConversationListenerParams;
use codex_app_server_protocol::RemoveConversationSubscriptionResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ResumeConversationParams;
use codex_app_server_protocol::ResumeConversationResponse;
use codex_app_server_protocol::ReviewDelivery as ApiReviewDelivery;
@@ -112,6 +113,7 @@ use codex_app_server_protocol::SendUserMessageResponse;
use codex_app_server_protocol::SendUserTurnParams;
use codex_app_server_protocol::SendUserTurnResponse;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequestResolvedNotification;
use codex_app_server_protocol::SessionConfiguredNotification;
use codex_app_server_protocol::SetDefaultModelParams;
use codex_app_server_protocol::SetDefaultModelResponse;
@@ -297,8 +299,12 @@ use tracing::info;
use tracing::warn;
use uuid::Uuid;
#[cfg(test)]
use codex_app_server_protocol::ServerRequest;
use crate::filters::compute_source_filters;
use crate::filters::source_kind_matches;
use crate::thread_state::ThreadListenerCommand;
use crate::thread_state::ThreadState;
use crate::thread_state::ThreadStateManager;
@@ -3221,11 +3227,11 @@ impl CodexMessageProcessor {
};
let command = crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse(
crate::thread_state::PendingThreadResumeRequest {
Box::new(crate::thread_state::PendingThreadResumeRequest {
request_id: request_id.clone(),
rollout_path,
config_snapshot,
},
}),
);
if listener_command_tx.send(command).is_err() {
let err = JSONRPCErrorError {
@@ -4844,7 +4850,9 @@ impl CodexMessageProcessor {
async fn finalize_thread_teardown(&mut self, thread_id: ThreadId) {
self.pending_thread_unloads.lock().await.remove(&thread_id);
self.outgoing.cancel_requests_for_thread(thread_id).await;
self.outgoing
.cancel_requests_for_thread(thread_id, None)
.await;
self.thread_state_manager
.remove_thread_state(thread_id)
.await;
@@ -4905,7 +4913,9 @@ impl CodexMessageProcessor {
self.pending_thread_unloads.lock().await.insert(thread_id);
// Any pending app-server -> client requests for this thread can no longer be
// answered; cancel their callbacks before shutdown/unload.
self.outgoing.cancel_requests_for_thread(thread_id).await;
self.outgoing
.cancel_requests_for_thread(thread_id, None)
.await;
self.thread_state_manager
.remove_thread_state(thread_id)
.await;
@@ -6507,21 +6517,15 @@ impl CodexMessageProcessor {
let Some(listener_command) = listener_command else {
break;
};
match listener_command {
crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse(
resume_request,
) => {
handle_pending_thread_resume_request(
conversation_id,
codex_home.as_path(),
&thread_state,
&thread_watch_manager,
&outgoing_for_task,
resume_request,
)
.await;
}
}
handle_thread_listener_command(
conversation_id,
codex_home.as_path(),
&thread_state,
&thread_watch_manager,
&outgoing_for_task,
listener_command,
)
.await;
}
}
}
@@ -6830,6 +6834,37 @@ impl CodexMessageProcessor {
}
}
async fn handle_thread_listener_command(
conversation_id: ThreadId,
codex_home: &Path,
thread_state: &Arc<Mutex<ThreadState>>,
thread_watch_manager: &ThreadWatchManager,
outgoing: &Arc<OutgoingMessageSender>,
listener_command: ThreadListenerCommand,
) {
match listener_command {
ThreadListenerCommand::SendThreadResumeResponse(resume_request) => {
handle_pending_thread_resume_request(
conversation_id,
codex_home,
thread_state,
thread_watch_manager,
outgoing,
*resume_request,
)
.await;
}
ThreadListenerCommand::ResolveServerRequest {
request_id,
completion_tx,
} => {
resolve_pending_server_request(conversation_id, thread_state, outgoing, request_id)
.await;
let _ = completion_tx.send(());
}
}
}
async fn handle_pending_thread_resume_request(
conversation_id: ThreadId,
codex_home: &Path,
@@ -6918,9 +6953,36 @@ async fn handle_pending_thread_resume_request(
reasoning_effort,
};
outgoing.send_response(request_id, response).await;
outgoing
.replay_requests_to_connection_for_thread(connection_id, conversation_id)
.await;
thread_state.lock().await.add_connection(connection_id);
}
async fn resolve_pending_server_request(
conversation_id: ThreadId,
thread_state: &Arc<Mutex<ThreadState>>,
outgoing: &Arc<OutgoingMessageSender>,
request_id: RequestId,
) {
let thread_id = conversation_id.to_string();
let subscribed_connection_ids = thread_state.lock().await.subscribed_connection_ids();
let outgoing = ThreadScopedOutgoingMessageSender::new(
outgoing.clone(),
subscribed_connection_ids,
conversation_id,
);
outgoing
.send_server_notification(ServerNotification::ServerRequestResolved(
ServerRequestResolvedNotification {
thread_id,
request_id,
},
))
.await;
}
async fn load_thread_for_running_resume_response(
conversation_id: ThreadId,
rollout_path: &Path,
@@ -7668,7 +7730,11 @@ pub(crate) fn summary_to_thread(summary: ConversationSummary) -> Thread {
#[cfg(test)]
mod tests {
use super::*;
use crate::outgoing_message::OutgoingEnvelope;
use crate::outgoing_message::OutgoingMessage;
use anyhow::Result;
use codex_app_server_protocol::ServerRequestPayload;
use codex_app_server_protocol::ToolRequestUserInputParams;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use pretty_assertions::assert_eq;
@@ -7862,6 +7928,67 @@ mod tests {
Ok(())
}
#[tokio::test]
async fn aborting_pending_request_clears_pending_state() -> Result<()> {
let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
let thread_state = Arc::new(Mutex::new(ThreadState::default()));
let connection_id = ConnectionId(7);
thread_state.lock().await.add_connection(connection_id);
let (outgoing_tx, mut outgoing_rx) = tokio::sync::mpsc::channel(8);
let outgoing = Arc::new(OutgoingMessageSender::new(outgoing_tx));
let thread_outgoing = ThreadScopedOutgoingMessageSender::new(
outgoing.clone(),
vec![connection_id],
thread_id,
);
let (request_id, client_request_rx) = thread_outgoing
.send_request(ServerRequestPayload::ToolRequestUserInput(
ToolRequestUserInputParams {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
item_id: "call-1".to_string(),
questions: vec![],
},
))
.await;
thread_outgoing.abort_pending_server_requests().await;
let request_message = outgoing_rx.recv().await.expect("request should be sent");
let OutgoingEnvelope::ToConnection {
connection_id: request_connection_id,
message:
OutgoingMessage::Request(ServerRequest::ToolRequestUserInput {
request_id: sent_request_id,
..
}),
} = request_message
else {
panic!("expected tool request to be sent to the subscribed connection");
};
assert_eq!(request_connection_id, connection_id);
assert_eq!(sent_request_id, request_id);
let response = client_request_rx
.await
.expect("callback should be resolved");
let error = response.expect_err("request should be aborted during cleanup");
assert_eq!(
error.message,
"client request resolved because the turn state was changed"
);
assert_eq!(error.data, Some(json!({ "reason": "turnTransition" })));
assert!(
outgoing
.pending_requests_for_thread(thread_id)
.await
.is_empty()
);
assert!(outgoing_rx.try_recv().is_err());
Ok(())
}
#[test]
fn summary_from_state_db_metadata_preserves_agent_nickname() -> Result<()> {
let conversation_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
codex-rs/app-server/src/dynamic_tools.rs
+2
View File
@@ -9,6 +9,7 @@ use tokio::sync::oneshot;
use tracing::error;
use crate::outgoing_message::ClientRequestResult;
use crate::server_request_error::is_turn_transition_server_request_error;
pub(crate) async fn on_call_response(
call_id: String,
@@ -18,6 +19,7 @@ pub(crate) async fn on_call_response(
let response = receiver.await;
let (response, _error) = match response {
Ok(Ok(value)) => decode_response(value),
Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
Ok(Err(err)) => {
error!("request failed with client error: {err:?}");
fallback_response("dynamic tool request failed")
codex-rs/app-server/src/lib.rs
+1
View File
@@ -63,6 +63,7 @@ mod fuzzy_file_search;
mod message_processor;
mod models;
mod outgoing_message;
mod server_request_error;
mod thread_state;
mod thread_status;
mod transport;
codex-rs/app-server/src/outgoing_message.rs
+244 -66
View File
@@ -17,6 +17,7 @@ use tokio::sync::oneshot;
use tracing::warn;
use crate::error_code::INTERNAL_ERROR_CODE;
use crate::server_request_error::TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON;
#[cfg(test)]
use codex_protocol::account::PlanType;
@@ -62,6 +63,7 @@ pub(crate) struct ThreadScopedOutgoingMessageSender {
struct PendingCallbackEntry {
callback: oneshot::Sender<ClientRequestResult>,
thread_id: Option<ThreadId>,
request: ServerRequest,
}
impl ThreadScopedOutgoingMessageSender {
@@ -80,12 +82,12 @@ impl ThreadScopedOutgoingMessageSender {
pub(crate) async fn send_request(
&self,
payload: ServerRequestPayload,
) -> oneshot::Receiver<ClientRequestResult> {
) -> (RequestId, oneshot::Receiver<ClientRequestResult>) {
self.outgoing
.send_request_to_thread_connections(
self.thread_id,
self.connection_ids.as_slice(),
.send_request_to_connections(
Some(self.connection_ids.as_slice()),
payload,
Some(self.thread_id),
)
.await
}
@@ -99,6 +101,20 @@ impl ThreadScopedOutgoingMessageSender {
.await;
}
pub(crate) async fn abort_pending_server_requests(&self) {
self.outgoing
.cancel_requests_for_thread(
self.thread_id,
Some(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: "client request resolved because the turn state was changed"
.to_string(),
data: Some(serde_json::json!({ "reason": TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON })),
}),
)
.await
}
pub(crate) async fn send_response<T: Serialize>(
&self,
request_id: ConnectionRequestId,
@@ -129,38 +145,23 @@ impl OutgoingMessageSender {
&self,
request: ServerRequestPayload,
) -> (RequestId, oneshot::Receiver<ClientRequestResult>) {
self.send_request_with_id_to_connections(&[], request, None)
.await
}
async fn send_request_to_thread_connections(
&self,
thread_id: ThreadId,
connection_ids: &[ConnectionId],
request: ServerRequestPayload,
) -> oneshot::Receiver<ClientRequestResult> {
if connection_ids.is_empty() {
let (_tx, rx) = oneshot::channel();
return rx;
}
let (_request_id, receiver) = self
.send_request_with_id_to_connections(connection_ids, request, Some(thread_id))
.await;
receiver
self.send_request_to_connections(None, request, None).await
}
fn next_request_id(&self) -> RequestId {
RequestId::Integer(self.next_server_request_id.fetch_add(1, Ordering::Relaxed))
}
async fn send_request_with_id_to_connections(
async fn send_request_to_connections(
&self,
connection_ids: &[ConnectionId],
connection_ids: Option<&[ConnectionId]>,
request: ServerRequestPayload,
thread_id: Option<ThreadId>,
) -> (RequestId, oneshot::Receiver<ClientRequestResult>) {
let id = self.next_request_id();
let outgoing_message_id = id.clone();
let request = request.request_with_id(outgoing_message_id.clone());
let (tx_approve, rx_approve) = oneshot::channel();
{
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
@@ -169,36 +170,39 @@ impl OutgoingMessageSender {
PendingCallbackEntry {
callback: tx_approve,
thread_id,
request: request.clone(),
},
);
}
let outgoing_message =
OutgoingMessage::Request(request.request_with_id(outgoing_message_id.clone()));
let send_result = if connection_ids.is_empty() {
self.sender
.send(OutgoingEnvelope::Broadcast {
message: outgoing_message,
})
.await
} else {
let mut send_error = None;
for connection_id in connection_ids {
if let Err(err) = self
.sender
.send(OutgoingEnvelope::ToConnection {
connection_id: *connection_id,
message: outgoing_message.clone(),
let outgoing_message = OutgoingMessage::Request(request);
let send_result = match connection_ids {
None => {
self.sender
.send(OutgoingEnvelope::Broadcast {
message: outgoing_message,
})
.await
{
send_error = Some(err);
break;
}
}
match send_error {
Some(err) => Err(err),
None => Ok(()),
Some(connection_ids) => {
let mut send_error = None;
for connection_id in connection_ids {
if let Err(err) = self
.sender
.send(OutgoingEnvelope::ToConnection {
connection_id: *connection_id,
message: outgoing_message.clone(),
})
.await
{
send_error = Some(err);
break;
}
}
match send_error {
Some(err) => Err(err),
None => Ok(()),
}
}
};
@@ -210,11 +214,28 @@ impl OutgoingMessageSender {
(outgoing_message_id, rx_approve)
}
pub(crate) async fn replay_requests_to_connection_for_thread(
&self,
connection_id: ConnectionId,
thread_id: ThreadId,
) {
let requests = self.pending_requests_for_thread(thread_id).await;
for request in requests {
if let Err(err) = self
.sender
.send(OutgoingEnvelope::ToConnection {
connection_id,
message: OutgoingMessage::Request(request),
})
.await
{
warn!("failed to resend request to client: {err:?}");
}
}
}
pub(crate) async fn notify_client_response(&self, id: RequestId, result: Result) {
let entry = {
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
request_id_to_callback.remove_entry(&id)
};
let entry = self.take_request_callback(&id).await;
match entry {
Some((id, entry)) => {
@@ -229,10 +250,7 @@ impl OutgoingMessageSender {
}
pub(crate) async fn notify_client_error(&self, id: RequestId, error: JSONRPCErrorError) {
let entry = {
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
request_id_to_callback.remove_entry(&id)
};
let entry = self.take_request_callback(&id).await;
match entry {
Some((id, entry)) => {
@@ -248,23 +266,62 @@ impl OutgoingMessageSender {
}
pub(crate) async fn cancel_request(&self, id: &RequestId) -> bool {
let entry = {
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
request_id_to_callback.remove_entry(id)
};
entry.is_some()
self.take_request_callback(id).await.is_some()
}
pub(crate) async fn cancel_requests_for_thread(&self, thread_id: ThreadId) {
async fn take_request_callback(
&self,
id: &RequestId,
) -> Option<(RequestId, PendingCallbackEntry)> {
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
let request_ids = request_id_to_callback
request_id_to_callback.remove_entry(id)
}
pub(crate) async fn pending_requests_for_thread(
&self,
thread_id: ThreadId,
) -> Vec<ServerRequest> {
let request_id_to_callback = self.request_id_to_callback.lock().await;
let mut requests = request_id_to_callback
.iter()
.filter_map(|(request_id, entry)| {
(entry.thread_id == Some(thread_id)).then_some(request_id.clone())
.filter_map(|(_, entry)| {
(entry.thread_id == Some(thread_id)).then_some(entry.request.clone())
})
.collect::<Vec<_>>();
for request_id in request_ids {
request_id_to_callback.remove(&request_id);
requests.sort_by(|left, right| left.id().cmp(right.id()));
requests
}
pub(crate) async fn cancel_requests_for_thread(
&self,
thread_id: ThreadId,
error: Option<JSONRPCErrorError>,
) {
let entries = {
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
let request_ids = request_id_to_callback
.iter()
.filter_map(|(request_id, entry)| {
(entry.thread_id == Some(thread_id)).then_some(request_id.clone())
})
.collect::<Vec<_>>();
let mut entries = Vec::with_capacity(request_ids.len());
for request_id in request_ids {
if let Some(entry) = request_id_to_callback.remove(&request_id) {
entries.push(entry);
}
}
entries
};
if let Some(error) = error {
for entry in entries {
if let Err(err) = entry.callback.send(Err(error.clone())) {
let request_id = entry.request.id();
warn!("could not notify callback for {request_id:?} due to: {err:?}",);
}
}
}
}
@@ -441,14 +498,18 @@ mod tests {
use codex_app_server_protocol::ApplyPatchApprovalParams;
use codex_app_server_protocol::AuthMode;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::DynamicToolCallParams;
use codex_app_server_protocol::FileChangeRequestApprovalParams;
use codex_app_server_protocol::LoginChatGptCompleteNotification;
use codex_app_server_protocol::ModelRerouteReason;
use codex_app_server_protocol::ModelReroutedNotification;
use codex_app_server_protocol::RateLimitSnapshot;
use codex_app_server_protocol::RateLimitWindow;
use codex_app_server_protocol::ToolRequestUserInputParams;
use codex_protocol::ThreadId;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::sync::Arc;
use tokio::time::timeout;
use uuid::Uuid;
@@ -723,4 +784,121 @@ mod tests {
.expect("waiter should receive a callback");
assert_eq!(result, Err(error));
}
#[tokio::test]
async fn pending_requests_for_thread_returns_thread_requests_in_request_id_order() {
let (tx, _rx) = mpsc::channel::<OutgoingEnvelope>(8);
let outgoing = Arc::new(OutgoingMessageSender::new(tx));
let thread_id = ThreadId::new();
let thread_outgoing = ThreadScopedOutgoingMessageSender::new(
outgoing.clone(),
vec![ConnectionId(1)],
thread_id,
);
let (dynamic_tool_request_id, _dynamic_tool_waiter) = thread_outgoing
.send_request(ServerRequestPayload::DynamicToolCall(
DynamicToolCallParams {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
call_id: "call-0".to_string(),
tool: "tool".to_string(),
arguments: json!({}),
},
))
.await;
let (first_request_id, _first_waiter) = thread_outgoing
.send_request(ServerRequestPayload::ToolRequestUserInput(
ToolRequestUserInputParams {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
item_id: "call-1".to_string(),
questions: vec![],
},
))
.await;
let (second_request_id, _second_waiter) = thread_outgoing
.send_request(ServerRequestPayload::FileChangeRequestApproval(
FileChangeRequestApprovalParams {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
item_id: "call-2".to_string(),
reason: None,
grant_root: None,
},
))
.await;
let pending_requests = outgoing.pending_requests_for_thread(thread_id).await;
assert_eq!(
pending_requests
.iter()
.map(ServerRequest::id)
.collect::<Vec<_>>(),
vec![
&dynamic_tool_request_id,
&first_request_id,
&second_request_id
]
);
}
#[tokio::test]
async fn cancel_requests_for_thread_cancels_all_thread_requests() {
let (tx, _rx) = mpsc::channel::<OutgoingEnvelope>(8);
let outgoing = Arc::new(OutgoingMessageSender::new(tx));
let thread_id = ThreadId::new();
let thread_outgoing = ThreadScopedOutgoingMessageSender::new(
outgoing.clone(),
vec![ConnectionId(1)],
thread_id,
);
let (_dynamic_tool_request_id, dynamic_tool_waiter) = thread_outgoing
.send_request(ServerRequestPayload::DynamicToolCall(
DynamicToolCallParams {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
call_id: "call-0".to_string(),
tool: "tool".to_string(),
arguments: json!({}),
},
))
.await;
let (_request_id, user_input_waiter) = thread_outgoing
.send_request(ServerRequestPayload::ToolRequestUserInput(
ToolRequestUserInputParams {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
item_id: "call-1".to_string(),
questions: vec![],
},
))
.await;
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: "tracked request cancelled".to_string(),
data: None,
};
outgoing
.cancel_requests_for_thread(thread_id, Some(error.clone()))
.await;
let dynamic_tool_result = timeout(Duration::from_secs(1), dynamic_tool_waiter)
.await
.expect("dynamic tool waiter should resolve")
.expect("dynamic tool waiter should receive a callback");
let user_input_result = timeout(Duration::from_secs(1), user_input_waiter)
.await
.expect("user input waiter should resolve")
.expect("user input waiter should receive a callback");
assert_eq!(dynamic_tool_result, Err(error.clone()));
assert_eq!(user_input_result, Err(error));
assert!(
outgoing
.pending_requests_for_thread(thread_id)
.await
.is_empty()
);
}
}
codex-rs/app-server/src/server_request_error.rs
+42
View File
@@ -0,0 +1,42 @@
use codex_app_server_protocol::JSONRPCErrorError;
pub(crate) const TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON: &str = "turnTransition";
pub(crate) fn is_turn_transition_server_request_error(error: &JSONRPCErrorError) -> bool {
error
.data
.as_ref()
.and_then(|data| data.get("reason"))
.and_then(serde_json::Value::as_str)
== Some(TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON)
}
#[cfg(test)]
mod tests {
use super::is_turn_transition_server_request_error;
use codex_app_server_protocol::JSONRPCErrorError;
use pretty_assertions::assert_eq;
use serde_json::json;
#[test]
fn turn_transition_error_is_detected() {
let error = JSONRPCErrorError {
code: -1,
message: "client request resolved because the turn state was changed".to_string(),
data: Some(json!({ "reason": "turnTransition" })),
};
assert_eq!(is_turn_transition_server_request_error(&error), true);
}
#[test]
fn unrelated_error_is_not_detected() {
let error = JSONRPCErrorError {
code: -1,
message: "boom".to_string(),
data: Some(json!({ "reason": "other" })),
};
assert_eq!(is_turn_transition_server_request_error(&error), false);
}
}
codex-rs/app-server/src/thread_state.rs
+10 -1
View File
@@ -1,5 +1,6 @@
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::ConnectionRequestId;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadHistoryBuilder;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnError;
@@ -28,8 +29,16 @@ pub(crate) struct PendingThreadResumeRequest {
pub(crate) config_snapshot: ThreadConfigSnapshot,
}
// ThreadListenerCommand is used to perform operations in the context of the thread listener, for serialization purposes.
pub(crate) enum ThreadListenerCommand {
SendThreadResumeResponse(PendingThreadResumeRequest),
// SendThreadResumeResponse is used to resume an already running thread by sending the thread's history to the client and atomically subscribing for new updates.
SendThreadResumeResponse(Box<PendingThreadResumeRequest>),
// ResolveServerRequest is used to notify the client that the request has been resolved.
// It is executed in the thread listener's context to ensure that the resolved notification is ordered with regard to the request itself.
ResolveServerRequest {
request_id: RequestId,
completion_tx: oneshot::Sender<()>,
},
}
/// Per-conversation accumulation of the latest states e.g. error message while a turn runs.