69d7a456bb
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.
75 lines
2.5 KiB
Rust
75 lines
2.5 KiB
Rust
use codex_app_server_protocol::DynamicToolCallOutputContentItem;
|
|
use codex_app_server_protocol::DynamicToolCallResponse;
|
|
use codex_core::CodexThread;
|
|
use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem as CoreDynamicToolCallOutputContentItem;
|
|
use codex_protocol::dynamic_tools::DynamicToolResponse as CoreDynamicToolResponse;
|
|
use codex_protocol::protocol::Op;
|
|
use std::sync::Arc;
|
|
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,
|
|
receiver: oneshot::Receiver<ClientRequestResult>,
|
|
conversation: Arc<CodexThread>,
|
|
) {
|
|
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")
|
|
}
|
|
Err(err) => {
|
|
error!("request failed: {err:?}");
|
|
fallback_response("dynamic tool request failed")
|
|
}
|
|
};
|
|
|
|
let DynamicToolCallResponse {
|
|
content_items,
|
|
success,
|
|
} = response.clone();
|
|
let core_response = CoreDynamicToolResponse {
|
|
content_items: content_items
|
|
.into_iter()
|
|
.map(CoreDynamicToolCallOutputContentItem::from)
|
|
.collect(),
|
|
success,
|
|
};
|
|
if let Err(err) = conversation
|
|
.submit(Op::DynamicToolResponse {
|
|
id: call_id.clone(),
|
|
response: core_response,
|
|
})
|
|
.await
|
|
{
|
|
error!("failed to submit DynamicToolResponse: {err}");
|
|
}
|
|
}
|
|
|
|
fn decode_response(value: serde_json::Value) -> (DynamicToolCallResponse, Option<String>) {
|
|
match serde_json::from_value::<DynamicToolCallResponse>(value) {
|
|
Ok(response) => (response, None),
|
|
Err(err) => {
|
|
error!("failed to deserialize DynamicToolCallResponse: {err}");
|
|
fallback_response("dynamic tool response was invalid")
|
|
}
|
|
}
|
|
}
|
|
|
|
fn fallback_response(message: &str) -> (DynamicToolCallResponse, Option<String>) {
|
|
(
|
|
DynamicToolCallResponse {
|
|
content_items: vec![DynamicToolCallOutputContentItem::InputText {
|
|
text: message.to_string(),
|
|
}],
|
|
success: false,
|
|
},
|
|
Some(message.to_string()),
|
|
)
|
|
}
|