diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index f943e38771..6d666cb30f 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -37,6 +37,7 @@ use crate::pager_overlay::Overlay; use crate::render::highlight::highlight_bash_to_lines; use crate::render::renderable::Renderable; use crate::resume_picker::SessionSelection; +use crate::thread_update::ThreadUpdate; use crate::tui; use crate::tui::TuiEvent; use crate::update_action::UpdateAction; @@ -65,7 +66,6 @@ use codex_protocol::config_types::CollaborationModeMask; use codex_protocol::config_types::Personality; #[cfg(target_os = "windows")] use codex_protocol::config_types::WindowsSandboxLevel; -use codex_protocol::items::TurnItem; use codex_protocol::openai_models::ModelAvailabilityNux; use codex_protocol::openai_models::ModelPreset; use codex_protocol::openai_models::ModelUpgrade; @@ -93,7 +93,6 @@ use ratatui::widgets::Paragraph; use ratatui::widgets::Wrap; use std::collections::BTreeMap; use std::collections::HashMap; -use std::collections::HashSet; use std::collections::VecDeque; use std::path::Path; use std::path::PathBuf; @@ -283,16 +282,15 @@ struct SessionSummary { #[derive(Debug, Clone)] struct ThreadEventSnapshot { - session_configured: Option, - events: Vec, + session_configured: Option, + updates: Vec, input_state: Option, } #[derive(Debug)] struct ThreadEventStore { - session_configured: Option, - buffer: VecDeque, - user_message_ids: HashSet, + session_configured: Option, + buffer: VecDeque, pending_interactive_replay: PendingInteractiveReplayState, input_state: Option, capacity: usize, @@ -304,7 +302,6 @@ impl ThreadEventStore { Self { session_configured: None, buffer: VecDeque::new(), - user_message_ids: HashSet::new(), pending_interactive_replay: PendingInteractiveReplayState::default(), input_state: None, capacity, @@ -312,53 +309,24 @@ impl ThreadEventStore { } } - fn new_with_session_configured(capacity: usize, event: Event) -> Self { + fn new_with_session_configured(capacity: usize, event: SessionConfiguredEvent) -> Self { let mut store = Self::new(capacity); store.session_configured = Some(event); store } - fn push_event(&mut self, event: Event) { - self.pending_interactive_replay.note_event(&event); - match &event.msg { - EventMsg::SessionConfigured(_) => { - self.session_configured = Some(event); - return; - } - EventMsg::ItemCompleted(completed) => { - if let TurnItem::UserMessage(item) = &completed.item { - if !event.id.is_empty() && self.user_message_ids.contains(&event.id) { - return; - } - let legacy = Event { - id: event.id, - msg: item.as_legacy_event(), - }; - self.push_legacy_event(legacy); - return; - } - } - _ => {} - } - - self.push_legacy_event(event); - } - - fn push_legacy_event(&mut self, event: Event) { - if let EventMsg::UserMessage(_) = &event.msg - && !event.id.is_empty() - && !self.user_message_ids.insert(event.id.clone()) - { + fn push_update(&mut self, update: ThreadUpdate) { + self.pending_interactive_replay.note_update(&update); + if let ThreadUpdate::SessionConfigured(session) = &update { + self.session_configured = Some(session.clone()); return; } - self.buffer.push_back(event); + self.buffer.push_back(update); if self.buffer.len() > self.capacity && let Some(removed) = self.buffer.pop_front() { - self.pending_interactive_replay.note_evicted_event(&removed); - if matches!(removed.msg, EventMsg::UserMessage(_)) && !removed.id.is_empty() { - self.user_message_ids.remove(&removed.id); - } + self.pending_interactive_replay + .note_evicted_update(&removed); } } @@ -367,12 +335,12 @@ impl ThreadEventStore { session_configured: self.session_configured.clone(), // Thread switches replay buffered events into a rebuilt ChatWidget. Only replay // interactive prompts that are still pending, or answered approvals/input will reappear. - events: self + updates: self .buffer .iter() - .filter(|event| { + .filter(|update| { self.pending_interactive_replay - .should_replay_snapshot_event(event) + .should_replay_snapshot_update(update) }) .cloned() .collect(), @@ -388,8 +356,8 @@ impl ThreadEventStore { PendingInteractiveReplayState::op_can_change_state(op) } - fn event_can_change_pending_thread_approvals(event: &Event) -> bool { - PendingInteractiveReplayState::event_can_change_pending_thread_approvals(event) + fn update_can_change_pending_thread_approvals(update: &ThreadUpdate) -> bool { + PendingInteractiveReplayState::update_can_change_pending_thread_approvals(update) } fn has_pending_thread_approvals(&self) -> bool { @@ -400,8 +368,8 @@ impl ThreadEventStore { #[derive(Debug)] struct ThreadEventChannel { - sender: mpsc::Sender, - receiver: Option>, + sender: mpsc::Sender, + receiver: Option>, store: Arc>, } @@ -415,7 +383,7 @@ impl ThreadEventChannel { } } - fn new_with_session_configured(capacity: usize, event: Event) -> Self { + fn new_with_session_configured(capacity: usize, event: SessionConfiguredEvent) -> Self { let (sender, receiver) = mpsc::channel(capacity); Self { sender, @@ -723,16 +691,15 @@ pub(crate) struct App { thread_event_listener_tasks: HashMap>, agent_navigation: AgentNavigationState, active_thread_id: Option, - active_thread_rx: Option>, + active_thread_rx: Option>, primary_thread_id: Option, primary_session_configured: Option, - pending_primary_events: VecDeque, + pending_primary_updates: VecDeque, next_app_server_request_id: i64, active_turn_ids: HashMap, pending_exec_approval_request_ids: HashMap, pending_patch_approval_request_ids: HashMap, pending_elicitation_request_ids: HashMap<(String, String), RequestId>, - pending_elicitation_request_queues: HashMap>, pending_permissions_request_ids: HashMap, pending_user_input_request_ids: HashMap, pending_dynamic_tool_request_ids: HashMap, @@ -1295,7 +1262,7 @@ impl App { async fn activate_thread_for_replay( &mut self, thread_id: ThreadId, - ) -> Option<(mpsc::Receiver, ThreadEventSnapshot)> { + ) -> Option<(mpsc::Receiver, ThreadEventSnapshot)> { let channel = self.thread_event_channels.get_mut(&thread_id)?; let receiver = channel.receiver.take()?; let mut store = channel.store.lock().await; @@ -1383,70 +1350,110 @@ impl App { async fn thread_cwd(&self, thread_id: ThreadId) -> Option { let channel = self.thread_event_channels.get(&thread_id)?; let store = channel.store.lock().await; - match store.session_configured.as_ref().map(|event| &event.msg) { - Some(EventMsg::SessionConfigured(session)) => Some(session.cwd.clone()), - _ => None, + store + .session_configured + .as_ref() + .map(|session| session.cwd.clone()) + } + + fn app_server_request_id_to_mcp_request_id( + request_id: &codex_app_server_protocol::RequestId, + ) -> codex_protocol::mcp::RequestId { + match request_id { + codex_app_server_protocol::RequestId::String(value) => { + codex_protocol::mcp::RequestId::String(value.clone()) + } + codex_app_server_protocol::RequestId::Integer(value) => { + codex_protocol::mcp::RequestId::Integer(*value) + } } } - async fn interactive_request_for_thread_event( + async fn interactive_request_for_thread_update( &self, thread_id: ThreadId, - event: &Event, + update: &ThreadUpdate, ) -> Option { let thread_label = Some(self.thread_label(thread_id)); - match &event.msg { - EventMsg::ExecApprovalRequest(ev) => { + match update { + ThreadUpdate::CommandExecutionRequestApproval { params, .. } => { Some(ThreadInteractiveRequest::Approval(ApprovalRequest::Exec { thread_id, thread_label, - id: ev.effective_approval_id(), - command: ev.command.clone(), - reason: ev.reason.clone(), - available_decisions: ev.effective_available_decisions(), - network_approval_context: ev.network_approval_context.clone(), - additional_permissions: ev.additional_permissions.clone(), + id: params + .approval_id + .clone() + .unwrap_or_else(|| params.item_id.clone()), + command: params + .command + .as_ref() + .map(|command| vec![command.clone()]) + .unwrap_or_default(), + reason: params.reason.clone(), + available_decisions: Vec::new(), + network_approval_context: params.network_approval_context.clone().and_then( + |context| serde_json::from_value(serde_json::to_value(context).ok()?).ok(), + ), + additional_permissions: params.additional_permissions.clone().and_then( + |permissions| { + serde_json::from_value(serde_json::to_value(permissions).ok()?).ok() + }, + ), })) } - EventMsg::ApplyPatchApprovalRequest(ev) => Some(ThreadInteractiveRequest::Approval( - ApprovalRequest::ApplyPatch { + ThreadUpdate::FileChangeRequestApproval { params, .. } => Some( + ThreadInteractiveRequest::Approval(ApprovalRequest::ApplyPatch { thread_id, thread_label, - id: ev.call_id.clone(), - reason: ev.reason.clone(), + id: params.item_id.clone(), + reason: params.reason.clone(), cwd: self .thread_cwd(thread_id) .await .unwrap_or_else(|| self.config.cwd.clone()), - changes: ev.changes.clone(), - }, - )), - EventMsg::ElicitationRequest(ev) => { - if let Some(request) = - McpServerElicitationFormRequest::from_event(thread_id, ev.clone()) - { + changes: HashMap::new(), + }), + ), + ThreadUpdate::McpServerElicitationRequest { request_id, params } => { + if let Some(request) = McpServerElicitationFormRequest::from_app_server_request( + thread_id, + request_id.clone(), + params.clone(), + ) { Some(ThreadInteractiveRequest::McpServerElicitation(request)) } else { Some(ThreadInteractiveRequest::Approval( ApprovalRequest::McpElicitation { thread_id, thread_label, - server_name: ev.server_name.clone(), - request_id: ev.id.clone(), - message: ev.request.message().to_string(), + server_name: params.server_name.clone(), + request_id: Self::app_server_request_id_to_mcp_request_id(request_id), + message: match ¶ms.request { + codex_app_server_protocol::McpServerElicitationRequest::Form { + message, + .. + } + | codex_app_server_protocol::McpServerElicitationRequest::Url { + message, + .. + } => message.clone(), + }, }, )) } } - EventMsg::RequestPermissions(ev) => Some(ThreadInteractiveRequest::Approval( - ApprovalRequest::Permissions { + ThreadUpdate::PermissionsRequestApproval { params, .. } => Some( + ThreadInteractiveRequest::Approval(ApprovalRequest::Permissions { thread_id, thread_label, - call_id: ev.call_id.clone(), - reason: ev.reason.clone(), - permissions: ev.permissions.clone(), - }, - )), + call_id: params.item_id.clone(), + reason: params.reason.clone(), + permissions: serde_json::from_value( + serde_json::to_value(params.permissions.clone()).ok()?, + ) + .unwrap_or_default(), + }), + ), _ => None, } } @@ -1496,11 +1503,15 @@ impl App { self.chat_widget.set_pending_thread_approvals(threads); } - async fn enqueue_thread_event(&mut self, thread_id: ThreadId, event: Event) -> Result<()> { + async fn enqueue_thread_update( + &mut self, + thread_id: ThreadId, + update: ThreadUpdate, + ) -> Result<()> { let refresh_pending_thread_approvals = - ThreadEventStore::event_can_change_pending_thread_approvals(&event); + ThreadEventStore::update_can_change_pending_thread_approvals(&update); let inactive_interactive_request = if self.active_thread_id != Some(thread_id) { - self.interactive_request_for_thread_event(thread_id, &event) + self.interactive_request_for_thread_update(thread_id, &update) .await } else { None @@ -1512,7 +1523,7 @@ impl App { let should_send = { let mut guard = store.lock().await; - guard.push_event(event.clone()); + guard.push_update(update.clone()); guard.active }; @@ -1520,11 +1531,11 @@ impl App { // Never await a bounded channel send on the main TUI loop: if the receiver falls behind, // `send().await` can block and the UI stops drawing. If the channel is full, wait in a // spawned task instead. - match sender.try_send(event) { + match sender.try_send(update) { Ok(()) => {} - Err(TrySendError::Full(event)) => { + Err(TrySendError::Full(update)) => { tokio::spawn(async move { - if let Err(err) = sender.send(event).await { + if let Err(err) = sender.send(update).await { tracing::warn!("thread {thread_id} event channel closed: {err}"); } }); @@ -1550,13 +1561,13 @@ impl App { Ok(()) } - async fn handle_routed_thread_event( + async fn handle_routed_thread_update( &mut self, thread_id: ThreadId, - event: Event, + update: ThreadUpdate, ) -> Result<()> { if !self.thread_event_channels.contains_key(&thread_id) { - let should_attach_channel = matches!(event.msg, EventMsg::SessionConfigured(_)) + let should_attach_channel = matches!(update, ThreadUpdate::SessionConfigured(_)) || self.agent_navigation.get(&thread_id).is_some(); if !should_attach_channel { tracing::debug!("dropping routed event for unknown thread {thread_id}"); @@ -1565,29 +1576,30 @@ impl App { self.ensure_thread_channel(thread_id); } - self.enqueue_thread_event(thread_id, event).await + self.enqueue_thread_update(thread_id, update).await } - async fn enqueue_primary_event(&mut self, event: Event) -> Result<()> { + async fn enqueue_primary_update(&mut self, update: ThreadUpdate) -> Result<()> { if let Some(thread_id) = self.primary_thread_id { - return self.enqueue_thread_event(thread_id, event).await; + return self.enqueue_thread_update(thread_id, update).await; } - if let EventMsg::SessionConfigured(session) = &event.msg { + if let ThreadUpdate::SessionConfigured(session) = &update { let thread_id = session.session_id; self.primary_thread_id = Some(thread_id); self.primary_session_configured = Some(session.clone()); self.upsert_agent_picker_thread(thread_id, None, None, false); self.ensure_thread_channel(thread_id); self.activate_thread_channel(thread_id).await; - self.enqueue_thread_event(thread_id, event).await?; + self.enqueue_thread_update(thread_id, update).await?; - let pending = std::mem::take(&mut self.pending_primary_events); - for pending_event in pending { - self.enqueue_thread_event(thread_id, pending_event).await?; + let pending = std::mem::take(&mut self.pending_primary_updates); + for pending_update in pending { + self.enqueue_thread_update(thread_id, pending_update) + .await?; } } else { - self.pending_primary_events.push_back(event); + self.pending_primary_updates.push_back(update); } Ok(()) } @@ -1757,7 +1769,7 @@ impl App { self.active_thread_id = None; self.active_thread_rx = None; self.primary_thread_id = None; - self.pending_primary_events.clear(); + self.pending_primary_updates.clear(); self.chat_widget.set_pending_thread_approvals(Vec::new()); self.sync_active_agent_label(); } @@ -1820,10 +1832,7 @@ impl App { ); self.reset_thread_event_state(); if let Err(err) = self - .enqueue_primary_event(Event { - id: String::new(), - msg: EventMsg::SessionConfigured(session_configured), - }) + .enqueue_primary_update(ThreadUpdate::SessionConfigured(session_configured)) .await { self.chat_widget @@ -1854,7 +1863,7 @@ impl App { let mut disconnected = false; loop { match rx.try_recv() { - Ok(event) => self.handle_codex_event_now(event), + Ok(event) => self.handle_thread_update_now(event), Err(TryRecvError::Empty) => break, Err(TryRecvError::Disconnected) => { disconnected = true; @@ -1898,19 +1907,37 @@ impl App { (active_thread_id != primary_thread_id).then_some((active_thread_id, primary_thread_id)) } + fn active_non_primary_closed_thread_target( + &self, + update: &ThreadUpdate, + ) -> Option<(ThreadId, ThreadId)> { + let ThreadUpdate::ThreadClosed(notification) = update else { + return None; + }; + let active_thread_id = self.active_thread_id?; + let primary_thread_id = self.primary_thread_id?; + if self.pending_shutdown_exit_thread_id == Some(active_thread_id) { + return None; + } + if notification.thread_id != active_thread_id.to_string() { + return None; + } + (active_thread_id != primary_thread_id).then_some((active_thread_id, primary_thread_id)) + } + fn replay_thread_snapshot( &mut self, snapshot: ThreadEventSnapshot, resume_restored_queue: bool, ) { - if let Some(event) = snapshot.session_configured { - self.handle_codex_event_replay(event); + if let Some(session) = snapshot.session_configured { + self.handle_thread_update_replay(ThreadUpdate::SessionConfigured(session)); } self.chat_widget.set_queue_autosend_suppressed(true); self.chat_widget .restore_thread_input_state(snapshot.input_state); - for event in snapshot.events { - self.handle_codex_event_replay(event); + for update in snapshot.updates { + self.handle_thread_update_replay(update); } self.chat_widget.set_queue_autosend_suppressed(false); if resume_restored_queue { @@ -2119,13 +2146,12 @@ impl App { active_thread_rx: None, primary_thread_id: None, primary_session_configured: None, - pending_primary_events: VecDeque::new(), + pending_primary_updates: VecDeque::new(), next_app_server_request_id: 1, active_turn_ids: HashMap::new(), pending_exec_approval_request_ids: HashMap::new(), pending_patch_approval_request_ids: HashMap::new(), pending_elicitation_request_ids: HashMap::new(), - pending_elicitation_request_queues: HashMap::new(), pending_permissions_request_ids: HashMap::new(), pending_user_input_request_ids: HashMap::new(), pending_dynamic_tool_request_ids: HashMap::new(), @@ -2175,11 +2201,8 @@ impl App { } }; if let Some(session_configured) = initial_session_configured { - app.enqueue_primary_event(Event { - id: String::new(), - msg: EventMsg::SessionConfigured(session_configured), - }) - .await?; + app.enqueue_primary_update(ThreadUpdate::SessionConfigured(session_configured)) + .await?; } // On startup, if Agent mode (workspace-write) or ReadOnly is active, warn about world-writable dirs on Windows. @@ -2259,7 +2282,7 @@ impl App { app.active_thread_rx.is_some() ) => { if let Some(event) = active { - if let Err(err) = app.handle_active_thread_event(tui, event).await { + if let Err(err) = app.handle_active_thread_update(tui, event).await { break Err(err); } } else { @@ -2463,10 +2486,9 @@ impl App { ), ); self.reset_thread_event_state(); - self.enqueue_primary_event(Event { - id: String::new(), - msg: EventMsg::SessionConfigured(session_configured), - }) + self.enqueue_primary_update(ThreadUpdate::SessionConfigured( + session_configured, + )) .await?; if let Some(summary) = summary { let mut lines: Vec> = @@ -2534,10 +2556,9 @@ impl App { ), ); self.reset_thread_event_state(); - self.enqueue_primary_event(Event { - id: String::new(), - msg: EventMsg::SessionConfigured(session_configured), - }) + self.enqueue_primary_update(ThreadUpdate::SessionConfigured( + session_configured, + )) .await?; if let Some(summary) = summary { let mut lines: Vec> = @@ -2628,10 +2649,12 @@ impl App { self.chat_widget.on_commit_tick(); } AppEvent::CodexEvent(event) => { - self.enqueue_primary_event(event).await?; + self.handle_codex_event_now(event); } AppEvent::ThreadEvent { thread_id, event } => { - self.handle_routed_thread_event(thread_id, event).await?; + if Some(thread_id) == self.active_thread_id { + self.handle_codex_event_now(event); + } } AppEvent::Exit(mode) => { return Ok(self.handle_exit_mode(app_server, mode).await); @@ -3852,15 +3875,42 @@ impl App { self.chat_widget.handle_codex_event_replay(event); } + fn handle_thread_update_now(&mut self, update: ThreadUpdate) { + if self.suppress_shutdown_complete && update.is_thread_closed() { + self.suppress_shutdown_complete = false; + return; + } + if let ThreadUpdate::ThreadRolledBack { num_turns } = update { + self.handle_backtrack_event(&EventMsg::ThreadRolledBack( + codex_protocol::protocol::ThreadRolledBackEvent { num_turns }, + )); + self.chat_widget + .handle_thread_update(ThreadUpdate::ThreadRolledBack { num_turns }); + return; + } + self.chat_widget.handle_thread_update(update.clone()); + if update.is_status_refresh_update() { + self.refresh_status_line(); + } + } + + fn handle_thread_update_replay(&mut self, update: ThreadUpdate) { + self.chat_widget.handle_thread_update_replay(update); + } + /// Handles an event emitted by the currently active thread. /// /// This function enforces shutdown intent routing: unexpected non-primary /// thread shutdowns fail over to the primary thread, while user-requested /// app exits consume only the tracked shutdown completion and then proceed. - async fn handle_active_thread_event(&mut self, tui: &mut tui::Tui, event: Event) -> Result<()> { + async fn handle_active_thread_update( + &mut self, + tui: &mut tui::Tui, + update: ThreadUpdate, + ) -> Result<()> { // Capture this before any potential thread switch: we only want to clear // the exit marker when the currently active thread acknowledges shutdown. - let pending_shutdown_exit_completed = matches!(&event.msg, EventMsg::ShutdownComplete) + let pending_shutdown_exit_completed = matches!(&update, ThreadUpdate::ThreadClosed(_)) && self.pending_shutdown_exit_thread_id == self.active_thread_id; // Processing order matters: @@ -3872,7 +3922,7 @@ impl App { // This preserves the mental model that user-requested exits do not trigger // failover, while true sub-agent deaths still do. if let Some((closed_thread_id, primary_thread_id)) = - self.active_non_primary_shutdown_target(&event.msg) + self.active_non_primary_closed_thread_target(&update) { self.mark_agent_picker_thread_closed(closed_thread_id); self.select_agent_thread(tui, primary_thread_id).await?; @@ -3897,7 +3947,7 @@ impl App { // thread, so unrelated shutdowns cannot consume this marker. self.pending_shutdown_exit_thread_id = None; } - self.handle_codex_event_now(event); + self.handle_thread_update_now(update); if self.backtrack_render_pending { tui.frame_requester().schedule_frame(); } @@ -4241,8 +4291,6 @@ mod tests { use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadRolledBackEvent; - use codex_protocol::protocol::TurnAbortReason; - use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; @@ -4258,6 +4306,92 @@ mod tests { use tempfile::tempdir; use tokio::time; + fn test_session_configured( + thread_id: ThreadId, + cwd: impl Into, + ) -> SessionConfiguredEvent { + SessionConfiguredEvent { + session_id: thread_id, + forked_from_id: None, + thread_name: None, + model: "gpt-test".to_string(), + model_provider_id: "test-provider".to_string(), + service_tier: None, + approval_policy: AskForApproval::Never, + approvals_reviewer: ApprovalsReviewer::User, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + cwd: cwd.into(), + reasoning_effort: None, + history_log_id: 0, + history_entry_count: 0, + initial_messages: None, + network_proxy: None, + rollout_path: Some(PathBuf::new()), + } + } + + fn test_exec_approval_update( + thread_id: ThreadId, + cwd: impl Into, + call_id: &str, + turn_id: &str, + reason: Option<&str>, + ) -> ThreadUpdate { + ThreadUpdate::CommandExecutionRequestApproval { + request_id: RequestId::String(format!("request-{call_id}")), + params: codex_app_server_protocol::CommandExecutionRequestApprovalParams { + thread_id: thread_id.to_string(), + turn_id: turn_id.to_string(), + item_id: call_id.to_string(), + approval_id: None, + reason: reason.map(str::to_string), + network_approval_context: None, + command: Some("echo hi".to_string()), + cwd: Some(cwd.into()), + command_actions: None, + additional_permissions: None, + skill_metadata: None, + proposed_execpolicy_amendment: None, + proposed_network_policy_amendments: None, + available_decisions: None, + }, + } + } + + fn test_thread_closed_update(thread_id: ThreadId) -> ThreadUpdate { + ThreadUpdate::ThreadClosed(codex_app_server_protocol::ThreadClosedNotification { + thread_id: thread_id.to_string(), + }) + } + + fn test_turn_started_update(thread_id: ThreadId, turn_id: &str) -> ThreadUpdate { + ThreadUpdate::TurnStarted(codex_app_server_protocol::TurnStartedNotification { + thread_id: thread_id.to_string(), + turn: codex_app_server_protocol::Turn { + id: turn_id.to_string(), + items: Vec::new(), + status: codex_app_server_protocol::TurnStatus::InProgress, + error: None, + }, + }) + } + + fn test_turn_completed_update( + thread_id: ThreadId, + turn_id: &str, + status: codex_app_server_protocol::TurnStatus, + ) -> ThreadUpdate { + ThreadUpdate::TurnCompleted(codex_app_server_protocol::TurnCompletedNotification { + thread_id: thread_id.to_string(), + turn: codex_app_server_protocol::Turn { + id: turn_id.to_string(), + items: Vec::new(), + status, + error: None, + }, + }) + } + async fn start_test_app_server(config: Config) -> InProcessAppServerClient { InProcessAppServerClient::start(codex_app_server_client::InProcessClientStartArgs { arg0_paths: Arg0DispatchPaths::default(), @@ -4384,50 +4518,18 @@ mod tests { -> Result<()> { let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await; let thread_id = ThreadId::new(); - let approval_event = Event { - id: "approval-event".to_string(), - msg: EventMsg::ExecApprovalRequest( - codex_protocol::protocol::ExecApprovalRequestEvent { - call_id: "call-1".to_string(), - approval_id: None, - turn_id: "turn-1".to_string(), - command: vec!["echo".to_string(), "hello".to_string()], - cwd: PathBuf::from("/tmp/project"), - reason: Some("needs approval".to_string()), - network_approval_context: None, - proposed_execpolicy_amendment: None, - proposed_network_policy_amendments: None, - additional_permissions: None, - skill_metadata: None, - available_decisions: None, - parsed_cmd: Vec::new(), - }, - ), - }; - let session_configured_event = Event { - id: "session-configured".to_string(), - msg: EventMsg::SessionConfigured(SessionConfiguredEvent { - session_id: thread_id, - forked_from_id: None, - thread_name: None, - model: "gpt-test".to_string(), - model_provider_id: "test-provider".to_string(), - service_tier: None, - approval_policy: AskForApproval::Never, - approvals_reviewer: ApprovalsReviewer::User, - sandbox_policy: SandboxPolicy::new_read_only_policy(), - cwd: PathBuf::from("/tmp/project"), - reasoning_effort: None, - history_log_id: 0, - history_entry_count: 0, - initial_messages: None, - network_proxy: None, - rollout_path: Some(PathBuf::new()), - }), - }; + let approval_event = test_exec_approval_update( + thread_id, + "/tmp/project", + "call-1", + "turn-1", + Some("needs approval"), + ); + let session_configured_event = + ThreadUpdate::SessionConfigured(test_session_configured(thread_id, "/tmp/project")); - app.enqueue_primary_event(approval_event.clone()).await?; - app.enqueue_primary_event(session_configured_event.clone()) + app.enqueue_primary_update(approval_event.clone()).await?; + app.enqueue_primary_update(session_configured_event.clone()) .await?; let rx = app @@ -4443,11 +4545,14 @@ mod tests { .expect("timed out waiting for buffered approval event") .expect("channel closed unexpectedly"); - assert!(matches!(first_event.msg, EventMsg::SessionConfigured(_))); - assert!(matches!(second_event.msg, EventMsg::ExecApprovalRequest(_))); + assert!(matches!(first_event, ThreadUpdate::SessionConfigured(_))); + assert!(matches!( + second_event, + ThreadUpdate::CommandExecutionRequestApproval { .. } + )); - app.handle_codex_event_now(first_event); - app.handle_codex_event_now(second_event); + app.handle_thread_update_now(first_event); + app.handle_thread_update_now(second_event); app.chat_widget .handle_key_event(KeyEvent::new(KeyCode::Char('y'), KeyModifiers::NONE)); @@ -4475,14 +4580,8 @@ mod tests { ); app.reset_thread_event_state(); - app.handle_routed_thread_event( - thread_id, - Event { - id: "stale-event".to_string(), - msg: EventMsg::ShutdownComplete, - }, - ) - .await?; + app.handle_routed_thread_update(thread_id, test_thread_closed_update(thread_id)) + .await?; assert!( !app.thread_event_channels.contains_key(&thread_id), @@ -4536,18 +4635,15 @@ mod tests { .insert(thread_id, ThreadEventChannel::new(1)); app.set_thread_active(thread_id, true).await; - let event = Event { - id: String::new(), - msg: EventMsg::ShutdownComplete, - }; + let event = test_thread_closed_update(thread_id); - app.enqueue_thread_event(thread_id, event.clone()).await?; + app.enqueue_thread_update(thread_id, event.clone()).await?; time::timeout( Duration::from_millis(50), - app.enqueue_thread_event(thread_id, event), + app.enqueue_thread_update(thread_id, event), ) .await - .expect("enqueue_thread_event blocked on a full channel")?; + .expect("enqueue_thread_update blocked on a full channel")?; let mut rx = app .thread_event_channels @@ -4577,27 +4673,7 @@ mod tests { thread_id, ThreadEventChannel::new_with_session_configured( THREAD_EVENT_CHANNEL_CAPACITY, - Event { - id: "session-configured".to_string(), - msg: EventMsg::SessionConfigured(SessionConfiguredEvent { - session_id: thread_id, - forked_from_id: None, - thread_name: None, - model: "gpt-test".to_string(), - model_provider_id: "test-provider".to_string(), - service_tier: None, - approval_policy: AskForApproval::Never, - approvals_reviewer: ApprovalsReviewer::User, - sandbox_policy: SandboxPolicy::new_read_only_policy(), - cwd: PathBuf::from("/tmp/project"), - reasoning_effort: None, - history_log_id: 0, - history_entry_count: 0, - initial_messages: None, - network_proxy: None, - rollout_path: Some(PathBuf::new()), - }), - }, + test_session_configured(thread_id, "/tmp/project"), ), ); app.activate_thread_channel(thread_id).await; @@ -4709,13 +4785,11 @@ mod tests { app.replay_thread_snapshot( ThreadEventSnapshot { session_configured: None, - events: vec![Event { - id: "turn-complete".to_string(), - msg: EventMsg::TurnComplete(TurnCompleteEvent { - turn_id: "turn-1".to_string(), - last_agent_message: None, - }), - }], + updates: vec![test_turn_completed_update( + thread_id, + "turn-1", + codex_app_server_protocol::TurnStatus::Completed, + )], input_state: Some(input_state), }, true, @@ -4792,13 +4866,11 @@ mod tests { app.replay_thread_snapshot( ThreadEventSnapshot { session_configured: None, - events: vec![Event { - id: "turn-complete".to_string(), - msg: EventMsg::TurnComplete(TurnCompleteEvent { - turn_id: "turn-1".to_string(), - last_agent_message: None, - }), - }], + updates: vec![test_turn_completed_update( + thread_id, + "turn-1", + codex_app_server_protocol::TurnStatus::Completed, + )], input_state: Some(input_state), }, false, @@ -4873,7 +4945,7 @@ mod tests { app.replay_thread_snapshot( ThreadEventSnapshot { session_configured: None, - events: vec![], + updates: vec![], input_state: Some(input_state), }, true, @@ -4948,22 +5020,13 @@ mod tests { app.replay_thread_snapshot( ThreadEventSnapshot { session_configured: None, - events: vec![ - Event { - id: "older-turn-complete".to_string(), - msg: EventMsg::TurnComplete(TurnCompleteEvent { - turn_id: "turn-0".to_string(), - last_agent_message: None, - }), - }, - Event { - id: "latest-turn-started".to_string(), - msg: EventMsg::TurnStarted(TurnStartedEvent { - turn_id: "turn-1".to_string(), - model_context_window: None, - collaboration_mode_kind: Default::default(), - }), - }, + updates: vec![ + test_turn_completed_update( + thread_id, + "turn-0", + codex_app_server_protocol::TurnStatus::Completed, + ), + test_turn_started_update(thread_id, "turn-1"), ], input_state: Some(input_state), }, @@ -5007,27 +5070,7 @@ mod tests { thread_id, ThreadEventChannel::new_with_session_configured( THREAD_EVENT_CHANNEL_CAPACITY, - Event { - id: "session-configured".to_string(), - msg: EventMsg::SessionConfigured(SessionConfiguredEvent { - session_id: thread_id, - forked_from_id: None, - thread_name: None, - model: "gpt-test".to_string(), - model_provider_id: "test-provider".to_string(), - service_tier: None, - approval_policy: AskForApproval::Never, - approvals_reviewer: ApprovalsReviewer::User, - sandbox_policy: SandboxPolicy::new_read_only_policy(), - cwd: PathBuf::from("/tmp/project"), - reasoning_effort: None, - history_log_id: 0, - history_entry_count: 0, - initial_messages: None, - network_proxy: None, - rollout_path: Some(PathBuf::new()), - }), - }, + test_session_configured(thread_id, "/tmp/project"), ), ); app.activate_thread_channel(thread_id).await; @@ -5136,7 +5179,7 @@ mod tests { app.replay_thread_snapshot( ThreadEventSnapshot { session_configured: None, - events: vec![], + updates: vec![], input_state: Some(input_state), }, true, @@ -5237,7 +5280,7 @@ mod tests { app.replay_thread_snapshot( ThreadEventSnapshot { session_configured: None, - events: vec![], + updates: vec![], input_state: Some(input_state), }, true, @@ -5313,13 +5356,11 @@ mod tests { app.replay_thread_snapshot( ThreadEventSnapshot { session_configured: None, - events: vec![Event { - id: "turn-aborted".to_string(), - msg: EventMsg::TurnAborted(TurnAbortedEvent { - turn_id: Some("turn-1".to_string()), - reason: TurnAbortReason::ReviewEnded, - }), - }], + updates: vec![test_turn_completed_update( + thread_id, + "turn-1", + codex_app_server_protocol::TurnStatus::Interrupted, + )], input_state: Some(input_state), }, true, @@ -5999,26 +6040,13 @@ smart_approvals = true let agent_channel = ThreadEventChannel::new(1); { let mut store = agent_channel.store.lock().await; - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::ExecApprovalRequest( - codex_protocol::protocol::ExecApprovalRequestEvent { - call_id: "call-1".to_string(), - approval_id: None, - turn_id: "turn-1".to_string(), - command: vec!["echo".to_string(), "hi".to_string()], - cwd: PathBuf::from("/tmp"), - reason: None, - network_approval_context: None, - proposed_execpolicy_amendment: None, - proposed_network_policy_amendments: None, - additional_permissions: None, - skill_metadata: None, - available_decisions: None, - parsed_cmd: Vec::new(), - }, - ), - }); + store.push_update(test_exec_approval_update( + agent_thread_id, + "/tmp", + "call-1", + "turn-1", + None, + )); } app.thread_event_channels .insert(agent_thread_id, agent_channel); @@ -6056,26 +6084,11 @@ smart_approvals = true agent_thread_id, ThreadEventChannel::new_with_session_configured( 1, - Event { - id: String::new(), - msg: EventMsg::SessionConfigured(SessionConfiguredEvent { - session_id: agent_thread_id, - forked_from_id: None, - thread_name: None, - model: "gpt-5".to_string(), - model_provider_id: "test-provider".to_string(), - service_tier: None, - approval_policy: AskForApproval::OnRequest, - approvals_reviewer: ApprovalsReviewer::User, - sandbox_policy: SandboxPolicy::new_workspace_write_policy(), - cwd: PathBuf::from("/tmp/agent"), - reasoning_effort: None, - history_log_id: 0, - history_entry_count: 0, - initial_messages: None, - network_proxy: None, - rollout_path: Some(PathBuf::from("/tmp/agent-rollout.jsonl")), - }), + SessionConfiguredEvent { + approval_policy: AskForApproval::OnRequest, + sandbox_policy: SandboxPolicy::new_workspace_write_policy(), + rollout_path: Some(PathBuf::from("/tmp/agent-rollout.jsonl")), + ..test_session_configured(agent_thread_id, "/tmp/agent") }, ), ); @@ -6086,28 +6099,15 @@ smart_approvals = true false, ); - app.enqueue_thread_event( + app.enqueue_thread_update( agent_thread_id, - Event { - id: "ev-approval".to_string(), - msg: EventMsg::ExecApprovalRequest( - codex_protocol::protocol::ExecApprovalRequestEvent { - call_id: "call-approval".to_string(), - approval_id: None, - turn_id: "turn-approval".to_string(), - command: vec!["echo".to_string(), "hi".to_string()], - cwd: PathBuf::from("/tmp/agent"), - reason: Some("need approval".to_string()), - network_approval_context: None, - proposed_execpolicy_amendment: None, - proposed_network_policy_amendments: None, - additional_permissions: None, - skill_metadata: None, - available_decisions: None, - parsed_cmd: Vec::new(), - }, - ), - }, + test_exec_approval_update( + agent_thread_id, + "/tmp/agent", + "call-approval", + "turn-approval", + Some("need approval"), + ), ) .await?; @@ -6439,13 +6439,12 @@ smart_approvals = true active_thread_rx: None, primary_thread_id: None, primary_session_configured: None, - pending_primary_events: VecDeque::new(), + pending_primary_updates: VecDeque::new(), next_app_server_request_id: 1, active_turn_ids: HashMap::new(), pending_exec_approval_request_ids: HashMap::new(), pending_patch_approval_request_ids: HashMap::new(), pending_elicitation_request_ids: HashMap::new(), - pending_elicitation_request_queues: HashMap::new(), pending_permissions_request_ids: HashMap::new(), pending_user_input_request_ids: HashMap::new(), pending_dynamic_tool_request_ids: HashMap::new(), @@ -6512,13 +6511,12 @@ smart_approvals = true active_thread_rx: None, primary_thread_id: None, primary_session_configured: None, - pending_primary_events: VecDeque::new(), + pending_primary_updates: VecDeque::new(), next_app_server_request_id: 1, active_turn_ids: HashMap::new(), pending_exec_approval_request_ids: HashMap::new(), pending_patch_approval_request_ids: HashMap::new(), pending_elicitation_request_ids: HashMap::new(), - pending_elicitation_request_queues: HashMap::new(), pending_permissions_request_ids: HashMap::new(), pending_user_input_request_ids: HashMap::new(), pending_dynamic_tool_request_ids: HashMap::new(), diff --git a/codex-rs/tui/src/app/app_server_adapter.rs b/codex-rs/tui/src/app/app_server_adapter.rs index a2706687e0..b77ca4f1ec 100644 --- a/codex-rs/tui/src/app/app_server_adapter.rs +++ b/codex-rs/tui/src/app/app_server_adapter.rs @@ -18,7 +18,6 @@ use codex_app_server_protocol::GetAccountParams; use codex_app_server_protocol::GetAccountRateLimitsResponse; use codex_app_server_protocol::GetAccountResponse; use codex_app_server_protocol::JSONRPCErrorError; -use codex_app_server_protocol::JSONRPCNotification; use codex_app_server_protocol::McpServerElicitationAction; use codex_app_server_protocol::McpServerElicitationRequestResponse; use codex_app_server_protocol::McpServerRefreshResponse; @@ -85,6 +84,8 @@ use codex_protocol::request_permissions::PermissionGrantScope; use serde::de::DeserializeOwned; use serde_json::Value; +use crate::thread_update::ThreadUpdate; + impl App { pub(super) async fn list_models_via_app_server( app_server_client: &InProcessAppServerClient, @@ -628,12 +629,10 @@ impl App { .map_err(|err| format!("failed to encode skills/list response: {err}"))?, ) .map_err(|err| format!("failed to convert skills/list response: {err}"))?; - self.enqueue_primary_event(Event { + self.handle_codex_event_now(Event { id: String::new(), msg: EventMsg::ListSkillsResponse(ListSkillsResponseEvent { skills }), - }) - .await - .map_err(|err| err.to_string())?; + }); } Op::RefreshMcpServers { config } => { let _: McpServerRefreshResponse = send_request_with_response( @@ -691,6 +690,8 @@ impl App { "thread/rollback", ) .await?; + self.route_thread_update(ThreadUpdate::ThreadRolledBack { num_turns }) + .await; } Op::Review { review_request } => { let response: ReviewStartResponse = send_request_with_response( @@ -758,10 +759,10 @@ impl App { .insert(params.item_id.clone(), request_id.clone()); } ServerRequest::McpServerElicitationRequest { request_id, params } => { - self.pending_elicitation_request_queues - .entry(params.server_name.clone()) - .or_default() - .push_back(request_id.clone()); + self.pending_elicitation_request_ids.insert( + (params.server_name.clone(), request_id.to_string()), + request_id.clone(), + ); } ServerRequest::PermissionsRequestApproval { request_id, params } => { self.pending_permissions_request_ids @@ -784,6 +785,24 @@ impl App { } } + async fn route_thread_update(&mut self, update: ThreadUpdate) { + let Some(thread_id_text) = update.thread_id() else { + self.handle_thread_update_now(update); + return; + }; + let Ok(thread_id) = ThreadId::from_string(&thread_id_text) else { + tracing::warn!("failed to parse thread id for thread update"); + return; + }; + if self.primary_thread_id.is_none() || self.primary_thread_id == Some(thread_id) { + if let Err(err) = self.enqueue_primary_update(update).await { + tracing::warn!("{err}"); + } + } else if let Err(err) = self.handle_routed_thread_update(thread_id, update).await { + tracing::warn!("{err}"); + } + } + async fn handle_server_notification( &mut self, app_server_client: &InProcessAppServerClient, @@ -791,9 +810,12 @@ impl App { ) { match notification { ServerNotification::TurnStarted(payload) => { + let active_turn_id = payload.turn.id.clone(); if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) { - self.active_turn_ids.insert(thread_id, payload.turn.id); + self.active_turn_ids.insert(thread_id, active_turn_id); } + self.route_thread_update(ThreadUpdate::TurnStarted(payload)) + .await; } ServerNotification::TurnCompleted(payload) => { if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) @@ -801,34 +823,146 @@ impl App { { self.active_turn_ids.remove(&thread_id); } + self.route_thread_update(ThreadUpdate::TurnCompleted(payload)) + .await; } ServerNotification::ThreadStarted(payload) => { + let agent_nickname = payload.thread.agent_nickname.clone(); + let agent_role = payload.thread.agent_role.clone(); if let Ok(thread_id) = ThreadId::from_string(&payload.thread.id) { - self.upsert_agent_picker_thread( - thread_id, - payload.thread.agent_nickname, - payload.thread.agent_role, - false, - ); + self.upsert_agent_picker_thread(thread_id, agent_nickname, agent_role, false); } + self.route_thread_update(ThreadUpdate::ThreadStarted(payload)) + .await; + } + ServerNotification::ThreadStatusChanged(payload) => { + self.route_thread_update(ThreadUpdate::ThreadStatusChanged(payload)) + .await; } ServerNotification::ThreadClosed(payload) => { if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) { self.mark_agent_picker_thread_closed(thread_id); self.active_turn_ids.remove(&thread_id); } + self.route_thread_update(ThreadUpdate::ThreadClosed(payload)) + .await; } ServerNotification::ThreadNameUpdated(payload) => { if let Ok(thread_id) = ThreadId::from_string(&payload.thread_id) && let Some(channel) = self.thread_event_channels.get(&thread_id) { let mut store = channel.store.lock().await; - if let Some(event) = store.session_configured.as_mut() - && let EventMsg::SessionConfigured(session) = &mut event.msg - { + if let Some(session) = store.session_configured.as_mut() { session.thread_name = payload.thread_name.clone(); } } + self.route_thread_update(ThreadUpdate::ThreadNameUpdated(payload)) + .await; + } + ServerNotification::ThreadTokenUsageUpdated(payload) => { + self.route_thread_update(ThreadUpdate::ThreadTokenUsageUpdated(payload)) + .await; + } + ServerNotification::TurnDiffUpdated(payload) => { + self.route_thread_update(ThreadUpdate::TurnDiffUpdated(payload)) + .await; + } + ServerNotification::TurnPlanUpdated(payload) => { + self.route_thread_update(ThreadUpdate::TurnPlanUpdated(payload)) + .await; + } + ServerNotification::ItemStarted(payload) => { + self.route_thread_update(ThreadUpdate::ItemStarted(payload)) + .await; + } + ServerNotification::ItemGuardianApprovalReviewStarted(payload) => { + self.route_thread_update(ThreadUpdate::ItemGuardianApprovalReviewStarted(payload)) + .await; + } + ServerNotification::ItemGuardianApprovalReviewCompleted(payload) => { + self.route_thread_update(ThreadUpdate::ItemGuardianApprovalReviewCompleted( + payload, + )) + .await; + } + ServerNotification::ItemCompleted(payload) => { + self.route_thread_update(ThreadUpdate::ItemCompleted(payload)) + .await; + } + ServerNotification::AgentMessageDelta(payload) => { + self.route_thread_update(ThreadUpdate::AgentMessageDelta(payload)) + .await; + } + ServerNotification::PlanDelta(payload) => { + self.route_thread_update(ThreadUpdate::PlanDelta(payload)) + .await; + } + ServerNotification::ReasoningSummaryTextDelta(payload) => { + self.route_thread_update(ThreadUpdate::ReasoningSummaryTextDelta(payload)) + .await; + } + ServerNotification::ReasoningSummaryPartAdded(payload) => { + self.route_thread_update(ThreadUpdate::ReasoningSummaryPartAdded(payload)) + .await; + } + ServerNotification::ReasoningTextDelta(payload) => { + self.route_thread_update(ThreadUpdate::ReasoningTextDelta(payload)) + .await; + } + ServerNotification::TerminalInteraction(payload) => { + self.route_thread_update(ThreadUpdate::TerminalInteraction(payload)) + .await; + } + ServerNotification::CommandExecutionOutputDelta(payload) => { + self.route_thread_update(ThreadUpdate::CommandExecutionOutputDelta(payload)) + .await; + } + ServerNotification::FileChangeOutputDelta(payload) => { + self.route_thread_update(ThreadUpdate::FileChangeOutputDelta(payload)) + .await; + } + ServerNotification::McpToolCallProgress(payload) => { + self.route_thread_update(ThreadUpdate::McpToolCallProgress(payload)) + .await; + } + ServerNotification::HookStarted(payload) => { + self.route_thread_update(ThreadUpdate::HookStarted(payload)) + .await; + } + ServerNotification::HookCompleted(payload) => { + self.route_thread_update(ThreadUpdate::HookCompleted(payload)) + .await; + } + ServerNotification::Error(payload) => { + self.route_thread_update(ThreadUpdate::Error(payload)).await; + } + ServerNotification::ModelRerouted(payload) => { + self.route_thread_update(ThreadUpdate::ModelRerouted(payload)) + .await; + } + ServerNotification::DeprecationNotice(payload) => { + self.route_thread_update(ThreadUpdate::DeprecationNotice(payload)) + .await; + } + ServerNotification::ThreadRealtimeStarted(payload) => { + self.route_thread_update(ThreadUpdate::ThreadRealtimeStarted(payload)) + .await; + } + ServerNotification::ThreadRealtimeItemAdded(payload) => { + self.route_thread_update(ThreadUpdate::ThreadRealtimeItemAdded(payload)) + .await; + } + ServerNotification::ThreadRealtimeOutputAudioDelta(payload) => { + self.route_thread_update(ThreadUpdate::ThreadRealtimeOutputAudioDelta(payload)) + .await; + } + ServerNotification::ThreadRealtimeError(payload) => { + self.route_thread_update(ThreadUpdate::ThreadRealtimeError(payload)) + .await; + } + ServerNotification::ThreadRealtimeClosed(payload) => { + self.route_thread_update(ThreadUpdate::ThreadRealtimeClosed(payload)) + .await; } ServerNotification::AccountUpdated(_) => { match Self::read_account_via_app_server(app_server_client).await { @@ -857,22 +991,24 @@ impl App { true, ); } + ServerNotification::SkillsChanged(_) => { + if let Some(thread_id) = self.primary_thread_id { + let _ = self + .submit_app_server_op_inner( + app_server_client, + thread_id, + Op::ListSkills { + cwds: Vec::new(), + force_reload: true, + }, + ) + .await; + } + } _ => {} } } - fn maybe_correlate_legacy_server_request(&mut self, event: &Event) { - if let EventMsg::ElicitationRequest(ev) = &event.msg - && let Some(queue) = self - .pending_elicitation_request_queues - .get_mut(&ev.server_name) - && let Some(request_id) = queue.pop_front() - { - self.pending_elicitation_request_ids - .insert((ev.server_name.clone(), ev.id.to_string()), request_id); - } - } - pub(super) async fn handle_app_server_event( &mut self, app_server_client: &InProcessAppServerClient, @@ -892,51 +1028,58 @@ impl App { .await; } InProcessServerEvent::LegacyNotification(notification) => { - match decode_legacy_notification(notification) { - Ok(decoded) => { - self.maybe_correlate_legacy_server_request(&decoded.event); - match decoded.conversation_id { - Some(conversation_id) => { - match ThreadId::from_string(&conversation_id) { - Ok(thread_id) => { - if self.primary_thread_id.is_none() - || self.primary_thread_id == Some(thread_id) - { - if let Err(err) = - self.enqueue_primary_event(decoded.event).await - { - tracing::warn!("{err}"); - } - } else if let Err(err) = self - .handle_routed_thread_event(thread_id, decoded.event) - .await - { - tracing::warn!("{err}"); - } - } - Err(err) => { - tracing::warn!( - conversation_id, - error = %err, - "failed to parse app-server conversation id" - ); - } - } - } - None => { - if let Err(err) = self.enqueue_primary_event(decoded.event).await { - tracing::warn!("{err}"); - } - } - } - } - Err(err) => { - tracing::warn!("{err}"); - } - } + tracing::warn!( + notification.method = %notification.method, + "unexpected legacy notification after TUI app-server typed-event migration" + ); } InProcessServerEvent::ServerRequest(request) => { self.note_server_request(&request); + match request.clone() { + ServerRequest::CommandExecutionRequestApproval { request_id, params } => { + self.route_thread_update(ThreadUpdate::CommandExecutionRequestApproval { + request_id, + params, + }) + .await; + } + ServerRequest::FileChangeRequestApproval { request_id, params } => { + self.route_thread_update(ThreadUpdate::FileChangeRequestApproval { + request_id, + params, + }) + .await; + } + ServerRequest::McpServerElicitationRequest { request_id, params } => { + self.route_thread_update(ThreadUpdate::McpServerElicitationRequest { + request_id, + params, + }) + .await; + } + ServerRequest::PermissionsRequestApproval { request_id, params } => { + self.route_thread_update(ThreadUpdate::PermissionsRequestApproval { + request_id, + params, + }) + .await; + } + ServerRequest::ToolRequestUserInput { request_id, params } => { + self.route_thread_update(ThreadUpdate::ToolRequestUserInput { + request_id, + params, + }) + .await; + } + ServerRequest::DynamicToolCall { request_id, params } => { + self.route_thread_update(ThreadUpdate::DynamicToolCall { + request_id, + params, + }) + .await; + } + _ => {} + } if let ServerRequest::ChatgptAuthTokensRefresh { request_id, params: _, @@ -1216,10 +1359,6 @@ fn elicitation_action_to_api(action: ElicitationAction) -> McpServerElicitationA } } -fn normalize_legacy_notification_method(method: &str) -> &str { - method.strip_prefix("codex/event/").unwrap_or(method) -} - fn lagged_event_warning_message(skipped: usize) -> String { format!("in-process app-server event stream lagged; dropped {skipped} events") } @@ -1303,56 +1442,6 @@ fn credits_snapshot_from_api( } } -struct DecodedLegacyNotification { - conversation_id: Option, - event: Event, -} - -fn decode_legacy_notification( - notification: JSONRPCNotification, -) -> Result { - let value = notification - .params - .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new())); - let method = notification.method; - let normalized_method = normalize_legacy_notification_method(&method).to_string(); - let serde_json::Value::Object(mut object) = value else { - return Err(format!( - "legacy notification `{method}` params were not an object" - )); - }; - let conversation_id = object - .get("conversationId") - .and_then(serde_json::Value::as_str) - .map(str::to_owned); - let mut event_payload = if let Some(serde_json::Value::Object(msg_payload)) = object.get("msg") - { - serde_json::Value::Object(msg_payload.clone()) - } else { - object.remove("conversationId"); - serde_json::Value::Object(object) - }; - let serde_json::Value::Object(ref mut object) = event_payload else { - return Err(format!( - "legacy notification `{method}` event payload was not an object" - )); - }; - object.insert( - "type".to_string(), - serde_json::Value::String(normalized_method), - ); - - let msg: EventMsg = serde_json::from_value(event_payload) - .map_err(|err| format!("failed to decode event: {err}"))?; - Ok(DecodedLegacyNotification { - conversation_id, - event: Event { - id: String::new(), - msg, - }, - }) -} - async fn resolve_server_request( client: &InProcessAppServerClient, request_id: RequestId, diff --git a/codex-rs/tui/src/app/pending_interactive_replay.rs b/codex-rs/tui/src/app/pending_interactive_replay.rs index 02efb2ec1b..bdeb133c3c 100644 --- a/codex-rs/tui/src/app/pending_interactive_replay.rs +++ b/codex-rs/tui/src/app/pending_interactive_replay.rs @@ -1,5 +1,4 @@ -use codex_protocol::protocol::Event; -use codex_protocol::protocol::EventMsg; +use crate::thread_update::ThreadUpdate; use codex_protocol::protocol::Op; use std::collections::HashMap; use std::collections::HashSet; @@ -7,11 +6,11 @@ use std::collections::HashSet; #[derive(Debug, Clone, PartialEq, Eq, Hash)] struct ElicitationRequestKey { server_name: String, - request_id: codex_protocol::mcp::RequestId, + request_id: String, } impl ElicitationRequestKey { - fn new(server_name: String, request_id: codex_protocol::mcp::RequestId) -> Self { + fn new(server_name: String, request_id: String) -> Self { Self { server_name, request_id, @@ -20,44 +19,30 @@ impl ElicitationRequestKey { } #[derive(Debug, Default)] -// Tracks which interactive prompts are still unresolved in the thread-event buffer. -// -// Thread snapshots are replayed when switching threads/agents. Most events should replay -// verbatim, but interactive prompts (approvals, request_user_input, MCP elicitations) must -// only replay if they are still pending. This state is updated from: -// - inbound events (`note_event`) -// - outbound ops that resolve a prompt (`note_outbound_op`) -// - buffer eviction (`note_evicted_event`) -// -// We keep both fast lookup sets (for snapshot filtering by call_id/request key) and -// turn-indexed queues/vectors so `TurnComplete`/`TurnAborted` can clear stale prompts tied -// to a turn. `request_user_input` removal is FIFO because the overlay answers queued prompts -// in FIFO order for a shared `turn_id`. pub(super) struct PendingInteractiveReplayState { - exec_approval_call_ids: HashSet, - exec_approval_call_ids_by_turn_id: HashMap>, - patch_approval_call_ids: HashSet, - patch_approval_call_ids_by_turn_id: HashMap>, + exec_approval_ids: HashSet, + exec_approval_ids_by_turn_id: HashMap>, + patch_approval_ids: HashSet, + patch_approval_ids_by_turn_id: HashMap>, elicitation_requests: HashSet, - request_permissions_call_ids: HashSet, - request_permissions_call_ids_by_turn_id: HashMap>, - request_user_input_call_ids: HashSet, - request_user_input_call_ids_by_turn_id: HashMap>, + request_permissions_ids: HashSet, + request_permissions_ids_by_turn_id: HashMap>, + request_user_input_ids: HashSet, + request_user_input_ids_by_turn_id: HashMap>, } impl PendingInteractiveReplayState { - pub(super) fn event_can_change_pending_thread_approvals(event: &Event) -> bool { + pub(super) fn update_can_change_pending_thread_approvals(update: &ThreadUpdate) -> bool { matches!( - &event.msg, - EventMsg::ExecApprovalRequest(_) - | EventMsg::ApplyPatchApprovalRequest(_) - | EventMsg::ElicitationRequest(_) - | EventMsg::RequestPermissions(_) - | EventMsg::ExecCommandBegin(_) - | EventMsg::PatchApplyBegin(_) - | EventMsg::TurnComplete(_) - | EventMsg::TurnAborted(_) - | EventMsg::ShutdownComplete + update, + ThreadUpdate::CommandExecutionRequestApproval { .. } + | ThreadUpdate::FileChangeRequestApproval { .. } + | ThreadUpdate::McpServerElicitationRequest { .. } + | ThreadUpdate::PermissionsRequestApproval { .. } + | ThreadUpdate::ToolRequestUserInput { .. } + | ThreadUpdate::TurnCompleted(_) + | ThreadUpdate::ThreadClosed(_) + | ThreadUpdate::ThreadRolledBack { .. } ) } @@ -76,21 +61,18 @@ impl PendingInteractiveReplayState { pub(super) fn note_outbound_op(&mut self, op: &Op) { match op { Op::ExecApproval { id, turn_id, .. } => { - self.exec_approval_call_ids.remove(id); + self.exec_approval_ids.remove(id); if let Some(turn_id) = turn_id { Self::remove_call_id_from_turn_map_entry( - &mut self.exec_approval_call_ids_by_turn_id, + &mut self.exec_approval_ids_by_turn_id, turn_id, id, ); } } Op::PatchApproval { id, .. } => { - self.patch_approval_call_ids.remove(id); - Self::remove_call_id_from_turn_map( - &mut self.patch_approval_call_ids_by_turn_id, - id, - ); + self.patch_approval_ids.remove(id); + Self::remove_call_id_from_turn_map(&mut self.patch_approval_ids_by_turn_id, id); } Op::ResolveElicitation { server_name, @@ -100,32 +82,29 @@ impl PendingInteractiveReplayState { self.elicitation_requests .remove(&ElicitationRequestKey::new( server_name.clone(), - request_id.clone(), + request_id.to_string(), )); } Op::RequestPermissionsResponse { id, .. } => { - self.request_permissions_call_ids.remove(id); + self.request_permissions_ids.remove(id); Self::remove_call_id_from_turn_map( - &mut self.request_permissions_call_ids_by_turn_id, + &mut self.request_permissions_ids_by_turn_id, id, ); } - // `Op::UserInputAnswer` identifies the turn, not the prompt call_id. The UI - // answers queued prompts for the same turn in FIFO order, so remove the oldest - // queued call_id for that turn. Op::UserInputAnswer { id, .. } => { let mut remove_turn_entry = false; - if let Some(call_ids) = self.request_user_input_call_ids_by_turn_id.get_mut(id) { - if !call_ids.is_empty() { - let call_id = call_ids.remove(0); - self.request_user_input_call_ids.remove(&call_id); + if let Some(item_ids) = self.request_user_input_ids_by_turn_id.get_mut(id) { + if !item_ids.is_empty() { + let item_id = item_ids.remove(0); + self.request_user_input_ids.remove(&item_id); } - if call_ids.is_empty() { + if item_ids.is_empty() { remove_turn_entry = true; } } if remove_turn_entry { - self.request_user_input_call_ids_by_turn_id.remove(id); + self.request_user_input_ids_by_turn_id.remove(id); } } Op::Shutdown => self.clear(), @@ -133,202 +112,169 @@ impl PendingInteractiveReplayState { } } - pub(super) fn note_event(&mut self, event: &Event) { - match &event.msg { - EventMsg::ExecApprovalRequest(ev) => { - let approval_id = ev.effective_approval_id(); - self.exec_approval_call_ids.insert(approval_id.clone()); - self.exec_approval_call_ids_by_turn_id - .entry(ev.turn_id.clone()) + pub(super) fn note_update(&mut self, update: &ThreadUpdate) { + match update { + ThreadUpdate::CommandExecutionRequestApproval { params, .. } => { + let approval_id = params + .approval_id + .clone() + .unwrap_or_else(|| params.item_id.clone()); + self.exec_approval_ids.insert(approval_id.clone()); + self.exec_approval_ids_by_turn_id + .entry(params.turn_id.clone()) .or_default() .push(approval_id); } - EventMsg::ExecCommandBegin(ev) => { - self.exec_approval_call_ids.remove(&ev.call_id); - Self::remove_call_id_from_turn_map( - &mut self.exec_approval_call_ids_by_turn_id, - &ev.call_id, - ); - } - EventMsg::ApplyPatchApprovalRequest(ev) => { - self.patch_approval_call_ids.insert(ev.call_id.clone()); - self.patch_approval_call_ids_by_turn_id - .entry(ev.turn_id.clone()) + ThreadUpdate::FileChangeRequestApproval { params, .. } => { + self.patch_approval_ids.insert(params.item_id.clone()); + self.patch_approval_ids_by_turn_id + .entry(params.turn_id.clone()) .or_default() - .push(ev.call_id.clone()); + .push(params.item_id.clone()); } - EventMsg::PatchApplyBegin(ev) => { - self.patch_approval_call_ids.remove(&ev.call_id); - Self::remove_call_id_from_turn_map( - &mut self.patch_approval_call_ids_by_turn_id, - &ev.call_id, - ); - } - EventMsg::ElicitationRequest(ev) => { + ThreadUpdate::McpServerElicitationRequest { request_id, params } => { self.elicitation_requests.insert(ElicitationRequestKey::new( - ev.server_name.clone(), - ev.id.clone(), + params.server_name.clone(), + request_id.to_string(), )); } - EventMsg::RequestUserInput(ev) => { - self.request_user_input_call_ids.insert(ev.call_id.clone()); - self.request_user_input_call_ids_by_turn_id - .entry(ev.turn_id.clone()) + ThreadUpdate::ToolRequestUserInput { params, .. } => { + self.request_user_input_ids.insert(params.item_id.clone()); + self.request_user_input_ids_by_turn_id + .entry(params.turn_id.clone()) .or_default() - .push(ev.call_id.clone()); + .push(params.item_id.clone()); } - EventMsg::RequestPermissions(ev) => { - self.request_permissions_call_ids.insert(ev.call_id.clone()); - self.request_permissions_call_ids_by_turn_id - .entry(ev.turn_id.clone()) + ThreadUpdate::PermissionsRequestApproval { params, .. } => { + self.request_permissions_ids.insert(params.item_id.clone()); + self.request_permissions_ids_by_turn_id + .entry(params.turn_id.clone()) .or_default() - .push(ev.call_id.clone()); + .push(params.item_id.clone()); } - // A turn ending (normally or aborted/replaced) invalidates any unresolved - // turn-scoped approvals, permission prompts, and request_user_input prompts. - EventMsg::TurnComplete(ev) => { - self.clear_exec_approval_turn(&ev.turn_id); - self.clear_patch_approval_turn(&ev.turn_id); - self.clear_request_permissions_turn(&ev.turn_id); - self.clear_request_user_input_turn(&ev.turn_id); + ThreadUpdate::TurnCompleted(notification) => { + self.clear_exec_approval_turn(¬ification.turn.id); + self.clear_patch_approval_turn(¬ification.turn.id); + self.clear_request_permissions_turn(¬ification.turn.id); + self.clear_request_user_input_turn(¬ification.turn.id); } - EventMsg::TurnAborted(ev) => { - if let Some(turn_id) = &ev.turn_id { - self.clear_exec_approval_turn(turn_id); - self.clear_patch_approval_turn(turn_id); - self.clear_request_permissions_turn(turn_id); - self.clear_request_user_input_turn(turn_id); - } - } - EventMsg::ShutdownComplete => self.clear(), + ThreadUpdate::ThreadClosed(_) | ThreadUpdate::ThreadRolledBack { .. } => self.clear(), _ => {} } } - pub(super) fn note_evicted_event(&mut self, event: &Event) { - match &event.msg { - EventMsg::ExecApprovalRequest(ev) => { - let approval_id = ev.effective_approval_id(); - self.exec_approval_call_ids.remove(&approval_id); + pub(super) fn note_evicted_update(&mut self, update: &ThreadUpdate) { + match update { + ThreadUpdate::CommandExecutionRequestApproval { params, .. } => { + let approval_id = params + .approval_id + .clone() + .unwrap_or_else(|| params.item_id.clone()); + self.exec_approval_ids.remove(&approval_id); Self::remove_call_id_from_turn_map_entry( - &mut self.exec_approval_call_ids_by_turn_id, - &ev.turn_id, + &mut self.exec_approval_ids_by_turn_id, + ¶ms.turn_id, &approval_id, ); } - EventMsg::ApplyPatchApprovalRequest(ev) => { - self.patch_approval_call_ids.remove(&ev.call_id); + ThreadUpdate::FileChangeRequestApproval { params, .. } => { + self.patch_approval_ids.remove(¶ms.item_id); Self::remove_call_id_from_turn_map_entry( - &mut self.patch_approval_call_ids_by_turn_id, - &ev.turn_id, - &ev.call_id, + &mut self.patch_approval_ids_by_turn_id, + ¶ms.turn_id, + ¶ms.item_id, ); } - EventMsg::ElicitationRequest(ev) => { + ThreadUpdate::McpServerElicitationRequest { request_id, params } => { self.elicitation_requests .remove(&ElicitationRequestKey::new( - ev.server_name.clone(), - ev.id.clone(), + params.server_name.clone(), + request_id.to_string(), )); } - EventMsg::RequestUserInput(ev) => { - self.request_user_input_call_ids.remove(&ev.call_id); - let mut remove_turn_entry = false; - if let Some(call_ids) = self - .request_user_input_call_ids_by_turn_id - .get_mut(&ev.turn_id) - { - call_ids.retain(|call_id| call_id != &ev.call_id); - if call_ids.is_empty() { - remove_turn_entry = true; - } - } - if remove_turn_entry { - self.request_user_input_call_ids_by_turn_id - .remove(&ev.turn_id); - } + ThreadUpdate::ToolRequestUserInput { params, .. } => { + self.request_user_input_ids.remove(¶ms.item_id); + Self::remove_call_id_from_turn_map_entry( + &mut self.request_user_input_ids_by_turn_id, + ¶ms.turn_id, + ¶ms.item_id, + ); } - EventMsg::RequestPermissions(ev) => { - self.request_permissions_call_ids.remove(&ev.call_id); - let mut remove_turn_entry = false; - if let Some(call_ids) = self - .request_permissions_call_ids_by_turn_id - .get_mut(&ev.turn_id) - { - call_ids.retain(|call_id| call_id != &ev.call_id); - if call_ids.is_empty() { - remove_turn_entry = true; - } - } - if remove_turn_entry { - self.request_permissions_call_ids_by_turn_id - .remove(&ev.turn_id); - } + ThreadUpdate::PermissionsRequestApproval { params, .. } => { + self.request_permissions_ids.remove(¶ms.item_id); + Self::remove_call_id_from_turn_map_entry( + &mut self.request_permissions_ids_by_turn_id, + ¶ms.turn_id, + ¶ms.item_id, + ); } _ => {} } } - pub(super) fn should_replay_snapshot_event(&self, event: &Event) -> bool { - match &event.msg { - EventMsg::ExecApprovalRequest(ev) => self - .exec_approval_call_ids - .contains(&ev.effective_approval_id()), - EventMsg::ApplyPatchApprovalRequest(ev) => { - self.patch_approval_call_ids.contains(&ev.call_id) + pub(super) fn should_replay_snapshot_update(&self, update: &ThreadUpdate) -> bool { + match update { + ThreadUpdate::CommandExecutionRequestApproval { params, .. } => self + .exec_approval_ids + .contains(params.approval_id.as_ref().unwrap_or(¶ms.item_id)), + ThreadUpdate::FileChangeRequestApproval { params, .. } => { + self.patch_approval_ids.contains(¶ms.item_id) } - EventMsg::ElicitationRequest(ev) => { - self.elicitation_requests - .contains(&ElicitationRequestKey::new( - ev.server_name.clone(), - ev.id.clone(), - )) + ThreadUpdate::McpServerElicitationRequest { request_id, params } => self + .elicitation_requests + .contains(&ElicitationRequestKey::new( + params.server_name.clone(), + request_id.to_string(), + )), + ThreadUpdate::ToolRequestUserInput { params, .. } => { + self.request_user_input_ids.contains(¶ms.item_id) } - EventMsg::RequestUserInput(ev) => { - self.request_user_input_call_ids.contains(&ev.call_id) - } - EventMsg::RequestPermissions(ev) => { - self.request_permissions_call_ids.contains(&ev.call_id) + ThreadUpdate::PermissionsRequestApproval { params, .. } => { + self.request_permissions_ids.contains(¶ms.item_id) } _ => true, } } pub(super) fn has_pending_thread_approvals(&self) -> bool { - !self.exec_approval_call_ids.is_empty() - || !self.patch_approval_call_ids.is_empty() + !self.exec_approval_ids.is_empty() + || !self.patch_approval_ids.is_empty() || !self.elicitation_requests.is_empty() - || !self.request_permissions_call_ids.is_empty() + || !self.request_permissions_ids.is_empty() + } + + fn clear(&mut self) { + *self = Self::default(); } fn clear_request_user_input_turn(&mut self, turn_id: &str) { - if let Some(call_ids) = self.request_user_input_call_ids_by_turn_id.remove(turn_id) { - for call_id in call_ids { - self.request_user_input_call_ids.remove(&call_id); + if let Some(item_ids) = self.request_user_input_ids_by_turn_id.remove(turn_id) { + for item_id in item_ids { + self.request_user_input_ids.remove(&item_id); } } } fn clear_request_permissions_turn(&mut self, turn_id: &str) { - if let Some(call_ids) = self.request_permissions_call_ids_by_turn_id.remove(turn_id) { - for call_id in call_ids { - self.request_permissions_call_ids.remove(&call_id); + if let Some(item_ids) = self.request_permissions_ids_by_turn_id.remove(turn_id) { + for item_id in item_ids { + self.request_permissions_ids.remove(&item_id); } } } fn clear_exec_approval_turn(&mut self, turn_id: &str) { - if let Some(call_ids) = self.exec_approval_call_ids_by_turn_id.remove(turn_id) { - for call_id in call_ids { - self.exec_approval_call_ids.remove(&call_id); + if let Some(item_ids) = self.exec_approval_ids_by_turn_id.remove(turn_id) { + for item_id in item_ids { + self.exec_approval_ids.remove(&item_id); } } } fn clear_patch_approval_turn(&mut self, turn_id: &str) { - if let Some(call_ids) = self.patch_approval_call_ids_by_turn_id.remove(turn_id) { - for call_id in call_ids { - self.patch_approval_call_ids.remove(&call_id); + if let Some(item_ids) = self.patch_approval_ids_by_turn_id.remove(turn_id) { + for item_id in item_ids { + self.patch_approval_ids.remove(&item_id); } } } @@ -359,366 +305,4 @@ impl PendingInteractiveReplayState { call_ids_by_turn_id.remove(turn_id); } } - - fn clear(&mut self) { - self.exec_approval_call_ids.clear(); - self.exec_approval_call_ids_by_turn_id.clear(); - self.patch_approval_call_ids.clear(); - self.patch_approval_call_ids_by_turn_id.clear(); - self.elicitation_requests.clear(); - self.request_permissions_call_ids.clear(); - self.request_permissions_call_ids_by_turn_id.clear(); - self.request_user_input_call_ids.clear(); - self.request_user_input_call_ids_by_turn_id.clear(); - } -} - -#[cfg(test)] -mod tests { - use super::super::ThreadEventStore; - use codex_protocol::protocol::Event; - use codex_protocol::protocol::EventMsg; - use codex_protocol::protocol::Op; - use codex_protocol::protocol::TurnAbortReason; - use pretty_assertions::assert_eq; - use std::collections::HashMap; - use std::path::PathBuf; - - #[test] - fn thread_event_snapshot_keeps_pending_request_user_input() { - let mut store = ThreadEventStore::new(8); - let request = Event { - id: "ev-1".to_string(), - msg: EventMsg::RequestUserInput( - codex_protocol::request_user_input::RequestUserInputEvent { - call_id: "call-1".to_string(), - turn_id: "turn-1".to_string(), - questions: Vec::new(), - }, - ), - }; - - store.push_event(request); - - let snapshot = store.snapshot(); - assert_eq!(snapshot.events.len(), 1); - assert!(matches!( - snapshot.events.first().map(|event| &event.msg), - Some(EventMsg::RequestUserInput(_)) - )); - } - - #[test] - fn thread_event_snapshot_drops_resolved_request_user_input_after_user_answer() { - let mut store = ThreadEventStore::new(8); - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::RequestUserInput( - codex_protocol::request_user_input::RequestUserInputEvent { - call_id: "call-1".to_string(), - turn_id: "turn-1".to_string(), - questions: Vec::new(), - }, - ), - }); - - store.note_outbound_op(&Op::UserInputAnswer { - id: "turn-1".to_string(), - response: codex_protocol::request_user_input::RequestUserInputResponse { - answers: HashMap::new(), - }, - }); - - let snapshot = store.snapshot(); - assert!( - snapshot.events.is_empty(), - "resolved request_user_input prompt should not replay on thread switch" - ); - } - - #[test] - fn thread_event_snapshot_drops_resolved_exec_approval_after_outbound_approval_id() { - let mut store = ThreadEventStore::new(8); - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::ExecApprovalRequest( - codex_protocol::protocol::ExecApprovalRequestEvent { - call_id: "call-1".to_string(), - approval_id: Some("approval-1".to_string()), - turn_id: "turn-1".to_string(), - command: vec!["echo".to_string(), "hi".to_string()], - cwd: PathBuf::from("/tmp"), - reason: None, - network_approval_context: None, - proposed_execpolicy_amendment: None, - proposed_network_policy_amendments: None, - additional_permissions: None, - skill_metadata: None, - available_decisions: None, - parsed_cmd: Vec::new(), - }, - ), - }); - - store.note_outbound_op(&Op::ExecApproval { - id: "approval-1".to_string(), - turn_id: Some("turn-1".to_string()), - decision: codex_protocol::protocol::ReviewDecision::Approved, - }); - - let snapshot = store.snapshot(); - assert!( - snapshot.events.is_empty(), - "resolved exec approval prompt should not replay on thread switch" - ); - } - - #[test] - fn thread_event_snapshot_drops_answered_request_user_input_for_multi_prompt_turn() { - let mut store = ThreadEventStore::new(8); - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::RequestUserInput( - codex_protocol::request_user_input::RequestUserInputEvent { - call_id: "call-1".to_string(), - turn_id: "turn-1".to_string(), - questions: Vec::new(), - }, - ), - }); - - store.note_outbound_op(&Op::UserInputAnswer { - id: "turn-1".to_string(), - response: codex_protocol::request_user_input::RequestUserInputResponse { - answers: HashMap::new(), - }, - }); - - store.push_event(Event { - id: "ev-2".to_string(), - msg: EventMsg::RequestUserInput( - codex_protocol::request_user_input::RequestUserInputEvent { - call_id: "call-2".to_string(), - turn_id: "turn-1".to_string(), - questions: Vec::new(), - }, - ), - }); - - let snapshot = store.snapshot(); - assert_eq!(snapshot.events.len(), 1); - assert!(matches!( - snapshot.events.first().map(|event| &event.msg), - Some(EventMsg::RequestUserInput(ev)) if ev.call_id == "call-2" - )); - } - - #[test] - fn thread_event_snapshot_keeps_newer_request_user_input_pending_when_same_turn_has_queue() { - let mut store = ThreadEventStore::new(8); - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::RequestUserInput( - codex_protocol::request_user_input::RequestUserInputEvent { - call_id: "call-1".to_string(), - turn_id: "turn-1".to_string(), - questions: Vec::new(), - }, - ), - }); - store.push_event(Event { - id: "ev-2".to_string(), - msg: EventMsg::RequestUserInput( - codex_protocol::request_user_input::RequestUserInputEvent { - call_id: "call-2".to_string(), - turn_id: "turn-1".to_string(), - questions: Vec::new(), - }, - ), - }); - - store.note_outbound_op(&Op::UserInputAnswer { - id: "turn-1".to_string(), - response: codex_protocol::request_user_input::RequestUserInputResponse { - answers: HashMap::new(), - }, - }); - - let snapshot = store.snapshot(); - assert_eq!(snapshot.events.len(), 1); - assert!(matches!( - snapshot.events.first().map(|event| &event.msg), - Some(EventMsg::RequestUserInput(ev)) if ev.call_id == "call-2" - )); - } - - #[test] - fn thread_event_snapshot_drops_resolved_patch_approval_after_outbound_approval() { - let mut store = ThreadEventStore::new(8); - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::ApplyPatchApprovalRequest( - codex_protocol::protocol::ApplyPatchApprovalRequestEvent { - call_id: "call-1".to_string(), - turn_id: "turn-1".to_string(), - changes: HashMap::new(), - reason: None, - grant_root: None, - }, - ), - }); - - store.note_outbound_op(&Op::PatchApproval { - id: "call-1".to_string(), - decision: codex_protocol::protocol::ReviewDecision::Approved, - }); - - let snapshot = store.snapshot(); - assert!( - snapshot.events.is_empty(), - "resolved patch approval prompt should not replay on thread switch" - ); - } - - #[test] - fn thread_event_snapshot_drops_pending_approvals_when_turn_aborts() { - let mut store = ThreadEventStore::new(8); - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::ExecApprovalRequest( - codex_protocol::protocol::ExecApprovalRequestEvent { - call_id: "exec-call-1".to_string(), - approval_id: Some("approval-1".to_string()), - turn_id: "turn-1".to_string(), - command: vec!["echo".to_string(), "hi".to_string()], - cwd: PathBuf::from("/tmp"), - reason: None, - network_approval_context: None, - proposed_execpolicy_amendment: None, - proposed_network_policy_amendments: None, - additional_permissions: None, - skill_metadata: None, - available_decisions: None, - parsed_cmd: Vec::new(), - }, - ), - }); - store.push_event(Event { - id: "ev-2".to_string(), - msg: EventMsg::ApplyPatchApprovalRequest( - codex_protocol::protocol::ApplyPatchApprovalRequestEvent { - call_id: "patch-call-1".to_string(), - turn_id: "turn-1".to_string(), - changes: HashMap::new(), - reason: None, - grant_root: None, - }, - ), - }); - store.push_event(Event { - id: "ev-3".to_string(), - msg: EventMsg::TurnAborted(codex_protocol::protocol::TurnAbortedEvent { - turn_id: Some("turn-1".to_string()), - reason: TurnAbortReason::Replaced, - }), - }); - - let snapshot = store.snapshot(); - assert!(snapshot.events.iter().all(|event| { - !matches!( - &event.msg, - EventMsg::ExecApprovalRequest(_) | EventMsg::ApplyPatchApprovalRequest(_) - ) - })); - } - - #[test] - fn thread_event_snapshot_drops_resolved_elicitation_after_outbound_resolution() { - let mut store = ThreadEventStore::new(8); - let request_id = codex_protocol::mcp::RequestId::String("request-1".to_string()); - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::ElicitationRequest(codex_protocol::approvals::ElicitationRequestEvent { - turn_id: Some("turn-1".to_string()), - server_name: "server-1".to_string(), - id: request_id.clone(), - request: codex_protocol::approvals::ElicitationRequest::Form { - meta: None, - message: "Please confirm".to_string(), - requested_schema: serde_json::json!({ - "type": "object", - "properties": {} - }), - }, - }), - }); - - store.note_outbound_op(&Op::ResolveElicitation { - server_name: "server-1".to_string(), - request_id, - decision: codex_protocol::approvals::ElicitationAction::Accept, - content: None, - meta: None, - }); - - let snapshot = store.snapshot(); - assert!( - snapshot.events.is_empty(), - "resolved elicitation prompt should not replay on thread switch" - ); - } - - #[test] - fn thread_event_store_reports_pending_thread_approvals() { - let mut store = ThreadEventStore::new(8); - assert_eq!(store.has_pending_thread_approvals(), false); - - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::ExecApprovalRequest( - codex_protocol::protocol::ExecApprovalRequestEvent { - call_id: "call-1".to_string(), - approval_id: None, - turn_id: "turn-1".to_string(), - command: vec!["echo".to_string(), "hi".to_string()], - cwd: PathBuf::from("/tmp"), - reason: None, - network_approval_context: None, - proposed_execpolicy_amendment: None, - proposed_network_policy_amendments: None, - additional_permissions: None, - skill_metadata: None, - available_decisions: None, - parsed_cmd: Vec::new(), - }, - ), - }); - - assert_eq!(store.has_pending_thread_approvals(), true); - - store.note_outbound_op(&Op::ExecApproval { - id: "call-1".to_string(), - turn_id: Some("turn-1".to_string()), - decision: codex_protocol::protocol::ReviewDecision::Approved, - }); - - assert_eq!(store.has_pending_thread_approvals(), false); - } - - #[test] - fn request_user_input_does_not_count_as_pending_thread_approval() { - let mut store = ThreadEventStore::new(8); - store.push_event(Event { - id: "ev-1".to_string(), - msg: EventMsg::RequestUserInput( - codex_protocol::request_user_input::RequestUserInputEvent { - call_id: "call-1".to_string(), - turn_id: "turn-1".to_string(), - questions: Vec::new(), - }, - ), - }); - - assert_eq!(store.has_pending_thread_approvals(), false); - } } diff --git a/codex-rs/tui/src/bottom_pane/mcp_server_elicitation.rs b/codex-rs/tui/src/bottom_pane/mcp_server_elicitation.rs index 975b8701d2..98263df2ac 100644 --- a/codex-rs/tui/src/bottom_pane/mcp_server_elicitation.rs +++ b/codex-rs/tui/src/bottom_pane/mcp_server_elicitation.rs @@ -5,6 +5,9 @@ use std::path::PathBuf; use codex_app_server_protocol::McpElicitationEnumSchema; use codex_app_server_protocol::McpElicitationPrimitiveSchema; use codex_app_server_protocol::McpElicitationSingleSelectEnumSchema; +use codex_app_server_protocol::McpServerElicitationRequest; +use codex_app_server_protocol::McpServerElicitationRequestParams; +use codex_app_server_protocol::RequestId as AppServerRequestId; use codex_protocol::ThreadId; use codex_protocol::approvals::ElicitationAction; use codex_protocol::approvals::ElicitationRequest; @@ -200,6 +203,47 @@ impl FooterTip { } impl McpServerElicitationFormRequest { + pub(crate) fn from_app_server_request( + thread_id: ThreadId, + request_id: AppServerRequestId, + params: McpServerElicitationRequestParams, + ) -> Option { + let request = match params.request { + McpServerElicitationRequest::Form { + meta, + message, + requested_schema, + } => ElicitationRequest::Form { + meta, + message, + requested_schema: serde_json::to_value(requested_schema).ok()?, + }, + McpServerElicitationRequest::Url { + meta, + message, + url, + elicitation_id, + } => ElicitationRequest::Url { + meta, + message, + url, + elicitation_id, + }, + }; + Self::from_event( + thread_id, + ElicitationRequestEvent { + turn_id: params.turn_id, + server_name: params.server_name, + id: match request_id { + AppServerRequestId::String(value) => McpRequestId::String(value), + AppServerRequestId::Integer(value) => McpRequestId::Integer(value), + }, + request, + }, + ) + } + pub(crate) fn from_event( thread_id: ThreadId, request: ElicitationRequestEvent, diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index c4dd63ce03..6006ec11a3 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -49,6 +49,7 @@ use crate::status::format_directory_display; use crate::status::format_tokens_compact; use crate::status::rate_limit_snapshot_display_for_limit; use crate::text_formatting::proper_join; +use crate::thread_update::ThreadUpdate; use crate::version::CODEX_CLI_VERSION; use codex_app_server_protocol::ConfigLayerSource; use codex_chatgpt::connectors; @@ -77,6 +78,7 @@ use codex_otel::RuntimeMetricsSummary; use codex_otel::SessionTelemetry; use codex_protocol::ThreadId; use codex_protocol::account::PlanType; +use codex_protocol::approvals::ElicitationRequest; use codex_protocol::approvals::ElicitationRequestEvent; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::CollaborationModeMask; @@ -4844,6 +4846,436 @@ impl ChatWidget { self.dispatch_event_msg(None, msg, Some(ReplayKind::ThreadSnapshot)); } + pub(crate) fn handle_thread_update(&mut self, update: ThreadUpdate) { + self.dispatch_thread_update(update, None); + } + + pub(crate) fn handle_thread_update_replay(&mut self, update: ThreadUpdate) { + self.dispatch_thread_update(update, Some(ReplayKind::ThreadSnapshot)); + } + + fn dispatch_thread_update(&mut self, update: ThreadUpdate, replay_kind: Option) { + let from_replay = replay_kind.is_some(); + let is_resume_initial_replay = + matches!(replay_kind, Some(ReplayKind::ResumeInitialMessages)); + if !is_resume_initial_replay { + self.restore_retry_status_header_if_present(); + } + + match update { + ThreadUpdate::SessionConfigured(event) => self.on_session_configured(event), + ThreadUpdate::ThreadNameUpdated(event) => { + match ThreadId::from_string(&event.thread_id) { + Ok(thread_id) => self.on_thread_name_updated( + codex_protocol::protocol::ThreadNameUpdatedEvent { + thread_id, + thread_name: event.thread_name, + }, + ), + Err(err) => tracing::warn!( + thread_id = %event.thread_id, + "ignoring thread/nameUpdated notification with invalid thread id: {err}" + ), + } + } + ThreadUpdate::ThreadTokenUsageUpdated(event) => { + let info = Some(TokenUsageInfo { + total_token_usage: token_usage_from_api(&event.token_usage.total), + last_token_usage: token_usage_from_api(&event.token_usage.last), + model_context_window: event.token_usage.model_context_window, + }); + self.set_token_info(info); + } + ThreadUpdate::TurnStarted(_event) => { + if !is_resume_initial_replay { + self.on_task_started(); + self.apply_turn_started_context_window(None); + } + } + ThreadUpdate::TurnCompleted(event) => { + let last_agent_message = + event.turn.items.iter().rev().find_map(|item| match item { + codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } => { + Some(text.clone()) + } + _ => None, + }); + match event.turn.status { + codex_app_server_protocol::TurnStatus::Completed + | codex_app_server_protocol::TurnStatus::Failed => { + self.on_task_complete(last_agent_message, from_replay); + if let Some(error) = event.turn.error { + self.on_error(error.message); + } + } + codex_app_server_protocol::TurnStatus::Interrupted => { + self.on_interrupted_turn(TurnAbortReason::Interrupted); + } + codex_app_server_protocol::TurnStatus::InProgress => {} + } + } + ThreadUpdate::TurnDiffUpdated(event) => self.on_turn_diff(event.diff), + ThreadUpdate::TurnPlanUpdated(event) => { + self.on_plan_update(update_plan_args_from_api(event.explanation, event.plan)) + } + ThreadUpdate::AgentMessageDelta(event) => self.on_agent_message_delta(event.delta), + ThreadUpdate::PlanDelta(event) => self.on_plan_delta(event.delta), + ThreadUpdate::ReasoningSummaryTextDelta(event) => { + self.on_agent_reasoning_delta(event.delta); + } + ThreadUpdate::ReasoningSummaryPartAdded(_) => self.on_reasoning_section_break(), + ThreadUpdate::ReasoningTextDelta(event) => self.on_agent_reasoning_delta(event.delta), + ThreadUpdate::TerminalInteraction(event) => { + self.on_terminal_interaction(codex_protocol::protocol::TerminalInteractionEvent { + call_id: event.item_id, + process_id: event.process_id, + stdin: event.stdin, + }); + } + ThreadUpdate::CommandExecutionOutputDelta(event) => { + self.on_exec_command_output_delta( + codex_protocol::protocol::ExecCommandOutputDeltaEvent { + call_id: event.item_id, + stream: codex_protocol::protocol::ExecOutputStream::Stdout, + chunk: event.delta.into_bytes(), + }, + ); + } + ThreadUpdate::FileChangeOutputDelta(_) => {} + ThreadUpdate::McpToolCallProgress(_) => {} + ThreadUpdate::ItemStarted(event) => { + self.handle_thread_item_started(event.item); + } + ThreadUpdate::ItemCompleted(event) => { + self.handle_thread_item_completed(event.item, from_replay); + } + ThreadUpdate::ItemGuardianApprovalReviewStarted(event) => { + self.on_guardian_assessment(codex_protocol::approvals::GuardianAssessmentEvent { + id: event.target_item_id, + turn_id: event.turn_id, + status: guardian_status_from_api(event.review.status), + risk_score: event.review.risk_score, + risk_level: event.review.risk_level.map(guardian_risk_level_from_api), + rationale: event.review.rationale, + action: event.action, + }); + } + ThreadUpdate::ItemGuardianApprovalReviewCompleted(event) => { + self.on_guardian_assessment(codex_protocol::approvals::GuardianAssessmentEvent { + id: event.target_item_id, + turn_id: event.turn_id, + status: guardian_status_from_api(event.review.status), + risk_score: event.review.risk_score, + risk_level: event.review.risk_level.map(guardian_risk_level_from_api), + rationale: event.review.rationale, + action: event.action, + }); + } + ThreadUpdate::CommandExecutionRequestApproval { params, .. } => { + self.on_exec_approval_request(String::new(), exec_approval_event_from_api(params)); + } + ThreadUpdate::FileChangeRequestApproval { params, .. } => { + self.on_apply_patch_approval_request( + String::new(), + patch_approval_event_from_api(params), + ); + } + ThreadUpdate::McpServerElicitationRequest { request_id, params } => { + self.on_elicitation_request(elicitation_event_from_api(request_id, params)); + } + ThreadUpdate::PermissionsRequestApproval { params, .. } => { + self.on_request_permissions(request_permissions_event_from_api(params)); + } + ThreadUpdate::ToolRequestUserInput { params, .. } => { + self.on_request_user_input(request_user_input_event_from_api(params)); + } + ThreadUpdate::Error(event) => self.on_error(event.error.message), + ThreadUpdate::ModelRerouted(_) => {} + ThreadUpdate::DeprecationNotice(event) => { + self.on_deprecation_notice(codex_protocol::protocol::DeprecationNoticeEvent { + summary: event.summary, + details: event.details, + }); + } + ThreadUpdate::HookStarted(event) => { + self.on_hook_started(codex_protocol::protocol::HookStartedEvent { + turn_id: event.turn_id, + run: hook_run_summary_from_api(event.run), + }); + } + ThreadUpdate::HookCompleted(event) => { + self.on_hook_completed(codex_protocol::protocol::HookCompletedEvent { + turn_id: event.turn_id, + run: hook_run_summary_from_api(event.run), + }); + } + ThreadUpdate::ThreadRealtimeStarted(event) => { + if !from_replay { + self.on_realtime_conversation_started( + codex_protocol::protocol::RealtimeConversationStartedEvent { + session_id: event.session_id, + }, + ); + } + } + ThreadUpdate::ThreadRealtimeItemAdded(_) => {} + ThreadUpdate::ThreadRealtimeOutputAudioDelta(_) => {} + ThreadUpdate::ThreadRealtimeError(event) => { + if !from_replay { + self.on_stream_error(event.message, None); + } + } + ThreadUpdate::ThreadRealtimeClosed(event) => { + if !from_replay { + self.on_realtime_conversation_closed( + codex_protocol::protocol::RealtimeConversationClosedEvent { + reason: event.reason, + }, + ); + } + } + ThreadUpdate::DynamicToolCall { .. } => {} + ThreadUpdate::ThreadStarted(_) + | ThreadUpdate::ThreadStatusChanged(_) + | ThreadUpdate::ThreadClosed(_) => {} + ThreadUpdate::ThreadRolledBack { num_turns } => { + self.last_copyable_output = None; + if from_replay { + self.app_event_tx + .send(AppEvent::ApplyThreadRollback { num_turns }); + } + } + } + + if !from_replay && self.agent_turn_running { + self.refresh_runtime_metrics(); + } + } + + fn handle_thread_item_started(&mut self, item: codex_app_server_protocol::ThreadItem) { + match item { + codex_app_server_protocol::ThreadItem::CommandExecution { + id, + command, + cwd, + process_id, + .. + } => { + let command = vec![command]; + let parsed_cmd = codex_shell_command::parse_command::parse_command(&command); + self.on_exec_command_begin(ExecCommandBeginEvent { + call_id: id, + process_id, + turn_id: String::new(), + command, + cwd, + parsed_cmd, + source: ExecCommandSource::Agent, + interaction_input: None, + }); + } + codex_app_server_protocol::ThreadItem::FileChange { id, changes, .. } => { + self.on_patch_apply_begin(PatchApplyBeginEvent { + call_id: id, + turn_id: String::new(), + auto_approved: false, + changes: file_changes_from_api(changes), + }); + } + codex_app_server_protocol::ThreadItem::McpToolCall { + id, + server, + tool, + arguments, + .. + } => { + self.on_mcp_tool_call_begin(McpToolCallBeginEvent { + call_id: id, + invocation: codex_protocol::protocol::McpInvocation { + server, + tool, + arguments: Some(arguments), + }, + }); + } + codex_app_server_protocol::ThreadItem::WebSearch { id, .. } => { + self.on_web_search_begin(codex_protocol::protocol::WebSearchBeginEvent { + call_id: id, + }); + } + codex_app_server_protocol::ThreadItem::ImageView { id, path } => { + self.on_view_image_tool_call(codex_protocol::protocol::ViewImageToolCallEvent { + call_id: id, + path: PathBuf::from(path), + }); + } + codex_app_server_protocol::ThreadItem::ImageGeneration { id, .. } => { + self.on_image_generation_begin(ImageGenerationBeginEvent { call_id: id }); + } + _ => {} + } + } + + fn handle_thread_item_completed( + &mut self, + item: codex_app_server_protocol::ThreadItem, + from_replay: bool, + ) { + match item { + codex_app_server_protocol::ThreadItem::UserMessage { content, .. } => { + let event = user_message_event_from_api(content); + if from_replay || self.should_render_realtime_user_message_event(&event) { + self.on_user_message_event(event); + } + } + codex_app_server_protocol::ThreadItem::AgentMessage { id, text, phase } => { + if from_replay || self.is_review_mode { + self.on_agent_message(text.clone()); + } + self.on_agent_message_item_completed(AgentMessageItem { + id, + content: vec![AgentMessageContent::Text { text }], + phase, + }); + } + codex_app_server_protocol::ThreadItem::Plan { text, .. } => { + self.on_plan_item_completed(text); + } + codex_app_server_protocol::ThreadItem::Reasoning { content, .. } => { + for part in content { + self.on_agent_reasoning_delta(part); + } + self.on_agent_reasoning_final(); + } + codex_app_server_protocol::ThreadItem::CommandExecution { + id, + command, + cwd, + process_id, + aggregated_output, + exit_code, + duration_ms, + status, + .. + } => { + let command = vec![command]; + let parsed_cmd = codex_shell_command::parse_command::parse_command(&command); + self.on_exec_command_end(ExecCommandEndEvent { + call_id: id, + process_id, + turn_id: String::new(), + command, + cwd, + parsed_cmd, + source: ExecCommandSource::Agent, + interaction_input: None, + stdout: aggregated_output.clone().unwrap_or_default(), + stderr: String::new(), + aggregated_output: aggregated_output.unwrap_or_default(), + exit_code: exit_code.unwrap_or_default(), + duration: std::time::Duration::from_millis( + duration_ms.unwrap_or_default().max(0) as u64, + ), + formatted_output: String::new(), + status: exec_status_from_api(status), + }); + } + codex_app_server_protocol::ThreadItem::FileChange { + id, + status, + changes, + .. + } => { + self.on_patch_apply_end(codex_protocol::protocol::PatchApplyEndEvent { + call_id: id, + turn_id: String::new(), + stdout: String::new(), + stderr: String::new(), + success: matches!( + status, + codex_app_server_protocol::PatchApplyStatus::Completed + ), + changes: file_changes_from_api(changes), + status: patch_apply_status_from_api(status), + }); + } + codex_app_server_protocol::ThreadItem::McpToolCall { + id, + server, + tool, + arguments, + result, + error, + duration_ms, + .. + } => { + self.on_mcp_tool_call_end(McpToolCallEndEvent { + call_id: id, + invocation: codex_protocol::protocol::McpInvocation { + server, + tool, + arguments: Some(arguments), + }, + duration: std::time::Duration::from_millis( + duration_ms.unwrap_or_default().max(0) as u64, + ), + result: match result { + Some(result) => Ok(mcp_result_from_api(result)), + None => Err(error.map(|err| err.message).unwrap_or_default()), + }, + }); + } + codex_app_server_protocol::ThreadItem::WebSearch { id, query, action } => { + self.on_web_search_end(codex_protocol::protocol::WebSearchEndEvent { + call_id: id, + query, + action: web_search_action_from_api(action), + }); + } + codex_app_server_protocol::ThreadItem::ImageGeneration { + id, + status, + revised_prompt, + result, + } => { + self.on_image_generation_end(ImageGenerationEndEvent { + call_id: id, + status, + revised_prompt, + result, + saved_path: None, + }); + } + codex_app_server_protocol::ThreadItem::EnteredReviewMode { review, .. } => { + self.on_entered_review_mode( + ReviewRequest { + target: ReviewTarget::Custom { + instructions: review.clone(), + }, + user_facing_hint: Some(review), + }, + from_replay, + ); + } + codex_app_server_protocol::ThreadItem::ExitedReviewMode { review, .. } => { + self.on_exited_review_mode(ExitedReviewModeEvent { + review_output: Some(codex_protocol::protocol::ReviewOutputEvent { + findings: Vec::new(), + overall_correctness: String::new(), + overall_explanation: review, + overall_confidence_score: 0.0, + }), + }); + } + codex_app_server_protocol::ThreadItem::ContextCompaction { .. } => { + self.on_agent_message("Context compacted".to_owned()); + } + codex_app_server_protocol::ThreadItem::ImageView { .. } + | codex_app_server_protocol::ThreadItem::DynamicToolCall { .. } + | codex_app_server_protocol::ThreadItem::CollabAgentToolCall { .. } => {} + } + } + /// Dispatch a protocol `EventMsg` to the appropriate handler. /// /// `id` is `Some` for live events and `None` for replayed events from @@ -8984,6 +9416,337 @@ fn hook_event_label(event_name: codex_protocol::protocol::HookEventName) -> &'st } } +fn token_usage_from_api( + usage: &codex_app_server_protocol::TokenUsageBreakdown, +) -> codex_protocol::protocol::TokenUsage { + codex_protocol::protocol::TokenUsage { + total_tokens: usage.total_tokens, + input_tokens: usage.input_tokens, + cached_input_tokens: usage.cached_input_tokens, + output_tokens: usage.output_tokens, + reasoning_output_tokens: usage.reasoning_output_tokens, + } +} + +fn update_plan_args_from_api( + explanation: Option, + plan: Vec, +) -> codex_protocol::plan_tool::UpdatePlanArgs { + codex_protocol::plan_tool::UpdatePlanArgs { + explanation, + plan: plan + .into_iter() + .map(|step| codex_protocol::plan_tool::PlanItemArg { + step: step.step, + status: match step.status { + codex_app_server_protocol::TurnPlanStepStatus::Pending => { + codex_protocol::plan_tool::StepStatus::Pending + } + codex_app_server_protocol::TurnPlanStepStatus::InProgress => { + codex_protocol::plan_tool::StepStatus::InProgress + } + codex_app_server_protocol::TurnPlanStepStatus::Completed => { + codex_protocol::plan_tool::StepStatus::Completed + } + }, + }) + .collect(), + } +} + +fn guardian_status_from_api( + status: codex_app_server_protocol::GuardianApprovalReviewStatus, +) -> GuardianAssessmentStatus { + match status { + codex_app_server_protocol::GuardianApprovalReviewStatus::InProgress => { + GuardianAssessmentStatus::InProgress + } + codex_app_server_protocol::GuardianApprovalReviewStatus::Approved => { + GuardianAssessmentStatus::Approved + } + codex_app_server_protocol::GuardianApprovalReviewStatus::Denied => { + GuardianAssessmentStatus::Denied + } + codex_app_server_protocol::GuardianApprovalReviewStatus::Aborted => { + GuardianAssessmentStatus::Aborted + } + } +} + +fn guardian_risk_level_from_api( + level: codex_app_server_protocol::GuardianRiskLevel, +) -> codex_protocol::approvals::GuardianRiskLevel { + match level { + codex_app_server_protocol::GuardianRiskLevel::Low => { + codex_protocol::approvals::GuardianRiskLevel::Low + } + codex_app_server_protocol::GuardianRiskLevel::Medium => { + codex_protocol::approvals::GuardianRiskLevel::Medium + } + codex_app_server_protocol::GuardianRiskLevel::High => { + codex_protocol::approvals::GuardianRiskLevel::High + } + } +} + +fn exec_approval_event_from_api( + params: codex_app_server_protocol::CommandExecutionRequestApprovalParams, +) -> ExecApprovalRequestEvent { + let command = params + .command + .map(|command| vec![command]) + .unwrap_or_default(); + ExecApprovalRequestEvent { + call_id: params.item_id, + approval_id: params.approval_id, + turn_id: params.turn_id, + command: command.clone(), + cwd: params.cwd.unwrap_or_default(), + reason: params.reason, + network_approval_context: params + .network_approval_context + .and_then(|context| serde_json::from_value(serde_json::to_value(context).ok()?).ok()), + proposed_execpolicy_amendment: None, + proposed_network_policy_amendments: None, + additional_permissions: params.additional_permissions.and_then(|permissions| { + serde_json::from_value(serde_json::to_value(permissions).ok()?).ok() + }), + skill_metadata: None, + available_decisions: None, + parsed_cmd: codex_shell_command::parse_command::parse_command(&command), + } +} + +fn patch_approval_event_from_api( + params: codex_app_server_protocol::FileChangeRequestApprovalParams, +) -> ApplyPatchApprovalRequestEvent { + ApplyPatchApprovalRequestEvent { + call_id: params.item_id, + turn_id: params.turn_id, + changes: HashMap::new(), + reason: params.reason, + grant_root: params.grant_root, + } +} + +fn elicitation_event_from_api( + request_id: codex_app_server_protocol::RequestId, + params: codex_app_server_protocol::McpServerElicitationRequestParams, +) -> ElicitationRequestEvent { + ElicitationRequestEvent { + turn_id: params.turn_id, + server_name: params.server_name, + id: match request_id { + codex_app_server_protocol::RequestId::String(value) => { + codex_protocol::mcp::RequestId::String(value) + } + codex_app_server_protocol::RequestId::Integer(value) => { + codex_protocol::mcp::RequestId::Integer(value) + } + }, + request: match params.request { + codex_app_server_protocol::McpServerElicitationRequest::Form { + meta, + message, + requested_schema, + } => ElicitationRequest::Form { + meta, + message, + requested_schema: serde_json::to_value(requested_schema) + .unwrap_or(serde_json::Value::Null), + }, + codex_app_server_protocol::McpServerElicitationRequest::Url { + meta, + message, + url, + elicitation_id, + } => ElicitationRequest::Url { + meta, + message, + url, + elicitation_id, + }, + }, + } +} + +fn request_permissions_event_from_api( + params: codex_app_server_protocol::PermissionsRequestApprovalParams, +) -> codex_protocol::request_permissions::RequestPermissionsEvent { + codex_protocol::request_permissions::RequestPermissionsEvent { + call_id: params.item_id, + turn_id: params.turn_id, + reason: params.reason, + permissions: serde_json::from_value( + serde_json::to_value(params.permissions).unwrap_or(serde_json::Value::Null), + ) + .unwrap_or_default(), + } +} + +fn request_user_input_event_from_api( + params: codex_app_server_protocol::ToolRequestUserInputParams, +) -> codex_protocol::request_user_input::RequestUserInputEvent { + codex_protocol::request_user_input::RequestUserInputEvent { + call_id: params.item_id, + turn_id: params.turn_id, + questions: params + .questions + .into_iter() + .map( + |question| codex_protocol::request_user_input::RequestUserInputQuestion { + id: question.id, + header: question.header, + question: question.question, + is_other: question.is_other, + is_secret: question.is_secret, + options: question.options.map(|options| { + options + .into_iter() + .map(|option| { + codex_protocol::request_user_input::RequestUserInputQuestionOption { + label: option.label, + description: option.description, + } + }) + .collect() + }), + }, + ) + .collect(), + } +} + +fn file_changes_from_api( + changes: Vec, +) -> HashMap { + changes + .into_iter() + .map(|change| { + let file_change = match change.kind { + codex_app_server_protocol::PatchChangeKind::Add => { + codex_protocol::protocol::FileChange::Add { + content: change.diff, + } + } + codex_app_server_protocol::PatchChangeKind::Delete => { + codex_protocol::protocol::FileChange::Delete { + content: change.diff, + } + } + codex_app_server_protocol::PatchChangeKind::Update { move_path } => { + codex_protocol::protocol::FileChange::Update { + unified_diff: change.diff, + move_path, + } + } + }; + (PathBuf::from(change.path), file_change) + }) + .collect() +} + +fn user_message_event_from_api( + content: Vec, +) -> codex_protocol::protocol::UserMessageEvent { + let mut message = String::new(); + let mut images = Vec::new(); + let mut local_images = Vec::new(); + let mut text_elements = Vec::new(); + for item in content { + match item { + codex_app_server_protocol::UserInput::Text { + text, + text_elements: api_elements, + } => { + message.push_str(&text); + text_elements.extend(api_elements.into_iter().map(Into::into)); + } + codex_app_server_protocol::UserInput::Image { url } => images.push(url), + codex_app_server_protocol::UserInput::LocalImage { path } => local_images.push(path), + codex_app_server_protocol::UserInput::Skill { .. } + | codex_app_server_protocol::UserInput::Mention { .. } => {} + } + } + codex_protocol::protocol::UserMessageEvent { + message, + images: Some(images), + local_images, + text_elements, + } +} + +fn exec_status_from_api( + status: codex_app_server_protocol::CommandExecutionStatus, +) -> codex_protocol::protocol::ExecCommandStatus { + match status { + codex_app_server_protocol::CommandExecutionStatus::InProgress + | codex_app_server_protocol::CommandExecutionStatus::Completed => { + codex_protocol::protocol::ExecCommandStatus::Completed + } + codex_app_server_protocol::CommandExecutionStatus::Failed => { + codex_protocol::protocol::ExecCommandStatus::Failed + } + codex_app_server_protocol::CommandExecutionStatus::Declined => { + codex_protocol::protocol::ExecCommandStatus::Declined + } + } +} + +fn patch_apply_status_from_api( + status: codex_app_server_protocol::PatchApplyStatus, +) -> codex_protocol::protocol::PatchApplyStatus { + match status { + codex_app_server_protocol::PatchApplyStatus::InProgress + | codex_app_server_protocol::PatchApplyStatus::Completed => { + codex_protocol::protocol::PatchApplyStatus::Completed + } + codex_app_server_protocol::PatchApplyStatus::Failed => { + codex_protocol::protocol::PatchApplyStatus::Failed + } + codex_app_server_protocol::PatchApplyStatus::Declined => { + codex_protocol::protocol::PatchApplyStatus::Declined + } + } +} + +fn hook_run_summary_from_api( + run: codex_app_server_protocol::HookRunSummary, +) -> codex_protocol::protocol::HookRunSummary { + serde_json::from_value(serde_json::to_value(run).unwrap_or(serde_json::Value::Null)) + .unwrap_or_else(|err| panic!("failed to convert app-server hook run summary: {err}")) +} + +fn mcp_result_from_api( + result: codex_app_server_protocol::McpToolCallResult, +) -> codex_protocol::mcp::CallToolResult { + codex_protocol::mcp::CallToolResult { + content: result.content, + structured_content: result.structured_content, + is_error: None, + meta: None, + } +} + +fn web_search_action_from_api( + action: Option, +) -> codex_protocol::models::WebSearchAction { + match action.unwrap_or(codex_app_server_protocol::WebSearchAction::Other) { + codex_app_server_protocol::WebSearchAction::Search { query, queries } => { + codex_protocol::models::WebSearchAction::Search { query, queries } + } + codex_app_server_protocol::WebSearchAction::OpenPage { url } => { + codex_protocol::models::WebSearchAction::OpenPage { url } + } + codex_app_server_protocol::WebSearchAction::FindInPage { url, pattern } => { + codex_protocol::models::WebSearchAction::FindInPage { url, pattern } + } + codex_app_server_protocol::WebSearchAction::Other => { + codex_protocol::models::WebSearchAction::Other + } + } +} + #[cfg(test)] pub(crate) fn show_review_commit_picker_with_entries( chat: &mut ChatWidget, diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index d0bd596cc3..9294f767b4 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -119,6 +119,7 @@ mod style; mod terminal_palette; mod text_formatting; mod theme_picker; +mod thread_update; mod tooltips; mod tui; mod ui_consts; diff --git a/codex-rs/tui/src/thread_update.rs b/codex-rs/tui/src/thread_update.rs new file mode 100644 index 0000000000..ea1935700c --- /dev/null +++ b/codex-rs/tui/src/thread_update.rs @@ -0,0 +1,170 @@ +use codex_app_server_protocol::AgentMessageDeltaNotification; +use codex_app_server_protocol::CommandExecutionOutputDeltaNotification; +use codex_app_server_protocol::CommandExecutionRequestApprovalParams; +use codex_app_server_protocol::DeprecationNoticeNotification; +use codex_app_server_protocol::DynamicToolCallParams; +use codex_app_server_protocol::ErrorNotification; +use codex_app_server_protocol::FileChangeOutputDeltaNotification; +use codex_app_server_protocol::FileChangeRequestApprovalParams; +use codex_app_server_protocol::HookCompletedNotification; +use codex_app_server_protocol::HookStartedNotification; +use codex_app_server_protocol::ItemCompletedNotification; +use codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification; +use codex_app_server_protocol::ItemGuardianApprovalReviewStartedNotification; +use codex_app_server_protocol::ItemStartedNotification; +use codex_app_server_protocol::McpServerElicitationRequestParams; +use codex_app_server_protocol::McpToolCallProgressNotification; +use codex_app_server_protocol::ModelReroutedNotification; +use codex_app_server_protocol::PermissionsRequestApprovalParams; +use codex_app_server_protocol::PlanDeltaNotification; +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::TerminalInteractionNotification; +use codex_app_server_protocol::ThreadClosedNotification; +use codex_app_server_protocol::ThreadNameUpdatedNotification; +use codex_app_server_protocol::ThreadRealtimeClosedNotification; +use codex_app_server_protocol::ThreadRealtimeErrorNotification; +use codex_app_server_protocol::ThreadRealtimeItemAddedNotification; +use codex_app_server_protocol::ThreadRealtimeOutputAudioDeltaNotification; +use codex_app_server_protocol::ThreadRealtimeStartedNotification; +use codex_app_server_protocol::ThreadStartedNotification; +use codex_app_server_protocol::ThreadStatusChangedNotification; +use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification; +use codex_app_server_protocol::ToolRequestUserInputParams; +use codex_app_server_protocol::TurnCompletedNotification; +use codex_app_server_protocol::TurnDiffUpdatedNotification; +use codex_app_server_protocol::TurnPlanUpdatedNotification; +use codex_app_server_protocol::TurnStartedNotification; +use codex_protocol::protocol::SessionConfiguredEvent; + +#[derive(Debug, Clone)] +pub(crate) enum ThreadUpdate { + SessionConfigured(SessionConfiguredEvent), + ThreadStarted(ThreadStartedNotification), + ThreadStatusChanged(ThreadStatusChangedNotification), + ThreadClosed(ThreadClosedNotification), + ThreadNameUpdated(ThreadNameUpdatedNotification), + ThreadTokenUsageUpdated(ThreadTokenUsageUpdatedNotification), + TurnStarted(TurnStartedNotification), + TurnCompleted(TurnCompletedNotification), + TurnDiffUpdated(TurnDiffUpdatedNotification), + TurnPlanUpdated(TurnPlanUpdatedNotification), + ItemStarted(ItemStartedNotification), + ItemGuardianApprovalReviewStarted(ItemGuardianApprovalReviewStartedNotification), + ItemGuardianApprovalReviewCompleted(ItemGuardianApprovalReviewCompletedNotification), + ItemCompleted(ItemCompletedNotification), + AgentMessageDelta(AgentMessageDeltaNotification), + PlanDelta(PlanDeltaNotification), + ReasoningSummaryTextDelta(ReasoningSummaryTextDeltaNotification), + ReasoningSummaryPartAdded(ReasoningSummaryPartAddedNotification), + ReasoningTextDelta(ReasoningTextDeltaNotification), + TerminalInteraction(TerminalInteractionNotification), + CommandExecutionOutputDelta(CommandExecutionOutputDeltaNotification), + FileChangeOutputDelta(FileChangeOutputDeltaNotification), + McpToolCallProgress(McpToolCallProgressNotification), + HookStarted(HookStartedNotification), + HookCompleted(HookCompletedNotification), + Error(ErrorNotification), + ModelRerouted(ModelReroutedNotification), + DeprecationNotice(DeprecationNoticeNotification), + ThreadRealtimeStarted(ThreadRealtimeStartedNotification), + ThreadRealtimeItemAdded(ThreadRealtimeItemAddedNotification), + ThreadRealtimeOutputAudioDelta(ThreadRealtimeOutputAudioDeltaNotification), + ThreadRealtimeError(ThreadRealtimeErrorNotification), + ThreadRealtimeClosed(ThreadRealtimeClosedNotification), + CommandExecutionRequestApproval { + request_id: RequestId, + params: CommandExecutionRequestApprovalParams, + }, + FileChangeRequestApproval { + request_id: RequestId, + params: FileChangeRequestApprovalParams, + }, + McpServerElicitationRequest { + request_id: RequestId, + params: McpServerElicitationRequestParams, + }, + PermissionsRequestApproval { + request_id: RequestId, + params: PermissionsRequestApprovalParams, + }, + ToolRequestUserInput { + request_id: RequestId, + params: ToolRequestUserInputParams, + }, + DynamicToolCall { + request_id: RequestId, + params: DynamicToolCallParams, + }, + ThreadRolledBack { + num_turns: u32, + }, +} + +impl ThreadUpdate { + pub(crate) fn thread_id(&self) -> Option { + match self { + Self::SessionConfigured(session) => Some(session.session_id.to_string()), + Self::ThreadStarted(notification) => Some(notification.thread.id.clone()), + Self::ThreadStatusChanged(notification) => Some(notification.thread_id.clone()), + Self::ThreadClosed(notification) => Some(notification.thread_id.clone()), + Self::ThreadNameUpdated(notification) => Some(notification.thread_id.clone()), + Self::ThreadTokenUsageUpdated(notification) => Some(notification.thread_id.clone()), + Self::TurnStarted(notification) => Some(notification.thread_id.clone()), + Self::TurnCompleted(notification) => Some(notification.thread_id.clone()), + Self::TurnDiffUpdated(notification) => Some(notification.thread_id.clone()), + Self::TurnPlanUpdated(notification) => Some(notification.thread_id.clone()), + Self::ItemStarted(notification) => Some(notification.thread_id.clone()), + Self::ItemGuardianApprovalReviewStarted(notification) => { + Some(notification.thread_id.clone()) + } + Self::ItemGuardianApprovalReviewCompleted(notification) => { + Some(notification.thread_id.clone()) + } + Self::ItemCompleted(notification) => Some(notification.thread_id.clone()), + Self::AgentMessageDelta(notification) => Some(notification.thread_id.clone()), + Self::PlanDelta(notification) => Some(notification.thread_id.clone()), + Self::ReasoningSummaryTextDelta(notification) => Some(notification.thread_id.clone()), + Self::ReasoningSummaryPartAdded(notification) => Some(notification.thread_id.clone()), + Self::ReasoningTextDelta(notification) => Some(notification.thread_id.clone()), + Self::TerminalInteraction(notification) => Some(notification.thread_id.clone()), + Self::CommandExecutionOutputDelta(notification) => Some(notification.thread_id.clone()), + Self::FileChangeOutputDelta(notification) => Some(notification.thread_id.clone()), + Self::McpToolCallProgress(notification) => Some(notification.thread_id.clone()), + Self::HookStarted(notification) => Some(notification.thread_id.clone()), + Self::HookCompleted(notification) => Some(notification.thread_id.clone()), + Self::Error(notification) => Some(notification.thread_id.clone()), + Self::ModelRerouted(notification) => Some(notification.thread_id.clone()), + Self::ThreadRealtimeStarted(notification) => Some(notification.thread_id.clone()), + Self::ThreadRealtimeItemAdded(notification) => Some(notification.thread_id.clone()), + Self::ThreadRealtimeOutputAudioDelta(notification) => { + Some(notification.thread_id.clone()) + } + Self::ThreadRealtimeError(notification) => Some(notification.thread_id.clone()), + Self::ThreadRealtimeClosed(notification) => Some(notification.thread_id.clone()), + Self::CommandExecutionRequestApproval { params, .. } => Some(params.thread_id.clone()), + Self::FileChangeRequestApproval { params, .. } => Some(params.thread_id.clone()), + Self::McpServerElicitationRequest { params, .. } => Some(params.thread_id.clone()), + Self::PermissionsRequestApproval { params, .. } => Some(params.thread_id.clone()), + Self::ToolRequestUserInput { params, .. } => Some(params.thread_id.clone()), + Self::DynamicToolCall { params, .. } => Some(params.thread_id.clone()), + Self::DeprecationNotice(_) | Self::ThreadRolledBack { .. } => None, + } + } + + pub(crate) fn is_status_refresh_update(&self) -> bool { + matches!( + self, + Self::SessionConfigured(_) + | Self::TurnStarted(_) + | Self::ThreadTokenUsageUpdated(_) + | Self::TurnCompleted(_) + ) + } + + pub(crate) fn is_thread_closed(&self) -> bool { + matches!(self, Self::ThreadClosed(_)) + } +}