diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 225b2aface..57f389fb8b 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -697,7 +697,8 @@ pub(crate) async fn apply_bespoke_event_handling( .send_server_notification(ServerNotification::ItemCompleted(completed)) .await; } - EventMsg::EnteredReviewMode(review_request) => { + EventMsg::EnteredReviewMode(event) => { + let review_request = event.review_request; let review = review_request .user_facing_hint .unwrap_or_else(|| review_prompts::user_facing_hint(&review_request.target)); diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 1826e8d960..e7ff377e57 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -2699,14 +2699,11 @@ async fn spawn_review_thread( text_elements: Vec::new(), }]; let tc = Arc::new(review_turn_context); - sess.spawn_task(tc.clone(), input, ReviewTask::new()).await; - - // Announce entering review mode so UIs can switch modes. let review_request = ReviewRequest { target: resolved.target, user_facing_hint: Some(resolved.user_facing_hint), }; - sess.send_event(&tc, EventMsg::EnteredReviewMode(review_request)) + sess.spawn_task(tc.clone(), input, ReviewTask::new(review_request)) .await; } @@ -4374,7 +4371,13 @@ mod tests { text: "start review".to_string(), text_elements: Vec::new(), }]; - sess.spawn_task(Arc::clone(&tc), input, ReviewTask::new()) + let review_request = ReviewRequest { + target: codex_protocol::protocol::ReviewTarget::Custom { + instructions: "start review".to_string(), + }, + user_facing_hint: None, + }; + sess.spawn_task(Arc::clone(&tc), input, ReviewTask::new(review_request)) .await; sess.abort_all_tasks(TurnAbortReason::Interrupted).await; diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 555452a681..7730ca6ed2 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -6,12 +6,15 @@ use crate::review_format::format_review_findings_block; use crate::review_format::render_review_output_text; use crate::state::TaskKind; use async_trait::async_trait; +use codex_protocol::ThreadId; use codex_protocol::config_types::WebSearchMode; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::EnteredReviewModeEvent; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExitedReviewModeEvent; use codex_protocol::protocol::ReviewOutputEvent; +use codex_protocol::protocol::ReviewRequest; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::user_input::UserInput; @@ -22,12 +25,16 @@ use tokio_util::sync::CancellationToken; use super::SessionTask; use super::SessionTaskContext; -#[derive(Clone, Copy)] -pub(crate) struct ReviewTask; +#[derive(Clone)] +/// Task driver for review-mode turns. +pub(crate) struct ReviewTask { + review_request: ReviewRequest, +} impl ReviewTask { - pub(crate) fn new() -> Self { - Self + /// Creates a new review task with the already-resolved request details. + pub(crate) fn new(review_request: ReviewRequest) -> Self { + Self { review_request } } } @@ -50,13 +57,93 @@ impl SessionTask for ReviewTask { .otel_manager .counter("codex.task.review", 1, &[]); - // Start sub-codex conversation and await its final status. - let output = match start_review_conversation(session.clone(), ctx.clone(), input).await { - Some(sub_agent) => { - wait_for_review_output(sub_agent.status_rx, cancellation_token.clone()).await - } - None => None, + // Build review agent config. + let sub_agent_config = { + let config = ctx.client.config(); + let mut sub_agent_config = config.as_ref().clone(); + sub_agent_config.web_search_mode = Some(WebSearchMode::Disabled); + sub_agent_config.base_instructions = Some(crate::REVIEW_PROMPT.to_string()); + let model = config + .review_model + .clone() + .unwrap_or_else(|| ctx.client.get_model()); + sub_agent_config.model = Some(model); + sub_agent_config }; + + // Spawn the review agent. + let agent_control = session.agent_control(); + let thread_id = agent_control + .spawn_agent( + sub_agent_config, + None, + Some(SessionSource::SubAgent(SubAgentSource::Review)), + ) + .await + .ok()?; + + let mut status_rx = agent_control.subscribe_status(thread_id).await.ok()?; + + session + .session + .send_event( + ctx.as_ref(), + EventMsg::EnteredReviewMode(EnteredReviewModeEvent { + review_request: self.review_request.clone(), + review_thread_id: thread_id, + }), + ) + .await; + + // Send input to the agent. + let output = match session + .agent_control() + .send_input(thread_id, input) + .await { + Ok(_) => { + let mut status = status_rx.borrow().clone(); + let parse_output = |message: String| { + if let Ok(ev) = serde_json::from_str::(&message) { + return ev; + } + if let (Some(start), Some(end)) = (message.find('{'), message.rfind('}')) + && start < end + && let Some(slice) = message.get(start..=end) + && let Ok(ev) = serde_json::from_str::(slice) + { + return ev; + } + ReviewOutputEvent { + overall_explanation: message, + ..Default::default() + } + }; + loop { + if is_final(&status) { + break match status { + AgentStatus::Completed(Some(message)) => { + Some(parse_output(message)) + } + _ => None, + }; + } + tokio::select! { + _ = cancellation_token.cancelled() => break None, + changed = status_rx.changed() => { + if changed.is_err() { + break None; + } + status = status_rx.borrow().clone(); + } + } + } + }, + Err(err) => { + tracing::error!("Error while sending input to review agent {err:?}"); + None + }, + }; + if !cancellation_token.is_cancelled() { exit_review_mode(session.clone_session(), output.clone(), ctx.clone()).await; } @@ -68,94 +155,6 @@ impl SessionTask for ReviewTask { } } -async fn start_review_conversation( - session: Arc, - ctx: Arc, - input: Vec, -) -> Option { - let config = ctx.client.config(); - let mut sub_agent_config = config.as_ref().clone(); - // Carry over review-only feature restrictions so the sub-agent cannot - // re-enable blocked tools (web search, view image). - sub_agent_config.web_search_mode = Some(WebSearchMode::Disabled); - - // Set explicit review rubric for the sub-agent - sub_agent_config.base_instructions = Some(crate::REVIEW_PROMPT.to_string()); - - let model = config - .review_model - .clone() - .unwrap_or_else(|| ctx.client.get_model()); - sub_agent_config.model = Some(model); - - let agent_control = session.agent_control(); - let thread_id = agent_control - .spawn_agent( - sub_agent_config, - None, - Some(SessionSource::SubAgent(SubAgentSource::Review)), - ) - .await - .ok()?; - - agent_control.send_input(thread_id, input).await.ok()?; - - let status_rx = agent_control.subscribe_status(thread_id).await.ok()?; - Some(ReviewSubAgent { status_rx }) -} - -struct ReviewSubAgent { - status_rx: watch::Receiver, -} - -async fn wait_for_review_output( - mut status_rx: watch::Receiver, - cancellation_token: CancellationToken, -) -> Option { - let mut status = status_rx.borrow().clone(); - while !is_final(&status) { - tokio::select! { - _ = cancellation_token.cancelled() => return None, - changed = status_rx.changed() => { - if changed.is_err() { - break; - } - status = status_rx.borrow().clone(); - } - } - } - review_output_from_status(status) -} - -fn review_output_from_status(status: AgentStatus) -> Option { - match status { - AgentStatus::Completed(Some(message)) => Some(parse_review_output_event(&message)), - _ => None, - } -} - -/// Parse a ReviewOutputEvent from a text blob returned by the reviewer model. -/// If the text is valid JSON matching ReviewOutputEvent, deserialize it. -/// Otherwise, attempt to extract the first JSON object substring and parse it. -/// If parsing still fails, return a structured fallback carrying the plain text -/// in `overall_explanation`. -fn parse_review_output_event(text: &str) -> ReviewOutputEvent { - if let Ok(ev) = serde_json::from_str::(text) { - return ev; - } - if let (Some(start), Some(end)) = (text.find('{'), text.rfind('}')) - && start < end - && let Some(slice) = text.get(start..=end) - && let Ok(ev) = serde_json::from_str::(slice) - { - return ev; - } - ReviewOutputEvent { - overall_explanation: text.to_string(), - ..Default::default() - } -} - /// Emits an ExitedReviewMode Event with optional ReviewOutput, /// and records a developer message with the review output. pub(crate) async fn exit_review_mode( diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 600707404e..5d4e4ddc13 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -791,7 +791,7 @@ pub enum EventMsg { ShutdownComplete, /// Entered review mode. - EnteredReviewMode(ReviewRequest), + EnteredReviewMode(EnteredReviewModeEvent), /// Exited review mode with an optional final result to apply. ExitedReviewMode(ExitedReviewModeEvent), @@ -1040,6 +1040,12 @@ pub struct ExitedReviewModeEvent { pub review_output: Option, } +#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] +pub struct EnteredReviewModeEvent { + pub review_request: ReviewRequest, + pub review_thread_id: ThreadId, +} + // Individual event payload types matching each `EventMsg` variant. #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index d20b6e1a40..bd02fc156b 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -27,6 +27,7 @@ use crate::resume_picker::SessionSelection; use crate::tui; use crate::tui::TuiEvent; use crate::update_action::UpdateAction; +use crate::visual_event_filter::should_forward_visual_event; use codex_ansi_escape::ansi_escape_line; use codex_core::AuthManager; use codex_core::ThreadManager; @@ -62,6 +63,7 @@ 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; @@ -362,6 +364,8 @@ pub(crate) struct App { external_approval_routes: HashMap, /// Buffered Codex events while external approvals are pending. paused_codex_events: VecDeque, + + review_thread_ids: HashSet, } impl App { @@ -386,6 +390,7 @@ impl App { } async fn shutdown_current_thread(&mut self) { + self.review_thread_ids.clear(); if let Some(thread_id) = self.chat_widget.thread_id() { // Clear any in-flight rollback guard when switching threads. self.backtrack.pending_rollback = None; @@ -548,6 +553,7 @@ impl App { skip_world_writable_scan_once: false, external_approval_routes: HashMap::new(), paused_codex_events: VecDeque::new(), + review_thread_ids: HashSet::new(), }; // On startup, if Agent mode (workspace-write) or ReadOnly is active, warn about world-writable dirs on Windows. @@ -899,8 +905,26 @@ impl App { tui.frame_requester().schedule_frame(); } } - AppEvent::ExternalApprovalRequest { thread_id, event } => { - self.handle_external_approval_request(thread_id, event); + AppEvent::ExternalThreadEvent { thread_id, event } => { + if matches!( + event.msg, + EventMsg::ExecApprovalRequest(_) | EventMsg::ApplyPatchApprovalRequest(_) + ) { + self.handle_external_approval_request(thread_id, event); + return Ok(AppRunControl::Continue); + } + if self.review_thread_ids.contains(&thread_id) + && should_forward_visual_event(&event.msg) + { + if !self.external_approval_routes.is_empty() { + self.paused_codex_events.push_back(event); + return Ok(AppRunControl::Continue); + } + self.handle_codex_event_now(event); + if self.backtrack_render_pending { + tui.frame_requester().schedule_frame(); + } + } } AppEvent::Exit(mode) => match mode { ExitMode::ShutdownFirst => self.chat_widget.submit_op(Op::Shutdown), @@ -1439,6 +1463,15 @@ impl App { self.suppress_shutdown_complete = false; return; } + match &event.msg { + EventMsg::EnteredReviewMode(ev) => { + self.review_thread_ids.insert(ev.review_thread_id); + } + EventMsg::ExitedReviewMode(_) => { + self.review_thread_ids.clear(); + } + _ => {} + } if let EventMsg::ListSkillsResponse(response) = &event.msg { let cwd = self.chat_widget.config_ref().cwd.clone(); let errors = errors_for_cwd(&cwd, response); @@ -1507,12 +1540,7 @@ impl App { break; } }; - match event.msg { - EventMsg::ExecApprovalRequest(_) | EventMsg::ApplyPatchApprovalRequest(_) => { - app_event_tx.send(AppEvent::ExternalApprovalRequest { thread_id, event }); - } - _ => {} - } + app_event_tx.send(AppEvent::ExternalThreadEvent { thread_id, event }); } }); Ok(()) @@ -1776,6 +1804,7 @@ mod tests { skip_world_writable_scan_once: false, external_approval_routes: HashMap::new(), paused_codex_events: VecDeque::new(), + review_thread_ids: HashSet::new(), } } @@ -1819,6 +1848,7 @@ mod tests { skip_world_writable_scan_once: false, external_approval_routes: HashMap::new(), paused_codex_events: VecDeque::new(), + review_thread_ids: HashSet::new(), }, rx, op_rx, diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index f34e1d7d5a..33fba8c564 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -42,7 +42,7 @@ pub(crate) enum WindowsSandboxFallbackReason { #[derive(Debug)] pub(crate) enum AppEvent { CodexEvent(Event), - ExternalApprovalRequest { + ExternalThreadEvent { thread_id: ThreadId, event: Event, }, diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 3f610c4faa..0009c65745 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -2686,8 +2686,8 @@ impl ChatWidget { self.on_user_message_event(ev); } } - EventMsg::EnteredReviewMode(review_request) => { - self.on_entered_review_mode(review_request, from_replay) + EventMsg::EnteredReviewMode(event) => { + self.on_entered_review_mode(event.review_request, from_replay) } EventMsg::ExitedReviewMode(review) => self.on_exited_review_mode(review), EventMsg::ContextCompacted(_) => self.on_agent_message("Context compacted".to_owned()), diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index b864a7b3ee..f29fad5461 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -30,6 +30,7 @@ use codex_core::protocol::AgentReasoningEvent; use codex_core::protocol::ApplyPatchApprovalRequestEvent; use codex_core::protocol::BackgroundEventEvent; use codex_core::protocol::CreditsSnapshot; +use codex_core::protocol::EnteredReviewModeEvent; use codex_core::protocol::Event; use codex_core::protocol::EventMsg; use codex_core::protocol::ExecApprovalRequestEvent; @@ -551,11 +552,14 @@ async fn entered_review_mode_uses_request_hint() { chat.handle_codex_event(Event { id: "review-start".into(), - msg: EventMsg::EnteredReviewMode(ReviewRequest { - target: ReviewTarget::BaseBranch { - branch: "feature".to_string(), + msg: EventMsg::EnteredReviewMode(EnteredReviewModeEvent { + review_request: ReviewRequest { + target: ReviewTarget::BaseBranch { + branch: "feature".to_string(), + }, + user_facing_hint: Some("feature branch".to_string()), }, - user_facing_hint: Some("feature branch".to_string()), + review_thread_id: ThreadId::new(), }), }); @@ -572,9 +576,12 @@ async fn entered_review_mode_defaults_to_current_changes_banner() { chat.handle_codex_event(Event { id: "review-start".into(), - msg: EventMsg::EnteredReviewMode(ReviewRequest { - target: ReviewTarget::UncommittedChanges, - user_facing_hint: None, + msg: EventMsg::EnteredReviewMode(EnteredReviewModeEvent { + review_request: ReviewRequest { + target: ReviewTarget::UncommittedChanges, + user_facing_hint: None, + }, + review_thread_id: ThreadId::new(), }), }); @@ -604,11 +611,14 @@ async fn review_restores_context_window_indicator() { chat.handle_codex_event(Event { id: "review-start".into(), - msg: EventMsg::EnteredReviewMode(ReviewRequest { - target: ReviewTarget::BaseBranch { - branch: "feature".to_string(), + msg: EventMsg::EnteredReviewMode(EnteredReviewModeEvent { + review_request: ReviewRequest { + target: ReviewTarget::BaseBranch { + branch: "feature".to_string(), + }, + user_facing_hint: Some("feature branch".to_string()), }, - user_facing_hint: Some("feature branch".to_string()), + review_thread_id: ThreadId::new(), }), }); @@ -4431,9 +4441,12 @@ async fn review_queues_user_messages_snapshot() { chat.handle_codex_event(Event { id: "review-1".into(), - msg: EventMsg::EnteredReviewMode(ReviewRequest { - target: ReviewTarget::UncommittedChanges, - user_facing_hint: Some("current changes".to_string()), + msg: EventMsg::EnteredReviewMode(EnteredReviewModeEvent { + review_request: ReviewRequest { + target: ReviewTarget::UncommittedChanges, + user_facing_hint: Some("current changes".to_string()), + }, + review_thread_id: ThreadId::new(), }), }); let _ = drain_insert_history(&mut rx); diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index e57497505a..37abd390db 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -89,6 +89,7 @@ pub mod update_action; mod update_prompt; mod updates; mod version; +mod visual_event_filter; mod wrapping; diff --git a/codex-rs/tui/src/visual_event_filter.rs b/codex-rs/tui/src/visual_event_filter.rs new file mode 100644 index 0000000000..69b611a628 --- /dev/null +++ b/codex-rs/tui/src/visual_event_filter.rs @@ -0,0 +1,43 @@ +use codex_core::protocol::EventMsg; + +/// Returns true when a protocol event should be forwarded into the visual transcript. +pub(crate) fn should_forward_visual_event(msg: &EventMsg) -> bool { + matches!( + msg, + EventMsg::AgentMessage(_) + | EventMsg::AgentMessageDelta(_) + | EventMsg::AgentMessageContentDelta(_) + | EventMsg::AgentReasoning(_) + | EventMsg::AgentReasoningDelta(_) + | EventMsg::AgentReasoningRawContent(_) + | EventMsg::AgentReasoningRawContentDelta(_) + | EventMsg::AgentReasoningSectionBreak(_) + | EventMsg::ReasoningContentDelta(_) + | EventMsg::ReasoningRawContentDelta(_) + | EventMsg::TurnStarted(_) + | EventMsg::TurnComplete(_) + | EventMsg::TurnAborted(_) + | EventMsg::TokenCount(_) + | EventMsg::Warning(_) + | EventMsg::Error(_) + | EventMsg::PlanUpdate(_) + | EventMsg::ExecCommandBegin(_) + | EventMsg::ExecCommandOutputDelta(_) + | EventMsg::TerminalInteraction(_) + | EventMsg::ExecCommandEnd(_) + | EventMsg::PatchApplyBegin(_) + | EventMsg::PatchApplyEnd(_) + | EventMsg::ViewImageToolCall(_) + | EventMsg::McpToolCallBegin(_) + | EventMsg::McpToolCallEnd(_) + | EventMsg::WebSearchBegin(_) + | EventMsg::WebSearchEnd(_) + | EventMsg::BackgroundEvent(_) + | EventMsg::StreamError(_) + | EventMsg::TurnDiff(_) + | EventMsg::UndoStarted(_) + | EventMsg::UndoCompleted(_) + | EventMsg::ContextCompacted(_) + | EventMsg::DeprecationNotice(_) + ) +} diff --git a/codex-rs/tui2/src/chatwidget.rs b/codex-rs/tui2/src/chatwidget.rs index 6c11077c80..84e264b9ba 100644 --- a/codex-rs/tui2/src/chatwidget.rs +++ b/codex-rs/tui2/src/chatwidget.rs @@ -2457,9 +2457,7 @@ impl ChatWidget { self.on_user_message_event(ev); } } - EventMsg::EnteredReviewMode(review_request) => { - self.on_entered_review_mode(review_request) - } + EventMsg::EnteredReviewMode(event) => self.on_entered_review_mode(event.review_request), EventMsg::ExitedReviewMode(review) => self.on_exited_review_mode(review), EventMsg::ContextCompacted(_) => self.on_agent_message("Context compacted".to_owned()), EventMsg::CollabAgentSpawnBegin(_) => {} diff --git a/codex-rs/tui2/src/chatwidget/tests.rs b/codex-rs/tui2/src/chatwidget/tests.rs index 9d5d7e67b2..d87cffa109 100644 --- a/codex-rs/tui2/src/chatwidget/tests.rs +++ b/codex-rs/tui2/src/chatwidget/tests.rs @@ -28,6 +28,7 @@ use codex_core::protocol::AgentReasoningEvent; use codex_core::protocol::ApplyPatchApprovalRequestEvent; use codex_core::protocol::BackgroundEventEvent; use codex_core::protocol::CreditsSnapshot; +use codex_core::protocol::EnteredReviewModeEvent; use codex_core::protocol::Event; use codex_core::protocol::EventMsg; use codex_core::protocol::ExecApprovalRequestEvent; @@ -539,11 +540,14 @@ async fn entered_review_mode_uses_request_hint() { chat.handle_codex_event(Event { id: "review-start".into(), - msg: EventMsg::EnteredReviewMode(ReviewRequest { - target: ReviewTarget::BaseBranch { - branch: "feature".to_string(), + msg: EventMsg::EnteredReviewMode(EnteredReviewModeEvent { + review_request: ReviewRequest { + target: ReviewTarget::BaseBranch { + branch: "feature".to_string(), + }, + user_facing_hint: Some("feature branch".to_string()), }, - user_facing_hint: Some("feature branch".to_string()), + review_thread_id: ThreadId::new(), }), }); @@ -560,9 +564,12 @@ async fn entered_review_mode_defaults_to_current_changes_banner() { chat.handle_codex_event(Event { id: "review-start".into(), - msg: EventMsg::EnteredReviewMode(ReviewRequest { - target: ReviewTarget::UncommittedChanges, - user_facing_hint: None, + msg: EventMsg::EnteredReviewMode(EnteredReviewModeEvent { + review_request: ReviewRequest { + target: ReviewTarget::UncommittedChanges, + user_facing_hint: None, + }, + review_thread_id: ThreadId::new(), }), }); @@ -592,11 +599,14 @@ async fn review_restores_context_window_indicator() { chat.handle_codex_event(Event { id: "review-start".into(), - msg: EventMsg::EnteredReviewMode(ReviewRequest { - target: ReviewTarget::BaseBranch { - branch: "feature".to_string(), + msg: EventMsg::EnteredReviewMode(EnteredReviewModeEvent { + review_request: ReviewRequest { + target: ReviewTarget::BaseBranch { + branch: "feature".to_string(), + }, + user_facing_hint: Some("feature branch".to_string()), }, - user_facing_hint: Some("feature branch".to_string()), + review_thread_id: ThreadId::new(), }), });