openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-05 15:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

codex: simplify side-thread preparation (#26754)

Browse Source
This commit is contained in:
Eric Traut
2026-06-06 17:44:47 -07:00
parent 3e1be002ca
commit 1664555828
10 changed files with 68 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
codex-rs/tui/src/app.rs
View File

@@ -1259,7 +1259,7 @@ See the Codex keymap documentation for supported actions and examples."
self.handle_key_event(tui, app_server, key_event).await;
}
TuiEvent::Paste(pasted) => {
if self.side_start_active() {
if self.pending_side_start.is_some() {
return Ok(AppRunControl::Continue);
}
// Many terminals convert newlines to \r when pasting (e.g., iTerm2),

2
codex-rs/tui/src/app/event_dispatch.rs
View File

@@ -1723,7 +1723,7 @@ impl App {
self.handle_start_side(app_server, parent_thread_id, user_message)
.await;
}
AppEvent::SideThreadPrepared { request_id, result } => {
AppEvent::SideThreadPrepared(request_id, result) => {
self.handle_side_thread_prepared(tui, app_server, request_id, result)
.await?;
}

7
codex-rs/tui/src/app/input.rs
View File

@@ -97,10 +97,9 @@ impl App {
app_server: &mut AppServerSession,
key_event: KeyEvent,
) {
// The composer is read-only while /side prepares, but global shortcuts are handled here
// before input reaches the composer. Freeze the whole visible thread so navigation and
// commands cannot race the pending transition; Esc is the one escape hatch.
if self.side_start_active() {
// Freeze shortcuts as well as the composer while /side prepares so nothing can race the
// pending transition. Esc is the one escape hatch.
if self.pending_side_start.is_some() {
if matches!(key_event.code, KeyCode::Esc)
&& matches!(key_event.kind, KeyEventKind::Press | KeyEventKind::Repeat)
{

89
codex-rs/tui/src/app/side.rs
View File

@@ -198,12 +198,11 @@ pub(super) struct SideThreadState {
pub(super) struct PendingSideStart {
pub(super) request_id: Uuid,
pub(super) parent_thread_id: ThreadId,
pub(super) side_state: SideThreadState,
pub(super) user_message: Option<crate::chatwidget::UserMessage>,
}
impl SideThreadState {
#[cfg(test)]
pub(super) fn new(parent_thread_id: ThreadId) -> Self {
Self {
parent_thread_id,
@@ -266,6 +265,11 @@ impl App {
parent_thread_id: ThreadId,
status: Option<SideParentStatus>,
) {
if let Some(pending) = self.pending_side_start.as_mut()
&& pending.side_state.parent_thread_id == parent_thread_id
{
pending.side_state.parent_status = status;
}
let mut changed = false;
for state in self
.side_threads
@@ -283,6 +287,15 @@ impl App {
}
pub(super) fn clear_side_parent_action_status(&mut self, parent_thread_id: ThreadId) {
if let Some(pending) = self.pending_side_start.as_mut()
&& pending.side_state.parent_thread_id == parent_thread_id
&& pending
.side_state
.parent_status
.is_some_and(SideParentStatus::is_actionable)
{
pending.side_state.parent_status = None;
}
let mut changed = false;
for state in self
.side_threads
@@ -469,10 +482,6 @@ impl App {
}
}
pub(super) fn side_start_active(&self) -> bool {
self.pending_side_start.is_some()
}
pub(super) fn side_start_error_message(err: &str) -> String {
if err.contains("no rollout found for thread id")
|| err.contains("includeTurns is unavailable before first user message")
@@ -503,11 +512,6 @@ impl App {
self.restore_side_user_message(user_message);
}
pub(super) fn take_pending_side_start(&mut self, request_id: Uuid) -> Option<PendingSideStart> {
self.pending_side_start
.take_if(|pending| pending.request_id == request_id)
}
pub(super) fn install_side_thread_snapshot(
store: &mut ThreadEventStore,
mut session: ThreadSessionState,
@@ -569,7 +573,7 @@ impl App {
let request_id = Uuid::new_v4();
self.pending_side_start = Some(PendingSideStart {
request_id,
parent_thread_id,
side_state: SideThreadState::new(parent_thread_id),
user_message,
});
self.refresh_in_memory_config_from_disk_best_effort("starting a side conversation")
@@ -580,8 +584,8 @@ impl App {
let fork_config =
app_server.session_config_with_effective_service_tier(&self.side_fork_config());
let app_event_tx = self.app_event_tx.clone();
// App-server responses share the same bounded transport as notifications. Awaiting this
// request on the TUI loop would stop notification draining and can deadlock both sides.
// Awaiting this on the TUI loop would stop draining the bounded notification transport and
// can deadlock the fork response behind notifications.
tokio::spawn(async move {
let result = super::side_server::prepare_side_thread(
request_handle,
@@ -591,7 +595,7 @@ impl App {
remote_cwd_override,
)
.await;
app_event_tx.send(AppEvent::SideThreadPrepared { request_id, result });
app_event_tx.send(AppEvent::SideThreadPrepared(request_id, result));
});
}
@@ -606,10 +610,12 @@ impl App {
>,
) -> Result<()> {
let Some(PendingSideStart {
parent_thread_id,
side_state,
mut user_message,
..
}) = self.take_pending_side_start(request_id)
}) = self
.pending_side_start
.take_if(|pending| pending.request_id == request_id)
else {
if let Some(thread_id) = match &result {
Ok(started) => Some(started.session.thread_id),
@@ -619,20 +625,15 @@ impl App {
self.discard_thread_local_state(thread_id).await;
if result.is_ok() {
let request_handle = app_server.request_handle();
tokio::spawn(async move {
if let Err(err) =
super::side_server::cleanup_side_thread(request_handle, thread_id).await
{
tracing::warn!(
thread_id = %thread_id,
"failed to clean up abandoned side thread: {err}"
);
}
});
tokio::spawn(super::side_server::cleanup_side_thread(
request_handle,
thread_id,
));
}
}
return Ok(());
};
let parent_thread_id = side_state.parent_thread_id;
self.chat_widget.set_side_start_pending(/*pending*/ false);
match result {
Ok(started) => {
@@ -642,34 +643,13 @@ impl App {
let mut store = channel.store.lock().await;
Self::install_side_thread_snapshot(&mut store, started.session, started.turns);
}
let parent_status =
if let Some(channel) = self.thread_event_channels.get(&parent_thread_id) {
channel.store.lock().await.side_parent_status()
} else {
None
};
self.side_threads.insert(
child_thread_id,
SideThreadState {
parent_thread_id,
parent_status,
},
);
self.upsert_agent_picker_thread(
child_thread_id,
/*agent_nickname*/ None,
/*agent_role*/ None,
/*is_closed*/ false,
);
// Known side threads take the selector's local replay path, so this does not wait
// for another app-server response on the TUI loop.
self.side_threads.insert(child_thread_id, side_state);
// Prepared side threads switch through the selector's local replay path.
if let Err(err) = self
.select_agent_thread(tui, app_server, child_thread_id)
.await
{
let discarded = self.discard_side_thread(app_server, child_thread_id).await;
if discarded
&& self.active_thread_id != Some(parent_thread_id)
if self.discard_side_thread(app_server, child_thread_id).await
&& let Err(restore_err) = self
.select_agent_thread(tui, app_server, parent_thread_id)
.await
@@ -678,13 +658,6 @@ impl App {
"failed to restore parent thread after side switch failure: \
{restore_err}"
);
} else if !discarded {
self.keep_side_thread_visible_after_cleanup_failure(
tui,
app_server,
child_thread_id,
)
.await;
}
self.restore_side_user_message(user_message.take());
self.chat_widget.add_error_message(format!(

30
codex-rs/tui/src/app/side_server.rs
View File

@@ -49,12 +49,7 @@ pub(super) async fn prepare_side_thread(
.await;
if let Err(err) = inject_result {
// The caller only receives fully prepared threads, so clean up this partial fork here.
if let Err(cleanup_err) = cleanup_side_thread(request_handle, child_thread_id).await {
tracing::warn!(
thread_id = %child_thread_id,
"failed to clean up side thread after inject failure: {cleanup_err}"
);
}
cleanup_side_thread(request_handle, child_thread_id).await;
return Err(SideThreadPrepareError {
thread_id: Some(child_thread_id),
message: format!(
@@ -68,7 +63,7 @@ pub(super) async fn prepare_side_thread(
pub(super) async fn cleanup_side_thread(
request_handle: AppServerRequestHandle,
thread_id: ThreadId,
) -> Result<()> {
) {
let interrupt_result = request_handle
.request_typed::<TurnInterruptResponse>(ClientRequest::TurnInterrupt {
request_id: RequestId::String(format!("side-thread-interrupt-{}", Uuid::new_v4())),
@@ -77,8 +72,7 @@ pub(super) async fn cleanup_side_thread(
turn_id: String::new(),
},
})
.await
.wrap_err("turn/interrupt failed while cleaning up TUI side thread");
.await;
let unsubscribe_result = request_handle
.request_typed::<ThreadUnsubscribeResponse>(ClientRequest::ThreadUnsubscribe {
request_id: RequestId::String(format!("side-thread-unsubscribe-{}", Uuid::new_v4())),
@@ -86,9 +80,17 @@ pub(super) async fn cleanup_side_thread(
thread_id: thread_id.to_string(),
},
})
.await
.wrap_err("thread/unsubscribe failed in TUI")
.map(drop);
interrupt_result?;
unsubscribe_result
.await;
if let Err(err) = interrupt_result {
tracing::warn!(
thread_id = %thread_id,
"failed to interrupt side thread during cleanup: {err}"
);
}
if let Err(err) = unsubscribe_result {
tracing::warn!(
thread_id = %thread_id,
"failed to unsubscribe side thread during cleanup: {err}"
);
}
}

19
codex-rs/tui/src/app/tests.rs
View File

@@ -3387,7 +3387,7 @@ async fn cancelling_pending_side_start_restores_message() {
let mut app = make_test_app().await;
app.pending_side_start = Some(PendingSideStart {
request_id: Uuid::new_v4(),
parent_thread_id: ThreadId::new(),
side_state: SideThreadState::new(ThreadId::new()),
user_message: Some(crate::chatwidget::UserMessage::from("side question")),
});
app.chat_widget.set_side_start_pending(/*pending*/ true);
@@ -3401,23 +3401,6 @@ async fn cancelling_pending_side_start_restores_message() {
);
}
#[tokio::test]
async fn side_completion_only_consumes_its_own_pending_start() {
let mut app = make_test_app().await;
let current_request_id = Uuid::new_v4();
app.pending_side_start = Some(PendingSideStart {
request_id: current_request_id,
parent_thread_id: ThreadId::new(),
user_message: None,
});
assert!(app.take_pending_side_start(Uuid::new_v4()).is_none());
assert_eq!(
app.pending_side_start.as_ref().unwrap().request_id,
current_request_id
);
}
#[tokio::test]
async fn side_discard_selection_keeps_current_side_thread() {
let mut app = make_test_app().await;

33
codex-rs/tui/src/app/thread_events.rs
View File

@@ -240,7 +240,7 @@ impl ThreadEventStore {
.has_pending_thread_approvals()
}
pub(super) fn side_parent_status(&self) -> Option<SideParentStatus> {
pub(super) fn side_parent_pending_status(&self) -> Option<SideParentStatus> {
if self
.pending_interactive_replay
.has_pending_thread_user_input()
@@ -252,20 +252,7 @@ impl ThreadEventStore {
{
Some(SideParentStatus::NeedsApproval)
} else {
self.buffer
.iter()
.rev()
.find_map(|event| {
let ThreadBufferedEvent::Notification(notification) = event else {
return None;
};
match SideParentStatusChange::for_notification(notification) {
Some(SideParentStatusChange::Set(status)) => Some(Some(status)),
Some(SideParentStatusChange::Clear) => Some(None),
Some(SideParentStatusChange::ClearActionable) | None => None,
}
})
.flatten()
None
}
}
@@ -520,22 +507,6 @@ mod tests {
assert_eq!(store.active_turn_id(), None);
}
#[test]
fn thread_event_store_reports_latest_side_parent_status() {
let mut store = ThreadEventStore::new(/*capacity*/ 8);
let thread_id = ThreadId::new();
store.push_notification(turn_completed_notification(
thread_id,
"turn-1",
TurnStatus::Failed,
));
assert_eq!(store.side_parent_status(), Some(SideParentStatus::Failed));
store.push_notification(turn_started_notification(thread_id, "turn-2"));
assert_eq!(store.side_parent_status(), None);
}
#[test]
fn thread_event_store_restores_active_turn_from_snapshot_turns() {
let thread_id = ThreadId::new();

6
codex-rs/tui/src/app/thread_routing.rs
View File

@@ -817,7 +817,7 @@ impl App {
};
let status = {
let store = channel.store.lock().await;
store.side_parent_status()
store.side_parent_pending_status()
};
if let Some(status) = status {
self.set_side_parent_status(thread_id, Some(status));
@@ -858,7 +858,7 @@ impl App {
guard.session = Some(session);
}
guard.push_notification(notification.clone());
(guard.active, guard.side_parent_status())
(guard.active, guard.side_parent_pending_status())
};
let notification_status_change = SideParentStatusChange::for_notification(&notification);
@@ -976,7 +976,7 @@ impl App {
let (should_send, pending_status) = {
let mut guard = store.lock().await;
guard.push_request(request.clone());
(guard.active, guard.side_parent_status())
(guard.active, guard.side_parent_pending_status())
};
let request_status = SideParentStatus::for_request(&request);

5
codex-rs/tui/src/app_event.rs
View File

@@ -154,10 +154,7 @@ pub(crate) enum AppEvent {
user_message: Option<UserMessage>,
},
/// Finish preparing a transient side conversation off the TUI event loop.
SideThreadPrepared {
request_id: Uuid,
result: Result<AppServerStartedThread, SideThreadPrepareError>,
},
SideThreadPrepared(Uuid, Result<AppServerStartedThread, SideThreadPrepareError>),
/// Submit an op to the specified thread, regardless of current focus.
SubmitThreadOp {

21
codex-rs/tui/src/session_log.rs
View File

@@ -197,24 +197,21 @@ pub(crate) fn log_inbound_app_event(event: &AppEvent) {
});
LOGGER.write_json_line(value);
}
AppEvent::SideThreadPrepared { request_id, result } => {
let value = json!({
"ts": now_ts(),
"dir": "to_tui",
"kind": "app_event",
"variant": "SideThreadPrepared",
"request_id": request_id,
"ok": result.is_ok(),
});
LOGGER.write_json_line(value);
}
// Noise or control flow – record variant only
other => {
let variant = match other {
AppEvent::SideThreadPrepared(..) => "SideThreadPrepared".to_string(),
_ => format!("{other:?}")
.split('(')
.next()
.unwrap_or("app_event")
.to_string(),
};
let value = json!({
"ts": now_ts(),
"dir": "to_tui",
"kind": "app_event",
"variant": format!("{other:?}").split('(').next().unwrap_or("app_event"),
"variant": variant,
});
LOGGER.write_json_line(value);
}