mirror of
https://github.com/openai/codex.git
synced 2026-10-21 09:54:56 +00:00
Avoid cloning deferred TUI lifecycle payloads (#34347)
## What changed - Move lifecycle and approval payloads into either the interrupt queue or the immediate handler instead of cloning them for both branches. - Borrow MCP elicitation request metadata while constructing the corresponding form, and consume apply-patch changes after collecting notification paths. ## Testing - Add a history replay test confirming deferred MCP start and completion events remain FIFO after streaming finishes. GitOrigin-RevId: 84ed34f6ff173463a1da016dc140405392c0ba45
This commit is contained in:
@@ -279,7 +279,7 @@ impl App {
|
||||
McpServerElicitationFormRequest::from_app_server_request(
|
||||
thread_id,
|
||||
request_id.clone(),
|
||||
params.clone(),
|
||||
params,
|
||||
)
|
||||
{
|
||||
Some(ThreadInteractiveRequest::McpServerElicitation(request))
|
||||
|
||||
@@ -207,7 +207,7 @@ impl McpServerElicitationFormRequest {
|
||||
pub(crate) fn from_app_server_request(
|
||||
thread_id: ThreadId,
|
||||
request_id: AppServerRequestId,
|
||||
request: McpServerElicitationRequestParams,
|
||||
request: &McpServerElicitationRequestParams,
|
||||
) -> Option<Self> {
|
||||
let McpServerElicitationRequestParams {
|
||||
server_name,
|
||||
@@ -226,10 +226,10 @@ impl McpServerElicitationFormRequest {
|
||||
let requested_schema = serde_json::to_value(requested_schema).ok()?;
|
||||
Self::from_parts(
|
||||
thread_id,
|
||||
server_name,
|
||||
server_name.clone(),
|
||||
request_id,
|
||||
meta,
|
||||
message,
|
||||
meta.as_ref(),
|
||||
message.clone(),
|
||||
requested_schema,
|
||||
)
|
||||
}
|
||||
@@ -238,13 +238,12 @@ impl McpServerElicitationFormRequest {
|
||||
thread_id: ThreadId,
|
||||
server_name: String,
|
||||
request_id: AppServerRequestId,
|
||||
meta: Option<Value>,
|
||||
meta: Option<&Value>,
|
||||
message: String,
|
||||
requested_schema: Value,
|
||||
) -> Option<Self> {
|
||||
let tool_suggestion = parse_tool_suggestion_request(meta.as_ref());
|
||||
let tool_suggestion = parse_tool_suggestion_request(meta);
|
||||
let is_tool_approval = meta
|
||||
.as_ref()
|
||||
.and_then(Value::as_object)
|
||||
.and_then(|meta| meta.get(APPROVAL_META_KIND_KEY))
|
||||
.and_then(Value::as_str)
|
||||
@@ -259,7 +258,7 @@ impl McpServerElicitationFormRequest {
|
||||
let is_message_only_schema = requested_schema.is_null() || is_empty_object_schema;
|
||||
let is_tool_approval_action = is_tool_approval && is_message_only_schema;
|
||||
let approval_display_params = if is_tool_approval_action {
|
||||
parse_tool_approval_display_params(meta.as_ref())
|
||||
parse_tool_approval_display_params(meta)
|
||||
} else {
|
||||
Vec::new()
|
||||
};
|
||||
@@ -277,7 +276,7 @@ impl McpServerElicitationFormRequest {
|
||||
description: Some(allow_description.to_string()),
|
||||
value: Value::String(APPROVAL_ACCEPT_ONCE_VALUE.to_string()),
|
||||
}];
|
||||
if approval_supports_persist_mode(meta.as_ref(), APPROVAL_PERSIST_SESSION_VALUE) {
|
||||
if approval_supports_persist_mode(meta, APPROVAL_PERSIST_SESSION_VALUE) {
|
||||
let description = if is_tool_approval_action {
|
||||
"Run the tool and remember this choice for this session."
|
||||
} else {
|
||||
@@ -289,7 +288,7 @@ impl McpServerElicitationFormRequest {
|
||||
value: Value::String(APPROVAL_ACCEPT_SESSION_VALUE.to_string()),
|
||||
});
|
||||
}
|
||||
if approval_supports_persist_mode(meta.as_ref(), APPROVAL_PERSIST_ALWAYS_VALUE) {
|
||||
if approval_supports_persist_mode(meta, APPROVAL_PERSIST_ALWAYS_VALUE) {
|
||||
let description = if is_tool_approval_action {
|
||||
"Run the tool and remember this choice for future tool calls."
|
||||
} else {
|
||||
@@ -1777,7 +1776,7 @@ mod tests {
|
||||
McpServerElicitationFormRequest::from_app_server_request(
|
||||
thread_id,
|
||||
request_id("request-1"),
|
||||
request,
|
||||
&request,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -2496,7 +2495,7 @@ mod tests {
|
||||
McpServerElicitationFormRequest::from_app_server_request(
|
||||
thread_id,
|
||||
request_id("request-2"),
|
||||
McpServerElicitationRequestParams {
|
||||
&McpServerElicitationRequestParams {
|
||||
thread_id: "thread-1".to_string(),
|
||||
turn_id: Some("turn-2".to_string()),
|
||||
server_name: "server-1".to_string(),
|
||||
|
||||
@@ -44,10 +44,10 @@ impl ChatWidget {
|
||||
return;
|
||||
}
|
||||
}
|
||||
let item2 = item.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_item_started(item),
|
||||
|s| s.handle_command_execution_started_now(item2),
|
||||
item,
|
||||
InterruptManager::push_item_started,
|
||||
Self::handle_command_execution_started_now,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -155,10 +155,10 @@ impl ChatWidget {
|
||||
return;
|
||||
}
|
||||
}
|
||||
let item2 = item.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_item_completed(item),
|
||||
|s| s.handle_command_execution_completed_now(item2),
|
||||
item,
|
||||
InterruptManager::push_item_completed,
|
||||
Self::handle_command_execution_completed_now,
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -396,19 +396,21 @@ impl ChatWidget {
|
||||
self.interrupts = mgr;
|
||||
}
|
||||
|
||||
/// Move a lifecycle payload into the interrupt queue or its immediate handler.
|
||||
#[inline]
|
||||
pub(super) fn defer_or_handle(
|
||||
pub(super) fn defer_or_handle<T>(
|
||||
&mut self,
|
||||
push: impl FnOnce(&mut InterruptManager),
|
||||
handle: impl FnOnce(&mut Self),
|
||||
payload: T,
|
||||
push: impl FnOnce(&mut InterruptManager, T),
|
||||
handle: impl FnOnce(&mut Self, T),
|
||||
) {
|
||||
// Preserve deterministic FIFO across queued interrupts: once anything
|
||||
// is queued due to an active write cycle, continue queueing until the
|
||||
// queue is flushed to avoid reordering (e.g., ExecEnd before ExecBegin).
|
||||
if self.stream_controller.is_some() || !self.interrupts.is_empty() {
|
||||
push(&mut self.interrupts);
|
||||
push(&mut self.interrupts, payload);
|
||||
} else {
|
||||
handle(self);
|
||||
handle(self, payload);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1135,6 +1135,64 @@ async fn replayed_in_progress_mcp_tool_call_stays_active() {
|
||||
assert!(!active.contains("MCP tool call completed without a result"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn deferred_mcp_lifecycle_events_keep_fifo_after_stream_finishes() {
|
||||
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
|
||||
let cwd = chat.config.cwd.to_path_buf();
|
||||
chat.stream_controller = Some(crate::streaming::controller::StreamController::new(
|
||||
/*width*/ Some(80),
|
||||
cwd.as_path(),
|
||||
chat.history_render_mode(),
|
||||
));
|
||||
|
||||
chat.on_mcp_tool_call_started(AppServerThreadItem::McpToolCall {
|
||||
id: "mcp-deferred".to_string(),
|
||||
server: "copilot-bridge".to_string(),
|
||||
tool: "copilot".to_string(),
|
||||
status: codex_app_server_protocol::McpToolCallStatus::InProgress,
|
||||
arguments: json!({"action": "wait"}),
|
||||
app_context: None,
|
||||
mcp_app_resource_uri: None,
|
||||
plugin_id: None,
|
||||
result: None,
|
||||
error: None,
|
||||
duration_ms: None,
|
||||
});
|
||||
assert!(!chat.interrupts.is_empty());
|
||||
|
||||
chat.stream_controller = None;
|
||||
chat.on_mcp_tool_call_completed(AppServerThreadItem::McpToolCall {
|
||||
id: "mcp-deferred".to_string(),
|
||||
server: "copilot-bridge".to_string(),
|
||||
tool: "copilot".to_string(),
|
||||
status: codex_app_server_protocol::McpToolCallStatus::Completed,
|
||||
arguments: json!({"action": "wait"}),
|
||||
app_context: None,
|
||||
mcp_app_resource_uri: None,
|
||||
plugin_id: None,
|
||||
result: Some(Box::new(codex_app_server_protocol::McpToolCallResult {
|
||||
content: vec![json!({"type": "text", "text": "deferred result"})],
|
||||
structured_content: None,
|
||||
meta: None,
|
||||
})),
|
||||
error: None,
|
||||
duration_ms: Some(5),
|
||||
});
|
||||
|
||||
assert!(!chat.interrupts.is_empty());
|
||||
assert!(drain_insert_history(&mut rx).is_empty());
|
||||
|
||||
chat.flush_interrupt_queue();
|
||||
|
||||
assert!(chat.interrupts.is_empty());
|
||||
assert!(chat.transcript.active_cell.is_none());
|
||||
let rendered = drain_insert_history(&mut rx)
|
||||
.into_iter()
|
||||
.map(|lines| lines_to_single_string(&lines))
|
||||
.collect::<String>();
|
||||
assert!(rendered.contains("deferred result"), "{rendered}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn live_reasoning_summary_is_not_rendered_twice_when_item_completes() {
|
||||
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
|
||||
|
||||
@@ -45,26 +45,26 @@ impl ChatWidget {
|
||||
}
|
||||
|
||||
pub(super) fn on_file_change_completed(&mut self, item: ThreadItem) {
|
||||
let item2 = item.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_item_completed(item),
|
||||
|s| s.handle_file_change_completed_now(item2),
|
||||
item,
|
||||
InterruptManager::push_item_completed,
|
||||
Self::handle_file_change_completed_now,
|
||||
);
|
||||
}
|
||||
|
||||
pub(super) fn on_mcp_tool_call_started(&mut self, item: ThreadItem) {
|
||||
let item2 = item.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_item_started(item),
|
||||
|s| s.handle_mcp_tool_call_started_now(item2),
|
||||
item,
|
||||
InterruptManager::push_item_started,
|
||||
Self::handle_mcp_tool_call_started_now,
|
||||
);
|
||||
}
|
||||
|
||||
pub(super) fn on_mcp_tool_call_completed(&mut self, item: ThreadItem) {
|
||||
let item2 = item.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_item_completed(item),
|
||||
|s| s.handle_mcp_tool_call_completed_now(item2),
|
||||
item,
|
||||
InterruptManager::push_item_completed,
|
||||
Self::handle_mcp_tool_call_completed_now,
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -7,10 +7,10 @@ use super::*;
|
||||
|
||||
impl ChatWidget {
|
||||
pub(super) fn on_exec_approval_request(&mut self, _id: String, ev: ExecApprovalRequestEvent) {
|
||||
let ev2 = ev.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_exec_approval(ev),
|
||||
|s| s.handle_exec_approval_now(ev2),
|
||||
ev,
|
||||
InterruptManager::push_exec_approval,
|
||||
Self::handle_exec_approval_now,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -19,10 +19,10 @@ impl ChatWidget {
|
||||
_id: String,
|
||||
ev: ApplyPatchApprovalRequestEvent,
|
||||
) {
|
||||
let ev2 = ev.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_apply_patch_approval(ev),
|
||||
|s| s.handle_apply_patch_approval_now(ev2),
|
||||
ev,
|
||||
InterruptManager::push_apply_patch_approval,
|
||||
Self::handle_apply_patch_approval_now,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -256,27 +256,26 @@ impl ChatWidget {
|
||||
request_id: AppServerRequestId,
|
||||
params: McpServerElicitationRequestParams,
|
||||
) {
|
||||
let request_id2 = request_id.clone();
|
||||
let params2 = params.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_elicitation(request_id, params),
|
||||
|s| s.handle_elicitation_request_now(request_id2, params2),
|
||||
(request_id, params),
|
||||
|q, (request_id, params)| q.push_elicitation(request_id, params),
|
||||
|s, (request_id, params)| s.handle_elicitation_request_now(request_id, params),
|
||||
);
|
||||
}
|
||||
|
||||
pub(super) fn on_request_user_input(&mut self, ev: ToolRequestUserInputParams) {
|
||||
let ev2 = ev.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_user_input(ev),
|
||||
|s| s.handle_request_user_input_now(ev2),
|
||||
ev,
|
||||
InterruptManager::push_user_input,
|
||||
Self::handle_request_user_input_now,
|
||||
);
|
||||
}
|
||||
|
||||
pub(super) fn on_request_permissions(&mut self, ev: RequestPermissionsEvent) {
|
||||
let ev2 = ev.clone();
|
||||
self.defer_or_handle(
|
||||
|q| q.push_request_permissions(ev),
|
||||
|s| s.handle_request_permissions_now(ev2),
|
||||
ev,
|
||||
InterruptManager::push_request_permissions,
|
||||
Self::handle_request_permissions_now,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -310,12 +309,13 @@ impl ChatWidget {
|
||||
pub(crate) fn handle_apply_patch_approval_now(&mut self, ev: ApplyPatchApprovalRequestEvent) {
|
||||
self.flush_answer_stream_with_separator();
|
||||
|
||||
let changed_paths = ev.changes.keys().cloned().collect();
|
||||
let request = ApprovalRequest::ApplyPatch(ApplyPatchApprovalRequest {
|
||||
thread_id: self.thread_id.unwrap_or_default(),
|
||||
thread_label: None,
|
||||
id: ev.call_id,
|
||||
reason: ev.reason,
|
||||
changes: ev.changes.clone(),
|
||||
changes: ev.changes,
|
||||
cwd: self.config.cwd.clone(),
|
||||
});
|
||||
self.bottom_pane
|
||||
@@ -327,7 +327,7 @@ impl ChatWidget {
|
||||
self.request_redraw();
|
||||
self.notify(Notification::EditApprovalRequested {
|
||||
cwd: self.config.cwd.to_path_buf(),
|
||||
changes: ev.changes.keys().cloned().collect(),
|
||||
changes: changed_paths,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -354,7 +354,7 @@ impl ChatWidget {
|
||||
} else if let Some(request) = McpServerElicitationFormRequest::from_app_server_request(
|
||||
thread_id,
|
||||
request_id.clone(),
|
||||
params.clone(),
|
||||
¶ms,
|
||||
) {
|
||||
self.bottom_pane
|
||||
.push_mcp_server_elicitation_request(request);
|
||||
|
||||
Reference in New Issue
Block a user