mirror of
https://github.com/openai/codex.git
synced 2026-10-19 08:16:48 +00:00
Preserve side conversations across thread navigation (#52184)
## Why Navigating to another thread discarded side conversations, preventing users from returning to them later. ## What changed - Keep side conversations and their event listeners, channels, and pending requests across thread switches and session resets. - Reopen a parent's existing side conversation with `/side` and submit any supplied question there. Resolve side toggles by the current thread's parent relationship. - Limit explicit return-and-discard behavior to the active side conversation, and shut down retained side conversations when exiting. ## Testing Extend regression tests to cover reopening a side conversation after navigating away, preserving its listener and pending approval during a reset, and restricting discard selection to the active side. GitOrigin-RevId: ff80566970414ba190ff650d5e413ef63dc65449
This commit is contained in:
@@ -690,25 +690,6 @@ impl App {
|
||||
}
|
||||
}
|
||||
};
|
||||
if !previous_running_thread_ids.is_empty() {
|
||||
for side_thread_id in Vec::from_iter(self.side_threads.keys().copied()) {
|
||||
let discarded = match startup_draft.as_deref_mut() {
|
||||
Some(draft) => {
|
||||
draft
|
||||
.run_until(
|
||||
tui,
|
||||
self.discard_side_thread(app_server, side_thread_id),
|
||||
)
|
||||
.await?
|
||||
}
|
||||
None => self.discard_side_thread(app_server, side_thread_id).await,
|
||||
};
|
||||
if !discarded {
|
||||
let _ = app_server.thread_unsubscribe(root_thread_id).await;
|
||||
return Ok(AppRunControl::Continue);
|
||||
}
|
||||
}
|
||||
}
|
||||
for (thread_id, requests) in previous_pending_requests {
|
||||
self.agents_overview
|
||||
.dispatched_requests
|
||||
@@ -849,6 +830,7 @@ impl App {
|
||||
for thread_id in previous_thread_ids {
|
||||
if previous_running_thread_ids.is_empty()
|
||||
&& thread_id != root_thread_id
|
||||
&& !self.side_threads.contains_key(&thread_id)
|
||||
&& self.voice_owner_thread_id() != Some(thread_id)
|
||||
&& Some(thread_id) != previous_displayed_thread_id
|
||||
&& !self.agents_overview.blank_sessions.contains_key(&thread_id)
|
||||
@@ -868,7 +850,7 @@ impl App {
|
||||
|| (self.thread_unavailable(root_thread_id)
|
||||
&& !self.chat_widget.is_external_writer_view())
|
||||
{
|
||||
self.select_agent_thread_and_discard_side(tui, app_server, root_thread_id)
|
||||
self.select_agent_thread(tui, app_server, root_thread_id)
|
||||
.await?;
|
||||
}
|
||||
let read_only = self.chat_widget.is_external_writer_view();
|
||||
|
||||
@@ -2961,7 +2961,7 @@ impl App {
|
||||
);
|
||||
}
|
||||
AppEvent::SelectAgentThread(thread_id) => {
|
||||
self.select_agent_thread_and_discard_side(tui, app_server, thread_id)
|
||||
self.select_agent_thread(tui, app_server, thread_id)
|
||||
.await?;
|
||||
}
|
||||
AppEvent::StartSide {
|
||||
@@ -3579,6 +3579,16 @@ impl App {
|
||||
// its shutdown completion does not trigger agent failover.
|
||||
self.pending_shutdown_exit_thread_id =
|
||||
self.active_thread_id.or(self.chat_widget.thread_id());
|
||||
if !self.side_threads.is_empty()
|
||||
&& tokio::time::timeout(
|
||||
SHUTDOWN_FIRST_EXIT_TIMEOUT,
|
||||
self.shutdown_side_threads(app_server),
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!("timed out waiting for side-conversation shutdown");
|
||||
}
|
||||
if self.pending_shutdown_exit_thread_id.is_some()
|
||||
|| self.voice_owner_thread_id().is_some()
|
||||
{
|
||||
|
||||
@@ -445,9 +445,7 @@ impl App {
|
||||
.adjacent_thread_id_with_backfill(app_server, AgentNavigationDirection::Previous)
|
||||
.await
|
||||
{
|
||||
let _ = self
|
||||
.select_agent_thread_and_discard_side(tui, app_server, thread_id)
|
||||
.await;
|
||||
let _ = self.select_agent_thread(tui, app_server, thread_id).await;
|
||||
}
|
||||
return;
|
||||
}
|
||||
@@ -462,9 +460,7 @@ impl App {
|
||||
.adjacent_thread_id_with_backfill(app_server, AgentNavigationDirection::Next)
|
||||
.await
|
||||
{
|
||||
let _ = self
|
||||
.select_agent_thread_and_discard_side(tui, app_server, thread_id)
|
||||
.await;
|
||||
let _ = self.select_agent_thread(tui, app_server, thread_id).await;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -808,7 +808,7 @@ impl App {
|
||||
}
|
||||
if voice_owner.is_some() {
|
||||
for (thread_id, channel) in &self.thread_event_channels {
|
||||
if Some(*thread_id) != voice_owner {
|
||||
if Some(*thread_id) != voice_owner && !self.side_threads.contains_key(thread_id) {
|
||||
for request in channel.store.lock().await.pending_replay_requests() {
|
||||
self.pending_app_server_requests
|
||||
.resolve_notification(&thread_id.to_string(), request.id());
|
||||
@@ -817,7 +817,9 @@ impl App {
|
||||
}
|
||||
}
|
||||
self.thread_event_listener_tasks.retain(|id, task| {
|
||||
if Some(*id) == voice_owner || self.agents_overview.dispatched_requests.contains_key(id)
|
||||
if Some(*id) == voice_owner
|
||||
|| self.side_threads.contains_key(id)
|
||||
|| self.agents_overview.dispatched_requests.contains_key(id)
|
||||
{
|
||||
true
|
||||
} else {
|
||||
@@ -826,14 +828,13 @@ impl App {
|
||||
}
|
||||
});
|
||||
self.thread_event_channels
|
||||
.retain(|id, _| Some(*id) == voice_owner);
|
||||
.retain(|id, _| Some(*id) == voice_owner || self.side_threads.contains_key(id));
|
||||
self.pending_realtime_speech_replay.clear();
|
||||
self.pending_realtime_transcript_replay.clear();
|
||||
self.realtime_replay_order.clear();
|
||||
self.pending_server_profiles.clear();
|
||||
self.agents_overview.activity.clear();
|
||||
self.agent_navigation.clear();
|
||||
self.side_threads.clear();
|
||||
self.active_thread_id = None;
|
||||
self.active_thread_rx = None;
|
||||
self.primary_thread_id = None;
|
||||
@@ -842,6 +843,16 @@ impl App {
|
||||
self.pending_primary_events.clear();
|
||||
if voice_owner.is_none() {
|
||||
self.pending_app_server_requests.clear();
|
||||
let side_thread_ids: Vec<_> = self.side_threads.keys().copied().collect();
|
||||
for thread_id in side_thread_ids {
|
||||
if let Some(channel) = self.thread_event_channels.get(&thread_id) {
|
||||
for request in channel.store.lock().await.pending_replay_requests() {
|
||||
let _ = self
|
||||
.pending_app_server_requests
|
||||
.note_server_request(&request);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
self.pending_startup_thread_start = false;
|
||||
self.pending_server_version_notice = None;
|
||||
|
||||
@@ -2,8 +2,8 @@
|
||||
//!
|
||||
//! A side conversation is an ephemeral fork used for a quick /side question while keeping the
|
||||
//! primary thread focused. This module owns the app-level lifecycle for those forks: switching into
|
||||
//! them, returning to their parent, and discarding them when normal thread navigation moves
|
||||
//! elsewhere. The fork receives hidden developer instructions that make inherited history reference
|
||||
//! them, returning to their parent, retaining them across normal navigation, and explicitly
|
||||
//! discarding them. The fork receives hidden developer instructions that make inherited history
|
||||
//! material only and steer the agent away from mutations unless the side conversation explicitly asks
|
||||
//! for them.
|
||||
|
||||
@@ -299,6 +299,12 @@ impl App {
|
||||
.map(|state| state.parent_thread_id)
|
||||
}
|
||||
|
||||
pub(super) fn side_thread_for_parent(&self, parent_thread_id: ThreadId) -> Option<ThreadId> {
|
||||
self.side_threads.iter().find_map(|(thread_id, state)| {
|
||||
(state.parent_thread_id == parent_thread_id).then_some(*thread_id)
|
||||
})
|
||||
}
|
||||
|
||||
pub(super) fn set_side_parent_status(
|
||||
&mut self,
|
||||
parent_thread_id: ThreadId,
|
||||
@@ -386,13 +392,8 @@ impl App {
|
||||
target_thread_id: ThreadId,
|
||||
) -> Option<ThreadId> {
|
||||
let active_thread_id = self.current_displayed_thread_id()?;
|
||||
let (&side_thread_id, state) = self.side_threads.iter().next()?;
|
||||
if target_thread_id == side_thread_id || target_thread_id == active_thread_id {
|
||||
return None;
|
||||
}
|
||||
|
||||
(active_thread_id == side_thread_id || active_thread_id == state.parent_thread_id)
|
||||
.then_some(side_thread_id)
|
||||
let state = self.side_threads.get(&active_thread_id)?;
|
||||
(target_thread_id == state.parent_thread_id).then_some(active_thread_id)
|
||||
}
|
||||
|
||||
pub(super) async fn toggle_side_conversation(
|
||||
@@ -403,12 +404,9 @@ impl App {
|
||||
let Some(active_thread_id) = self.current_displayed_thread_id() else {
|
||||
return Ok(());
|
||||
};
|
||||
let Some((&side_thread_id, state)) = self.side_threads.iter().next() else {
|
||||
return Ok(());
|
||||
};
|
||||
let target_thread_id = if active_thread_id == side_thread_id {
|
||||
let target_thread_id = if let Some(state) = self.side_threads.get(&active_thread_id) {
|
||||
state.parent_thread_id
|
||||
} else if active_thread_id == state.parent_thread_id {
|
||||
} else if let Some(side_thread_id) = self.side_thread_for_parent(active_thread_id) {
|
||||
side_thread_id
|
||||
} else {
|
||||
return Ok(());
|
||||
@@ -725,12 +723,18 @@ impl App {
|
||||
return Ok(AppRunControl::Continue);
|
||||
}
|
||||
|
||||
if let Some((&side_thread_id, state)) = self.side_threads.iter().next()
|
||||
&& (parent_thread_id != state.parent_thread_id
|
||||
|| !self.discard_side_thread(app_server, side_thread_id).await)
|
||||
{
|
||||
self.restore_side_user_message(user_message.take());
|
||||
self.sync_side_thread_ui();
|
||||
if let Some(side_thread_id) = self.side_thread_for_parent(parent_thread_id) {
|
||||
self.select_agent_thread(tui, app_server, side_thread_id)
|
||||
.await?;
|
||||
if self.active_thread_id == Some(side_thread_id) {
|
||||
if let Some(user_message) = user_message.take() {
|
||||
let _ = self
|
||||
.chat_widget
|
||||
.submit_user_message_as_plain_user_turn(user_message);
|
||||
}
|
||||
} else {
|
||||
self.restore_side_user_message(user_message.take());
|
||||
}
|
||||
return Ok(AppRunControl::Continue);
|
||||
}
|
||||
|
||||
@@ -783,7 +787,7 @@ impl App {
|
||||
return Ok(AppRunControl::Continue);
|
||||
}
|
||||
if let Err(err) = self
|
||||
.select_agent_thread_and_discard_side(tui, app_server, child_thread_id)
|
||||
.select_agent_thread(tui, app_server, child_thread_id)
|
||||
.await
|
||||
{
|
||||
let discarded = self
|
||||
|
||||
@@ -857,6 +857,13 @@ async fn reset_thread_event_state_aborts_listener_tasks() {
|
||||
|
||||
let mut app = make_test_app().await;
|
||||
let thread_id = ThreadId::new();
|
||||
let side_thread_id = ThreadId::new();
|
||||
let side_request = exec_approval_request(
|
||||
side_thread_id,
|
||||
"turn-1",
|
||||
"call-1",
|
||||
/*approval_id*/ None,
|
||||
);
|
||||
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
|
||||
let (dropped_tx, dropped_rx) = tokio::sync::oneshot::channel();
|
||||
let handle = tokio::spawn(async move {
|
||||
@@ -865,6 +872,21 @@ async fn reset_thread_event_state_aborts_listener_tasks() {
|
||||
std::future::pending::<()>().await;
|
||||
});
|
||||
app.thread_event_listener_tasks.insert(thread_id, handle);
|
||||
app.thread_event_listener_tasks
|
||||
.insert(side_thread_id, tokio::spawn(std::future::pending()));
|
||||
app.thread_event_channels.insert(
|
||||
side_thread_id,
|
||||
ThreadEventChannel::new(THREAD_EVENT_CHANNEL_CAPACITY),
|
||||
);
|
||||
app.thread_event_channels[&side_thread_id]
|
||||
.store
|
||||
.lock()
|
||||
.await
|
||||
.push_request(side_request.clone());
|
||||
app.side_threads
|
||||
.insert(side_thread_id, SideThreadState::new(thread_id));
|
||||
app.pending_app_server_requests
|
||||
.note_server_request(&side_request);
|
||||
app.pending_server_profiles.insert(
|
||||
thread_id,
|
||||
PermissionProfileSelection {
|
||||
@@ -880,7 +902,19 @@ async fn reset_thread_event_state_aborts_listener_tasks() {
|
||||
|
||||
app.reset_thread_event_state().await;
|
||||
|
||||
assert_eq!(app.thread_event_listener_tasks.is_empty(), true);
|
||||
assert_eq!(
|
||||
app.thread_event_listener_tasks
|
||||
.keys()
|
||||
.copied()
|
||||
.collect::<Vec<_>>(),
|
||||
vec![side_thread_id]
|
||||
);
|
||||
assert!(app.thread_event_channels.contains_key(&side_thread_id));
|
||||
assert!(app.side_threads.contains_key(&side_thread_id));
|
||||
assert!(
|
||||
app.pending_app_server_requests
|
||||
.contains_server_request(&side_request)
|
||||
);
|
||||
assert!(app.pending_server_profiles.is_empty());
|
||||
time::timeout(Duration::from_millis(50), dropped_rx)
|
||||
.await
|
||||
@@ -2890,6 +2924,17 @@ async fn handle_start_side_seeds_navigation_before_thread_started() -> Result<()
|
||||
)
|
||||
.expect("create source rollout"),
|
||||
)?;
|
||||
let other_thread_id = ThreadId::from_string(
|
||||
&app_test_support::create_fake_rollout(
|
||||
config.codex_home.as_path(),
|
||||
"2025-01-06T12-00-00",
|
||||
"2025-01-06T12:00:00Z",
|
||||
"Other task",
|
||||
Some(config.model_provider_id.as_str()),
|
||||
/*git_info*/ None,
|
||||
)
|
||||
.expect("create other rollout"),
|
||||
)?;
|
||||
let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker(&config)).await?;
|
||||
let started = app_server
|
||||
.resume_thread(
|
||||
@@ -2969,6 +3014,19 @@ async fn handle_start_side_seeds_navigation_before_thread_started() -> Result<()
|
||||
app.chat_widget.config_ref()
|
||||
))
|
||||
);
|
||||
app.select_agents_overview_thread(&mut tui, &mut app_server, other_thread_id)
|
||||
.await?;
|
||||
app.select_agents_overview_thread(&mut tui, &mut app_server, parent_thread_id)
|
||||
.await?;
|
||||
Box::pin(app.handle_start_side(
|
||||
&mut tui,
|
||||
&mut app_server,
|
||||
parent_thread_id,
|
||||
/*user_message*/ None,
|
||||
))
|
||||
.await?;
|
||||
assert_eq!(app.active_thread_id, Some(side_thread_id));
|
||||
assert_eq!(app.side_threads.len(), 1);
|
||||
app.select_agent_thread(&mut tui, &mut app_server, parent_thread_id)
|
||||
.await?;
|
||||
app.select_permission_profile(
|
||||
@@ -5651,7 +5709,7 @@ async fn side_restore_user_message_puts_inline_question_back_in_composer() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn side_discard_selection_keeps_current_side_thread() {
|
||||
async fn side_discard_selection_only_closes_the_active_side() {
|
||||
let mut app = make_test_app().await;
|
||||
let parent_thread_id = ThreadId::new();
|
||||
let side_thread_id = ThreadId::new();
|
||||
@@ -5671,7 +5729,7 @@ async fn side_discard_selection_keeps_current_side_thread() {
|
||||
app.active_thread_id = Some(parent_thread_id);
|
||||
assert_eq!(
|
||||
app.side_thread_to_discard_after_switch(ThreadId::new()),
|
||||
Some(side_thread_id)
|
||||
None
|
||||
);
|
||||
assert_eq!(
|
||||
app.side_thread_to_discard_after_switch(side_thread_id),
|
||||
|
||||
@@ -2175,12 +2175,9 @@ impl App {
|
||||
self.mark_agent_picker_thread_closed(closed_thread_id);
|
||||
if self.side_threads.contains_key(&closed_thread_id) {
|
||||
self.discard_closed_side_thread(closed_thread_id).await;
|
||||
self.select_agent_thread(tui, app_server, primary_thread_id)
|
||||
.await?;
|
||||
} else {
|
||||
self.select_agent_thread_and_discard_side(tui, app_server, primary_thread_id)
|
||||
.await?;
|
||||
}
|
||||
self.select_agent_thread(tui, app_server, primary_thread_id)
|
||||
.await?;
|
||||
if self.active_thread_id == Some(primary_thread_id) {
|
||||
self.chat_widget.add_info_message(
|
||||
format!(
|
||||
|
||||
@@ -125,11 +125,11 @@ impl App {
|
||||
app_server: &mut AppServerSession,
|
||||
destination: Option<ThreadId>,
|
||||
) {
|
||||
self.shutdown_side_threads(app_server).await;
|
||||
let thread_ids: Vec<_> = self.thread_event_channels.keys().copied().collect();
|
||||
for thread_id in thread_ids {
|
||||
if self.voice_owner_thread_id() == Some(thread_id)
|
||||
|| destination == Some(thread_id)
|
||||
|| self.side_threads.contains_key(&thread_id)
|
||||
|| self
|
||||
.agents_overview
|
||||
.dispatched_requests
|
||||
|
||||
Reference in New Issue
Block a user