mirror of
https://github.com/openai/codex.git
synced 2026-10-27 18:48:37 +00:00
Part 5
This commit is contained in:
Eric Traut
@@ -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<ThreadManager>) -> 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<Op>,
|
||||
@@ -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<codex_core::CodexThread>,
|
||||
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 {
|
||||
|
||||
@@ -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<Op>` used by the UI to submit operations.
|
||||
pub(crate) fn spawn_agent(
|
||||
config: Config,
|
||||
app_event_tx: AppEventSender,
|
||||
server: Arc<ThreadManager>,
|
||||
) -> UnboundedSender<Op> {
|
||||
let (codex_op_tx, mut codex_op_rx) = unbounded_channel::<Op>();
|
||||
|
||||
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<CodexThread> 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<CodexThread>,
|
||||
session_configured: codex_protocol::protocol::SessionConfiguredEvent,
|
||||
app_event_tx: AppEventSender,
|
||||
) -> UnboundedSender<Op> {
|
||||
let (codex_op_tx, mut codex_op_rx) = unbounded_channel::<Op>();
|
||||
|
||||
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<CodexThread> 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<CodexThread>) -> UnboundedSender<Op> {
|
||||
let (codex_op_tx, mut codex_op_rx) = unbounded_channel::<Op>();
|
||||
|
||||
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
|
||||
|
||||
@@ -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::<AppEvent>();
|
||||
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::<AppEvent>().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::<AppEvent>();
|
||||
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::<AppEvent>().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);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user