diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 9c4f1b871f..c4dd63ce03 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -273,8 +273,6 @@ use crate::tui::FrameRequester; mod interrupts; use self::interrupts::InterruptManager; mod agent; -use self::agent::spawn_agent; -use self::agent::spawn_agent_from_existing; pub(crate) use self::agent::spawn_app_event_forwarder; mod session_header; use self::session_header::SessionHeader; @@ -295,7 +293,6 @@ use crate::streaming::controller::StreamController; use chrono::Local; use codex_app_server_protocol::Account; -use codex_core::ThreadManager; use codex_file_search::FileMatch; use codex_protocol::openai_models::InputModality; use codex_protocol::openai_models::ModelPreset; @@ -3535,195 +3532,6 @@ impl ChatWidget { self.had_work_activity = true; } - pub(crate) fn new(common: ChatWidgetInit, thread_manager: Arc) -> Self { - let ChatWidgetInit { - config, - frame_requester, - app_event_tx, - initial_user_message, - enhanced_keys_supported, - account, - available_models, - available_collaboration_modes, - feedback, - is_first_run, - feedback_audience, - model, - startup_tooltip_override, - status_line_invalid_items_warned, - session_telemetry, - } = common; - let model = model.filter(|m| !m.trim().is_empty()); - let mut config = config; - config.model = model.clone(); - let prevent_idle_sleep = config.features.enabled(Feature::PreventIdleSleep); - let mut rng = rand::rng(); - let placeholder = PLACEHOLDERS[rng.random_range(0..PLACEHOLDERS.len())].to_string(); - let codex_op_tx = spawn_agent(config.clone(), app_event_tx.clone(), thread_manager); - - let model_override = model.as_deref(); - let model_for_header = model - .clone() - .unwrap_or_else(|| DEFAULT_MODEL_DISPLAY_NAME.to_string()); - let active_collaboration_mask = Self::initial_collaboration_mask( - &config, - &available_collaboration_modes, - model_override, - ); - let header_model = active_collaboration_mask - .as_ref() - .and_then(|mask| mask.model.clone()) - .unwrap_or_else(|| model_for_header.clone()); - let fallback_default = Settings { - model: header_model.clone(), - reasoning_effort: None, - developer_instructions: None, - }; - // Collaboration modes start in Default mode. - let current_collaboration_mode = CollaborationMode { - mode: ModeKind::Default, - settings: fallback_default, - }; - - let active_cell = Some(Self::placeholder_session_header_cell(&config)); - - let current_cwd = Some(config.cwd.clone()); - let queued_message_edit_binding = - queued_message_edit_binding_for_terminal(terminal_info().name); - let mut widget = Self { - app_event_tx: app_event_tx.clone(), - frame_requester: frame_requester.clone(), - codex_op_tx, - bottom_pane: BottomPane::new(BottomPaneParams { - frame_requester, - app_event_tx, - has_input_focus: true, - enhanced_keys_supported, - placeholder_text: placeholder, - disable_paste_burst: config.disable_paste_burst, - animations_enabled: config.animations, - skills: None, - }), - active_cell, - active_cell_revision: 0, - config, - skills_all: Vec::new(), - skills_initial_state: None, - current_collaboration_mode, - active_collaboration_mask, - account, - available_models, - available_collaboration_modes, - session_telemetry, - session_header: SessionHeader::new(header_model), - initial_user_message, - token_info: None, - rate_limit_snapshots_by_limit_id: BTreeMap::new(), - plan_type: None, - rate_limit_warnings: RateLimitWarningState::default(), - rate_limit_switch_prompt: RateLimitSwitchPromptState::default(), - rate_limit_poller: None, - adaptive_chunking: AdaptiveChunkingPolicy::default(), - stream_controller: None, - plan_stream_controller: None, - last_copyable_output: None, - running_commands: HashMap::new(), - pending_collab_spawn_requests: HashMap::new(), - suppressed_exec_calls: HashSet::new(), - last_unified_wait: None, - unified_exec_wait_streak: None, - turn_sleep_inhibitor: SleepInhibitor::new(prevent_idle_sleep), - task_complete_pending: false, - unified_exec_processes: Vec::new(), - agent_turn_running: false, - mcp_startup_status: None, - connectors_cache: ConnectorsCacheState::default(), - connectors_partial_snapshot: None, - connectors_prefetch_in_flight: false, - connectors_force_refetch_pending: false, - interrupts: InterruptManager::new(), - reasoning_buffer: String::new(), - full_reasoning_buffer: String::new(), - current_status: StatusIndicatorState::working(), - pending_guardian_review_status: PendingGuardianReviewStatus::default(), - retry_status_header: None, - pending_status_indicator_restore: false, - suppress_queue_autosend: false, - thread_id: None, - thread_name: None, - forked_from: None, - queued_user_messages: VecDeque::new(), - pending_steers: VecDeque::new(), - submit_pending_steers_after_interrupt: false, - queued_message_edit_binding, - show_welcome_banner: is_first_run, - startup_tooltip_override, - suppress_session_configured_redraw: false, - pending_notification: None, - quit_shortcut_expires_at: None, - quit_shortcut_key: None, - is_review_mode: false, - pre_review_token_info: None, - needs_final_message_separator: false, - had_work_activity: false, - saw_plan_update_this_turn: false, - saw_plan_item_this_turn: false, - plan_delta_buffer: String::new(), - plan_item_active: false, - last_separator_elapsed_secs: None, - turn_runtime_metrics: RuntimeMetricsSummary::default(), - last_rendered_width: std::cell::Cell::new(None), - feedback, - feedback_audience, - current_rollout_path: None, - current_cwd, - session_network_proxy: None, - status_line_invalid_items_warned, - status_line_branch: None, - status_line_branch_cwd: None, - status_line_branch_pending: false, - status_line_branch_lookup_complete: false, - external_editor_state: ExternalEditorState::Closed, - realtime_conversation: RealtimeConversationUiState::default(), - last_rendered_user_message_event: None, - }; - - widget.prefetch_rate_limits(); - widget.bottom_pane.set_voice_transcription_enabled( - widget.config.features.enabled(Feature::VoiceTranscription), - ); - widget - .bottom_pane - .set_realtime_conversation_enabled(widget.realtime_conversation_enabled()); - widget - .bottom_pane - .set_audio_device_selection_enabled(widget.realtime_audio_device_selection_enabled()); - widget - .bottom_pane - .set_status_line_enabled(!widget.configured_status_line_items().is_empty()); - widget.bottom_pane.set_collaboration_modes_enabled(true); - widget.sync_fast_command_enabled(); - widget.sync_personality_command_enabled(); - widget - .bottom_pane - .set_queued_message_edit_binding(widget.queued_message_edit_binding); - #[cfg(target_os = "windows")] - widget.bottom_pane.set_windows_degraded_sandbox_active( - codex_core::windows_sandbox::ELEVATED_SANDBOX_NUX_ENABLED - && matches!( - WindowsSandboxLevel::from_config(&widget.config), - WindowsSandboxLevel::RestrictedToken - ), - ); - widget.update_collaboration_mode_indicator(); - - widget - .bottom_pane - .set_connectors_enabled(widget.connectors_enabled()); - - widget - } - pub(crate) fn new_with_op_sender( common: ChatWidgetInit, codex_op_tx: UnboundedSender, @@ -3770,7 +3578,6 @@ impl ChatWidget { reasoning_effort: None, developer_instructions: None, }; - // Collaboration modes start in Default mode. let current_collaboration_mode = CollaborationMode { mode: ModeKind::Default, settings: fallback_default, @@ -3905,198 +3712,6 @@ impl ChatWidget { widget } - /// Create a ChatWidget attached to an existing conversation (e.g., a fork). - pub(crate) fn new_from_existing( - common: ChatWidgetInit, - conversation: std::sync::Arc, - session_configured: codex_protocol::protocol::SessionConfiguredEvent, - ) -> Self { - let ChatWidgetInit { - config, - frame_requester, - app_event_tx, - initial_user_message, - enhanced_keys_supported, - account, - available_models, - available_collaboration_modes, - feedback, - is_first_run: _, - feedback_audience, - model, - startup_tooltip_override: _, - status_line_invalid_items_warned, - session_telemetry, - } = common; - let model = model.filter(|m| !m.trim().is_empty()); - let prevent_idle_sleep = config.features.enabled(Feature::PreventIdleSleep); - let mut rng = rand::rng(); - let placeholder = PLACEHOLDERS[rng.random_range(0..PLACEHOLDERS.len())].to_string(); - - let model_override = model.as_deref(); - let header_model = model - .clone() - .unwrap_or_else(|| session_configured.model.clone()); - let active_collaboration_mask = Self::initial_collaboration_mask( - &config, - &available_collaboration_modes, - model_override, - ); - let header_model = active_collaboration_mask - .as_ref() - .and_then(|mask| mask.model.clone()) - .unwrap_or(header_model); - - let current_cwd = Some(session_configured.cwd.clone()); - let codex_op_tx = - spawn_agent_from_existing(conversation, session_configured, app_event_tx.clone()); - - let fallback_default = Settings { - model: header_model.clone(), - reasoning_effort: None, - developer_instructions: None, - }; - // Collaboration modes start in Default mode. - let current_collaboration_mode = CollaborationMode { - mode: ModeKind::Default, - settings: fallback_default, - }; - - let queued_message_edit_binding = - queued_message_edit_binding_for_terminal(terminal_info().name); - let mut widget = Self { - app_event_tx: app_event_tx.clone(), - frame_requester: frame_requester.clone(), - codex_op_tx, - bottom_pane: BottomPane::new(BottomPaneParams { - frame_requester, - app_event_tx, - has_input_focus: true, - enhanced_keys_supported, - placeholder_text: placeholder, - disable_paste_burst: config.disable_paste_burst, - animations_enabled: config.animations, - skills: None, - }), - active_cell: None, - active_cell_revision: 0, - config, - skills_all: Vec::new(), - skills_initial_state: None, - current_collaboration_mode, - active_collaboration_mask, - account, - available_models, - available_collaboration_modes, - session_telemetry, - session_header: SessionHeader::new(header_model), - initial_user_message, - token_info: None, - rate_limit_snapshots_by_limit_id: BTreeMap::new(), - plan_type: None, - rate_limit_warnings: RateLimitWarningState::default(), - rate_limit_switch_prompt: RateLimitSwitchPromptState::default(), - rate_limit_poller: None, - adaptive_chunking: AdaptiveChunkingPolicy::default(), - stream_controller: None, - plan_stream_controller: None, - last_copyable_output: None, - running_commands: HashMap::new(), - pending_collab_spawn_requests: HashMap::new(), - suppressed_exec_calls: HashSet::new(), - last_unified_wait: None, - unified_exec_wait_streak: None, - turn_sleep_inhibitor: SleepInhibitor::new(prevent_idle_sleep), - task_complete_pending: false, - unified_exec_processes: Vec::new(), - agent_turn_running: false, - mcp_startup_status: None, - connectors_cache: ConnectorsCacheState::default(), - connectors_partial_snapshot: None, - connectors_prefetch_in_flight: false, - connectors_force_refetch_pending: false, - interrupts: InterruptManager::new(), - reasoning_buffer: String::new(), - full_reasoning_buffer: String::new(), - current_status: StatusIndicatorState::working(), - pending_guardian_review_status: PendingGuardianReviewStatus::default(), - retry_status_header: None, - pending_status_indicator_restore: false, - suppress_queue_autosend: false, - thread_id: None, - thread_name: None, - forked_from: None, - queued_user_messages: VecDeque::new(), - pending_steers: VecDeque::new(), - submit_pending_steers_after_interrupt: false, - queued_message_edit_binding, - show_welcome_banner: false, - startup_tooltip_override: None, - suppress_session_configured_redraw: true, - pending_notification: None, - quit_shortcut_expires_at: None, - quit_shortcut_key: None, - is_review_mode: false, - pre_review_token_info: None, - needs_final_message_separator: false, - had_work_activity: false, - saw_plan_update_this_turn: false, - saw_plan_item_this_turn: false, - plan_delta_buffer: String::new(), - plan_item_active: false, - last_separator_elapsed_secs: None, - turn_runtime_metrics: RuntimeMetricsSummary::default(), - last_rendered_width: std::cell::Cell::new(None), - feedback, - feedback_audience, - current_rollout_path: None, - current_cwd, - session_network_proxy: None, - status_line_invalid_items_warned, - status_line_branch: None, - status_line_branch_cwd: None, - status_line_branch_pending: false, - status_line_branch_lookup_complete: false, - external_editor_state: ExternalEditorState::Closed, - realtime_conversation: RealtimeConversationUiState::default(), - last_rendered_user_message_event: None, - }; - - widget.prefetch_rate_limits(); - widget.bottom_pane.set_voice_transcription_enabled( - widget.config.features.enabled(Feature::VoiceTranscription), - ); - widget - .bottom_pane - .set_realtime_conversation_enabled(widget.realtime_conversation_enabled()); - widget - .bottom_pane - .set_audio_device_selection_enabled(widget.realtime_audio_device_selection_enabled()); - widget - .bottom_pane - .set_status_line_enabled(!widget.configured_status_line_items().is_empty()); - widget.bottom_pane.set_collaboration_modes_enabled(true); - widget.sync_fast_command_enabled(); - widget.sync_personality_command_enabled(); - widget - .bottom_pane - .set_queued_message_edit_binding(widget.queued_message_edit_binding); - #[cfg(target_os = "windows")] - widget.bottom_pane.set_windows_degraded_sandbox_active( - codex_core::windows_sandbox::ELEVATED_SANDBOX_NUX_ENABLED - && matches!( - WindowsSandboxLevel::from_config(&widget.config), - WindowsSandboxLevel::RestrictedToken - ), - ); - widget.update_collaboration_mode_indicator(); - widget - .bottom_pane - .set_connectors_enabled(widget.connectors_enabled()); - - widget - } - pub(crate) fn handle_key_event(&mut self, key_event: KeyEvent) { match key_event { KeyEvent { diff --git a/codex-rs/tui/src/chatwidget/agent.rs b/codex-rs/tui/src/chatwidget/agent.rs index 385bc41091..5178209f74 100644 --- a/codex-rs/tui/src/chatwidget/agent.rs +++ b/codex-rs/tui/src/chatwidget/agent.rs @@ -1,11 +1,3 @@ -use std::sync::Arc; - -use codex_core::CodexThread; -use codex_core::NewThread; -use codex_core::ThreadManager; -use codex_core::config::Config; -use codex_protocol::protocol::Event; -use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::Op; use tokio::sync::mpsc::UnboundedSender; use tokio::sync::mpsc::unbounded_channel; @@ -13,141 +5,6 @@ use tokio::sync::mpsc::unbounded_channel; use crate::app_event::AppEvent; use crate::app_event_sender::AppEventSender; -const TUI_NOTIFY_CLIENT: &str = "codex-tui"; - -async fn initialize_app_server_client_name(thread: &CodexThread) { - if let Err(err) = thread - .set_app_server_client_name(Some(TUI_NOTIFY_CLIENT.to_string())) - .await - { - tracing::error!("failed to set app server client name: {err}"); - } -} - -/// Spawn the agent bootstrapper and op forwarding loop, returning the -/// `UnboundedSender` used by the UI to submit operations. -pub(crate) fn spawn_agent( - config: Config, - app_event_tx: AppEventSender, - server: Arc, -) -> UnboundedSender { - let (codex_op_tx, mut codex_op_rx) = unbounded_channel::(); - - let app_event_tx_clone = app_event_tx; - tokio::spawn(async move { - let NewThread { - thread, - session_configured, - .. - } = match server.start_thread(config).await { - Ok(v) => v, - Err(err) => { - let message = format!("Failed to initialize codex: {err}"); - tracing::error!("{message}"); - app_event_tx_clone.send(AppEvent::CodexEvent(Event { - id: "".to_string(), - msg: EventMsg::Error(err.to_error_event(None)), - })); - app_event_tx_clone.send(AppEvent::FatalExitRequest(message)); - tracing::error!("failed to initialize codex: {err}"); - return; - } - }; - initialize_app_server_client_name(thread.as_ref()).await; - - // Forward the captured `SessionConfigured` event so it can be rendered in the UI. - let ev = codex_protocol::protocol::Event { - // The `id` does not matter for rendering, so we can use a fake value. - id: "".to_string(), - msg: codex_protocol::protocol::EventMsg::SessionConfigured(session_configured), - }; - app_event_tx_clone.send(AppEvent::CodexEvent(ev)); - - let thread_clone = thread.clone(); - tokio::spawn(async move { - while let Some(op) = codex_op_rx.recv().await { - let id = thread_clone.submit(op).await; - if let Err(e) = id { - tracing::error!("failed to submit op: {e}"); - } - } - }); - - while let Ok(event) = thread.next_event().await { - let is_shutdown_complete = matches!(event.msg, EventMsg::ShutdownComplete); - app_event_tx_clone.send(AppEvent::CodexEvent(event)); - if is_shutdown_complete { - // ShutdownComplete is terminal for a thread; drop this receiver task so - // the Arc can be released and thread resources can clean up. - break; - } - } - }); - - codex_op_tx -} - -/// Spawn agent loops for an existing thread (e.g., a forked thread). -/// Sends the provided `SessionConfiguredEvent` immediately, then forwards subsequent -/// events and accepts Ops for submission. -pub(crate) fn spawn_agent_from_existing( - thread: std::sync::Arc, - session_configured: codex_protocol::protocol::SessionConfiguredEvent, - app_event_tx: AppEventSender, -) -> UnboundedSender { - let (codex_op_tx, mut codex_op_rx) = unbounded_channel::(); - - let app_event_tx_clone = app_event_tx; - tokio::spawn(async move { - initialize_app_server_client_name(thread.as_ref()).await; - - // Forward the captured `SessionConfigured` event so it can be rendered in the UI. - let ev = codex_protocol::protocol::Event { - id: "".to_string(), - msg: codex_protocol::protocol::EventMsg::SessionConfigured(session_configured), - }; - app_event_tx_clone.send(AppEvent::CodexEvent(ev)); - - let thread_clone = thread.clone(); - tokio::spawn(async move { - while let Some(op) = codex_op_rx.recv().await { - let id = thread_clone.submit(op).await; - if let Err(e) = id { - tracing::error!("failed to submit op: {e}"); - } - } - }); - - while let Ok(event) = thread.next_event().await { - let is_shutdown_complete = matches!(event.msg, EventMsg::ShutdownComplete); - app_event_tx_clone.send(AppEvent::CodexEvent(event)); - if is_shutdown_complete { - // ShutdownComplete is terminal for a thread; drop this receiver task so - // the Arc can be released and thread resources can clean up. - break; - } - } - }); - - codex_op_tx -} - -/// Spawn an op-forwarding loop for an existing thread without subscribing to events. -pub(crate) fn spawn_op_forwarder(thread: std::sync::Arc) -> UnboundedSender { - let (codex_op_tx, mut codex_op_rx) = unbounded_channel::(); - - tokio::spawn(async move { - initialize_app_server_client_name(thread.as_ref()).await; - while let Some(op) = codex_op_rx.recv().await { - if let Err(e) = thread.submit(op).await { - tracing::error!("failed to submit op: {e}"); - } - } - }); - - codex_op_tx -} - /// Spawn an op forwarder that routes all widget submissions back onto the app event bus. /// /// The app layer owns app-server request translation during the migration, so the chat widget diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index 13c64a001d..151a880ca3 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -1755,7 +1755,7 @@ async fn helpers_are_available_and_do_not_panic() { let init = ChatWidgetInit { config: cfg, frame_requester: FrameRequester::test_dummy(), - app_event_tx: tx, + app_event_tx: tx.clone(), initial_user_message: None, enhanced_keys_supported: false, account: Some(codex_app_server_protocol::Account::ApiKey {}), @@ -1774,7 +1774,7 @@ async fn helpers_are_available_and_do_not_panic() { status_line_invalid_items_warned: Arc::new(AtomicBool::new(false)), session_telemetry, }; - let mut w = ChatWidget::new(init, thread_manager); + let mut w = ChatWidget::new_with_op_sender(init, spawn_app_event_forwarder(tx)); // Basic construction sanity. let _ = &mut w; } @@ -5625,10 +5625,12 @@ async fn collaboration_modes_defaults_to_code_on_startup() { cfg.model_provider.clone(), ), ); + let (app_event_tx_raw, _app_event_rx) = unbounded_channel::(); + let app_event_tx = AppEventSender::new(app_event_tx_raw); let init = ChatWidgetInit { config: cfg, frame_requester: FrameRequester::test_dummy(), - app_event_tx: AppEventSender::new(unbounded_channel::().0), + app_event_tx: app_event_tx.clone(), initial_user_message: None, enhanced_keys_supported: false, account: Some(codex_app_server_protocol::Account::ApiKey {}), @@ -5648,7 +5650,7 @@ async fn collaboration_modes_defaults_to_code_on_startup() { session_telemetry, }; - let chat = ChatWidget::new(init, thread_manager); + let chat = ChatWidget::new_with_op_sender(init, spawn_app_event_forwarder(app_event_tx)); assert_eq!(chat.active_collaboration_mode_kind(), ModeKind::Default); assert_eq!(chat.current_model(), resolved_model); } @@ -5679,10 +5681,12 @@ async fn experimental_mode_plan_is_ignored_on_startup() { cfg.model_provider.clone(), ), ); + let (app_event_tx_raw, _app_event_rx) = unbounded_channel::(); + let app_event_tx = AppEventSender::new(app_event_tx_raw); let init = ChatWidgetInit { config: cfg, frame_requester: FrameRequester::test_dummy(), - app_event_tx: AppEventSender::new(unbounded_channel::().0), + app_event_tx: app_event_tx.clone(), initial_user_message: None, enhanced_keys_supported: false, account: Some(codex_app_server_protocol::Account::ApiKey {}), @@ -5702,7 +5706,7 @@ async fn experimental_mode_plan_is_ignored_on_startup() { session_telemetry, }; - let chat = ChatWidget::new(init, thread_manager); + let chat = ChatWidget::new_with_op_sender(init, spawn_app_event_forwarder(app_event_tx)); assert_eq!(chat.active_collaboration_mode_kind(), ModeKind::Default); assert_eq!(chat.current_model(), resolved_model); }