Reconnect TUI app-server sessions automatically (#41916)

## What changed

- Reconnect out-of-process TUI sessions after the app-server transport drops, bootstrap a fresh client, and resume the active thread.
- Preserve cached transcripts, drafts, runtime settings, task-tool capabilities, and live notification routing across the new connection.
- Keep recovered or possibly submitted input paused for manual review instead of resending it, and leave conversations that cannot be resumed available as read-only cached views.
- Show reconnect progress and actionable failure states while still allowing drafts to be edited or copied.

## Testing

- Add app-level coverage for successful recovery, failed and unavailable resumes, input quarantine, and preserved session state.
- Extend the terminal integration test to verify that a draft survives reconnection, the thread is resumed without starting a turn, and notifications arrive through the replacement connection.

GitOrigin-RevId: 5dcb867c89e226a0c545d6313a77424c233cd47a
This commit is contained in:
Eric Traut
2026-08-31 21:36:49 +00:00
committed by copyberry
parent 9a4b78579a
commit 907c34e867
20 changed files with 1082 additions and 46 deletions
+1 -1
View File
@@ -50,7 +50,7 @@ pub(crate) struct AgentNavigationState {
/// Spawned child threads whose instructions are owned by their parent agent.
parent_owned_threads: HashSet<ThreadId>,
/// Coalesces root refreshes while rejecting replies from a previous session.
picker_refresh: Option<(ThreadId, Uuid)>,
pub(super) picker_refresh: Option<(ThreadId, Uuid)>,
}
/// Direction of keyboard traversal through the stable picker order.
+11
View File
@@ -370,6 +370,17 @@ impl App {
return;
}
if !self.chat_widget.has_active_view()
&& self
.current_displayed_thread_id()
.is_some_and(|id| self.thread_unavailable(id))
&& !(key_event.modifiers.contains(KeyModifiers::CONTROL)
&& matches!(key_event.code, KeyCode::Char('c' | 'd')))
{
self.chat_widget.handle_disconnected_key(key_event);
return;
}
if matches!(key_event.code, KeyCode::Esc)
&& matches!(key_event.kind, KeyEventKind::Press | KeyEventKind::Repeat)
{
+285 -2
View File
@@ -1,7 +1,11 @@
//! Preserve local work after an external connection is lost. Offline input and stale
//! async completions are quarantined; recovery requires relaunching the TUI.
//! TUI-only out-of-process reconnection. Each attempt initializes a fresh client and rejoins existing
//! threads using ordinary resume/history semantics; no user operation is retried.
//! Offline input and old async completions are quarantined. Cached conversations stay read-only
//! until they can be reattached; command-center recovery is handled separately.
use super::*;
use crate::app_server_session::ResumeModelSettings;
use crate::dynamic_tools_mcp::ThreadToolTransport;
#[derive(Clone, Copy, Default, PartialEq, Eq)]
pub(super) enum ReconnectPresentation {
@@ -13,10 +17,110 @@ pub(super) enum ReconnectPresentation {
#[derive(Default)]
pub(super) struct ReconnectState {
pub(super) offline: bool,
pub(super) failed: bool,
pub(super) presentation: ReconnectPresentation,
}
pub(super) struct Reconnected {
session: AppServerSession,
bootstrap: AppServerBootstrap,
thread: Option<AppServerStartedThread>,
}
pub(super) async fn reconnect(
target: AppServerTarget,
config: Config,
thread_id: Option<ThreadId>,
remote_cwd: Option<PathBuf>,
task_tools: ThreadToolTransport,
presentation: ReconnectPresentation,
) -> Result<Reconnected> {
let mode = target.thread_params_mode();
let endpoint = match target {
AppServerTarget::Remote { endpoint } | AppServerTarget::LocalDaemon { endpoint } => {
endpoint
}
AppServerTarget::Embedded => {
color_eyre::eyre::bail!("in-process sessions have no connection to restore")
}
};
if let ThreadToolTransport::Mcp(server) = &task_tools {
server.suspend();
}
if presentation == ReconnectPresentation::Conversation && thread_id.is_none() {
color_eyre::eyre::bail!(
"The initial thread may have been created, but its ID was not received. Nothing was retried. Your prompt is editable; inspect your tasks before relaunching."
);
}
// Connecting already has transport deadlines. Give healthy history/inventory hydration one
// shared budget instead of repeatedly discarding its progress on a short per-attempt timer.
let deadline = tokio::time::Instant::now() + Duration::from_secs(/*secs*/ 120);
for delay in [0, 1, 2, 4, 8] {
let attempt = async {
tokio::time::sleep(Duration::from_secs(delay)).await;
let client = crate::connect_remote_app_server(endpoint.clone()).await?;
let mut session = AppServerSession::new(client, mode)
.with_startup_config(&config)
.with_remote_cwd_override(remote_cwd.clone())
.with_thread_tool_transport(task_tools.clone());
let bootstrap = session.bootstrap(&config).await?;
let thread = if let Some(thread_id) = thread_id {
match session
.resume_thread(
config.clone(),
thread_id,
ResumeModelSettings::PreserveExistingThread,
)
.await
{
Ok(thread) => Some(thread),
Err(error)
if matches!(
error.downcast_ref::<TypedRequestError>(),
Some(TypedRequestError::Transport { .. })
) =>
{
return Err(error);
}
Err(error)
if matches!(
error.downcast_ref::<TypedRequestError>(),
Some(TypedRequestError::Server { source, .. }) if source.code == -32600
) || presentation == ReconnectPresentation::Overview =>
{
None
}
Err(error) => return Err(error),
}
} else {
None
};
Ok::<_, color_eyre::Report>(Reconnected {
session,
bootstrap,
thread,
})
};
let result = tokio::time::timeout_at(deadline, attempt).await;
match result {
Ok(Ok(connected)) => return Ok(connected),
Ok(Err(_)) => {}
Err(_) => break,
}
// Transport errors can contain endpoint credentials. Do not render or log them.
}
color_eyre::eyre::bail!("app-server session could not be restored")
}
impl App {
pub(super) fn thread_unavailable(&self, id: ThreadId) -> bool {
!matches!(self.app_server_target, AppServerTarget::Embedded)
&& self
.thread_event_channels
.get(&id)
.is_some_and(|channel| channel.attachment() == ThreadEventAttachment::ReplayOnly)
}
pub(super) fn recover_transport_error(&mut self, error: &color_eyre::Report) -> bool {
let disconnected = matches!(
error.downcast_ref::<TypedRequestError>(),
@@ -31,6 +135,7 @@ impl App {
}
if !self.reconnect.offline {
self.reconnect.offline = true;
self.reconnect.failed = false;
self.cancel_pending_key_chord();
self.overlay = None;
self.commit_animation = None;
@@ -63,4 +168,182 @@ impl App {
}
true
}
pub(super) async fn finish_reconnect(
&mut self,
tui: &mut tui::Tui,
app_server: &mut AppServerSession,
app_event_rx: &mut mpsc::UnboundedReceiver<AppEvent>,
connected: Reconnected,
) -> Result<()> {
let Reconnected {
mut session,
bootstrap,
thread,
} = connected;
let displayed = self.current_displayed_thread_id();
let mut input = self.chat_widget.capture_thread_input_state();
if let Some(input) = input.as_mut() {
input.recovered_queue = true;
}
self.store_active_thread_receiver().await;
// Old request handles stay attached to the dead connection. Rotating the event channel
// prevents their late completions from submitting follow-up operations on the new one.
let (tx, rx) = mpsc::unbounded_channel();
self.app_event_tx = AppEventSender::new(tx);
*app_event_rx = rx;
self.agent_navigation.picker_refresh = None;
self.last_subagent_backfill_attempt = None;
self.rate_limit_refresh_state.invalidate_recovery();
session.inherit_task_tool_capabilities(app_server);
*app_server = session;
self.chat_widget.remote_connection =
crate::status::remote_connection::remote_connection_status_value(
&self.app_server_target,
app_server.server_version(),
);
self.workspace_command_runner = Some(Arc::new(AppServerWorkspaceCommandRunner::new(
app_server.request_handle(),
)));
self.file_search =
FileSearchManager::new(self.config.cwd.to_path_buf(), self.app_event_tx.clone());
self.model_catalog = Arc::new(ModelCatalog::new(bootstrap.available_models));
self.pending_app_server_requests.clear();
self.pending_primary_events.clear();
self.pending_plugin_enabled_writes.clear();
self.pending_hook_enabled_writes.clear();
self.temporary_structured_requests.clear();
self.agents_overview.dispatched_requests.clear();
self.agents_overview.request_id = None;
self.agents_overview.refresh_pending = false;
for input in self.agents_overview.input_states.values_mut() {
input.recovered_queue = true;
}
self.pending_startup_thread_start = false;
// Move cached UI state into fresh channels. Old producers retain the old sender/store,
// so their late requests cannot leak into recovery. Background threads attach on selection.
for channel in self.thread_event_channels.values_mut() {
let mut replacement = ThreadEventChannel::new(THREAD_EVENT_CHANNEL_CAPACITY);
replacement.mark_replay_only();
let mut store = std::mem::replace(
&mut *channel.store.lock().await,
ThreadEventStore::new(THREAD_EVENT_CHANNEL_CAPACITY),
);
store.pending_interactive_replay = Default::default();
store.pending_interrupt_turn_id = None;
store.active = false;
store
.buffer
.retain(|event| matches!(event, ThreadBufferedEvent::Notification(_)));
if let Some(input) = store.input_state.as_mut() {
input.recovered_queue = true;
}
*replacement.store.lock().await = store;
*channel = replacement;
}
if let Some(id) = self.current_displayed_thread_id()
&& let Some(mut input) = input.clone()
{
input.recovered_queue = true;
self.agents_overview.input_states.insert(id, input);
}
if let Some(mut started) = thread {
let id = started.session.thread_id;
// A lost settings response can leave the server snapshot behind the user's choice.
// Side conversations can replace the app-wide overrides, so only restore choices
// that still match this conversation's own cached settings.
if (self.current_displayed_thread_id() == Some(id)
|| self.primary_thread_id == Some(id))
&& let Some(channel) = self.thread_event_channels.get(&id)
&& let Some(cached) = channel.store.lock().await.session.as_ref()
{
if let Some(policy) = self.runtime_approval_policy_override
&& policy == cached.approval_policy
{
started.session.approval_policy = policy;
}
if let Some(profile) = &self.runtime_permission_profile_override
&& profile.permission_profile == cached.permission_profile
&& profile.active_permission_profile == cached.active_permission_profile
&& self.config.approvals_reviewer == cached.approvals_reviewer
{
started.session.permission_profile = profile.permission_profile.clone();
started.session.active_permission_profile =
profile.active_permission_profile.clone();
started.session.approvals_reviewer = self.config.approvals_reviewer;
}
}
self.agents_overview.input_states.remove(&id);
if started
.turns
.iter()
.any(|turn| turn.status == TurnStatus::InProgress)
{
self.agent_navigation.mark_running(id);
} else {
self.agent_navigation.mark_stopped(id);
}
if self.primary_thread_id == Some(id) {
self.primary_session_configured = Some(started.session.clone());
}
if started.blocks_direct_input {
self.agent_navigation.mark_parent_owned(id);
}
let channel = ThreadEventChannel::new(THREAD_EVENT_CHANNEL_CAPACITY);
{
let mut store = channel.store.lock().await;
store.set_session(started.session, started.turns);
store.input_state = input.clone();
}
self.thread_event_channels.insert(id, channel);
}
if let Some(id) = displayed
&& let Some((receiver, snapshot)) = self.activate_thread_for_replay(id).await
{
self.active_thread_id = Some(id);
self.active_thread_rx = Some(receiver);
self.recap.seed_from_turns(&snapshot.turns, Instant::now());
self.render_thread_snapshot(
tui, app_server, id, snapshot, /*resume_restored_queue*/ false,
)?;
self.config = self.chat_widget.config_ref().clone();
self.refresh_pending_thread_approvals().await;
if self.thread_unavailable(id) {
self.agent_navigation.mark_stopped(id);
self.chat_widget.pause_unavailable_thread();
self.chat_widget.add_info_message("This conversation is unavailable. Its cached transcript and draft remain here; input is paused. Open the agent picker or return to the parent to continue.".into(), /*hint*/ None);
} else {
self.recap
.schedule_check(id, self.app_event_tx.clone(), Instant::now());
}
} else {
self.active_thread_id = None;
self.active_thread_rx = None;
self.primary_thread_id = None;
self.primary_session_configured = None;
let init = self.chatwidget_init_for_forked_or_resumed_thread(
tui,
self.config.clone(),
/*initial_user_message*/ None,
);
self.replace_chat_widget(ChatWidget::new_with_app_event(init));
self.chat_widget.restore_reconnected_input(input);
}
// Only accept fresh task-tool calls once this connection and its event queue are adopted.
if let ThreadToolTransport::Mcp(server) = app_server.thread_tool_transport() {
server.reconnect(app_server.request_handle(), self.app_event_tx.clone());
}
self.reconnect.offline = false;
self.chat_widget.update_account_state(
bootstrap.status_account_display,
bootstrap.plan_type,
bootstrap.has_chatgpt_account,
matches!(bootstrap.auth_mode, Some(TelemetryAuthMode::Chatgpt)),
);
self.feedback_audience = bootstrap.feedback_audience;
self.chat_widget.add_info_message(
"Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused.".into(), /*hint*/ None,
);
Ok(())
}
}
+30 -20
View File
@@ -487,6 +487,7 @@ impl App {
.agent_navigation
.get(&thread_id)
.is_some_and(|entry| entry.is_closed);
is_replay_only |= self.thread_unavailable(thread_id);
let mut attached_replay_only = false;
if self.should_attach_live_thread_for_selection(thread_id) {
match self
@@ -495,9 +496,7 @@ impl App {
{
Ok(live_attached) => {
attached_replay_only = !live_attached;
if attached_replay_only {
is_replay_only = true;
}
is_replay_only |= attached_replay_only;
}
Err(err) => {
self.chat_widget.add_error_message(format!(
@@ -542,7 +541,6 @@ impl App {
if snapshot.input_state.is_none() {
snapshot.input_state = self.agents_overview.input_states.remove(&thread_id);
}
let blocks_direct_input = self.agent_navigation.is_parent_owned(thread_id);
self.active_thread_id = Some(thread_id);
self.active_thread_rx = Some(receiver);
@@ -558,23 +556,9 @@ impl App {
self.recap
.schedule_check(thread_id, self.app_event_tx.clone(), now);
let init = self.chatwidget_init_for_forked_or_resumed_thread(
tui,
self.config.clone(),
/*initial_user_message*/ None,
);
self.replace_chat_widget(ChatWidget::new_with_app_event(init));
self.chat_widget
.set_task_mentions_enabled(app_server.task_tools_available(thread_id));
self.chat_widget
.note_rendered_width(tui.terminal.last_known_screen_size.width);
if blocks_direct_input {
self.chat_widget.set_parent_owned_thread();
}
self.reset_for_thread_switch(tui)?;
self.replay_thread_snapshot(snapshot, !is_replay_only);
self.render_thread_snapshot(tui, app_server, thread_id, snapshot, !is_replay_only)?;
if is_replay_only {
self.chat_widget.pause_unavailable_thread();
let message = if attached_replay_only {
format!(
"Agent thread {thread_id} could not be resumed live. Replaying saved transcript."
@@ -589,6 +573,32 @@ impl App {
Ok(())
}
pub(super) fn render_thread_snapshot(
&mut self,
tui: &mut tui::Tui,
app_server: &AppServerSession,
thread_id: ThreadId,
snapshot: ThreadEventSnapshot,
resume_restored_queue: bool,
) -> Result<()> {
let init = self.chatwidget_init_for_forked_or_resumed_thread(
tui,
self.config.clone(),
/*initial_user_message*/ None,
);
self.replace_chat_widget(ChatWidget::new_with_app_event(init));
self.chat_widget
.set_task_mentions_enabled(app_server.task_tools_available(thread_id));
self.chat_widget
.note_rendered_width(tui.terminal.last_known_screen_size.width);
if self.agent_navigation.is_parent_owned(thread_id) {
self.chat_widget.set_parent_owned_thread();
}
self.reset_for_thread_switch(tui)?;
self.replay_thread_snapshot(snapshot, resume_restored_queue);
Ok(())
}
pub(super) fn should_attach_live_thread_for_selection(&self, thread_id: ThreadId) -> bool {
!self.thread_event_channels.contains_key(&thread_id)
&& self
+34
View File
@@ -683,6 +683,7 @@ See the Codex keymap documentation for supported actions and examples."
}
let mut listen_for_app_server_events = true;
let mut reconnect = None;
let mut waiting_for_initial_session_configured = wait_for_initial_session_configured;
let mut waiting_for_initial_session_header = true;
@@ -711,6 +712,20 @@ See the Codex keymap documentation for supported actions and examples."
Ok(exit_reason)
} else {
loop {
if app.reconnect.offline
&& !app.reconnect.failed
&& reconnect.is_none()
&& app.reconnect.presentation == reconnect::ReconnectPresentation::Conversation
{
reconnect = Some(Box::pin(reconnect::reconnect(
app.app_server_target.clone(),
app.config.clone(),
app.current_displayed_thread_id(),
app_server.remote_cwd_override().map(Path::to_path_buf),
app_server.thread_tool_transport(),
app.reconnect.presentation,
)));
}
// Replay queues history and operations. A buffered closure must not switch
// widgets before those app events have been applied.
let has_pending_app_events = !app_event_rx.is_empty();
@@ -815,6 +830,25 @@ See the Codex keymap documentation for supported actions and examples."
}
AppRunControl::Continue
}
result = async { match reconnect.as_mut() { Some(future) => future.await, None => std::future::pending().await } }, if reconnect.is_some() && !has_pending_app_events => {
reconnect = None;
match result {
Ok(connected) => {
app.finish_reconnect(tui, &mut app_server, &mut app_event_rx, connected).await?;
listen_for_app_server_events = true;
waiting_for_initial_session_configured = false;
}
Err(error) => {
app.reconnect.failed = true;
app.chat_widget.reconnect_failed();
app.chat_widget.add_error_message(error.to_string());
if let Ok(mut state) = app.agents_overview.view_state.lock() {
state.connection_notice = Some("Reconnect failed — agent list is stale; relaunch to retry");
}
}
}
AppRunControl::Continue
}
() = async {
match app.chat_widget.terminal_title_next_refresh {
Some(deadline) => {
+3
View File
@@ -8633,3 +8633,6 @@ async fn side_backtrack_rejection_reports_unavailable_message_snapshot() {
async fn start_config_write_test_app_server(app: &App) -> Result<AppServerSession> {
Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await
}
#[path = "tests/active_reconnect_tests.rs"]
mod active_reconnect;
@@ -0,0 +1,462 @@
//! Active recovery keeps local input and policy while reusing ordinary thread resume.
use super::*;
use crate::app::reconnect::ReconnectPresentation;
use crate::app::reconnect::reconnect;
use pretty_assertions::assert_eq;
use serde_json::json;
use tokio::net::TcpListener;
use super::disconnect::serve_reconnect_requests;
#[tokio::test]
async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> Result<()> {
for (recovered_queue, edit_offline) in [(true, false), (false, false), (true, true)] {
let (mut app, mut events, mut ops) = make_test_app_with_channels().await;
let id = ThreadId::new();
let cwd = app.config.cwd.clone();
app.config.model = Some("gpt-test".into());
// Avoid platform-specific path widths in the mode-preservation snapshot.
app.config.tui_status_line = Some(vec!["model-with-reasoning".into()]);
app.config
.permissions
.set_permission_profile(PermissionProfile::read_only())?;
app.runtime_approval_policy_override = Some(AskForApproval::OnRequest);
app.runtime_permission_profile_override =
Some(RuntimePermissionProfileOverride::from_config(&app.config));
app.active_thread_id = Some(id);
app.primary_thread_id = Some(id);
let mut thread_session = test_thread_session(id, cwd.to_path_buf());
thread_session.approval_policy = AskForApproval::OnRequest;
thread_session.permission_profile = PermissionProfile::read_only();
app.ensure_thread_channel(id)
.store
.lock()
.await
.set_session(thread_session.clone(), Vec::new());
app.chat_widget.handle_thread_session(thread_session);
app.chat_widget.set_collaboration_mask(
crate::collaboration_modes::plan_mask(app.model_catalog.as_ref()).unwrap(),
);
let expected_mode = app.chat_widget.effective_collaboration_mode().with_updates(
Some("gpt-test".into()),
Some(None),
/*developer_instructions*/ None,
);
let expected_submitted_mode = expected_mode.clone();
if edit_offline {
app.chat_widget
.restore_user_message_to_composer("unacknowledged prompt".into());
app.chat_widget
.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
next_user_turn_op(&mut ops);
}
if recovered_queue {
app.chat_widget
.set_queue_autosend_suppressed(/*suppressed*/ true);
app.chat_widget
.restore_user_message_to_composer("old queued input".into());
app.chat_widget
.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
}
app.chat_widget
.restore_user_message_to_composer("kept draft".into());
let listener = TcpListener::bind("127.0.0.1:0").await?;
let endpoint = crate::resolve_remote_addr(&format!("ws://{}", listener.local_addr()?))?;
app.app_server_target = AppServerTarget::Remote { endpoint };
let thread = json!({
"id": id, "sessionId": id, "preview": "only once", "ephemeral": false,
"modelProvider": "test-provider", "createdAt": 1, "updatedAt": 2,
"status": {"type": "idle"}, "cwd": cwd, "cliVersion": "0.0.0", "source": "cli",
"turns": [test_turn("accepted", TurnStatus::Completed, vec![ThreadItem::UserMessage {
id: "input-1".into(), client_id: None, content: vec![UserInput::Text { text: "only once".into(), text_elements: Vec::new() }],
}])]
});
let server = tokio::spawn(async move {
let mut methods = Vec::new();
for attempt in 0..2 {
let (stream, _) = listener.accept().await?;
methods.extend(serve_reconnect_requests(tokio_tungstenite::accept_async(stream).await?, |request| std::future::ready(match request.method.as_str() {
"thread/resume" if attempt == 0 => Some(json!({"error": {"code": -32603, "message": "temporarily unavailable"}})),
"thread/resume" => {
let params = request.params.as_ref().unwrap();
assert_eq!(params["threadId"], id.to_string());
assert!(params["model"].is_null());
Some(json!({"result": {"thread": thread, "model": "gpt-test", "modelProvider": "test-provider", "cwd": cwd,
"approvalPolicy": "never", "approvalsReviewer": "user", "sandbox": {"type": "dangerFullAccess"}, "reasoningEffort": null}}))
}
"thread/read" => Some(json!({"result": {"thread": thread}})),
"thread/list" | "thread/loaded/list" => Some(json!({"result": {"data": [], "nextCursor": null}})),
"thread/goal/get" => Some(json!({"result": {"goal": null}})),
"turn/start" => {
assert!(!recovered_queue);
let params = request.params.as_ref().unwrap();
assert_eq!(params["input"][0]["text"], "fresh follow-up");
assert_eq!(params["approvalPolicy"], "on-request");
assert_eq!(params["sandboxPolicy"]["type"], "readOnly");
assert_eq!(params["collaborationMode"], json!(expected_submitted_mode));
Some(json!({"result": {"turn": {"id": "fresh", "items": [], "status": "inProgress"}}}))
}
method => panic!("unexpected reconnect request: {method}"),
})).await?);
}
Ok::<_, color_eyre::Report>(methods)
});
let mut session = crate::start_embedded_app_server_for_picker(&app.config).await?;
std::fs::write(
app.config
.codex_home
.join("tui-thread-reference-capabilities"),
"not a directory",
)?;
session.remember_task_tool_thread(id);
let transport = crate::dynamic_tools_mcp::ThreadToolTransport::Mcp(Arc::new(
crate::dynamic_tools_mcp::DynamicToolMcpServer::start(
session.request_handle(),
codex_app_server_protocol::ThreadStartParams::default(),
app.app_event_tx.clone(),
app.dynamic_tool_status_updates.clone(),
/*managed_requirement*/ None,
)
.await?,
));
let mut overrides = None;
transport.configure_mcp(&mut overrides);
let mcp = overrides.unwrap().remove("mcp_servers.codex_tui").unwrap();
let client = codex_http_client::HttpClientBuilder::new().build_direct()?;
let call_mcp = || {
client
.post(mcp["url"].as_str().unwrap())
.header(
"Authorization",
mcp["http_headers"]["Authorization"].as_str().unwrap(),
)
.header("Accept", "application/json, text/event-stream")
.header("MCP-Protocol-Version", "2026-07-28")
.header("MCP-Method", "tools/call")
.header("MCP-Name", "list_threads")
.json(
&json!({"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {
"name": "list_threads", "arguments": {}, "_meta": {"threadId": id,
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientCapabilities": {}}
}}),
)
};
app.chat_widget.remote_connection =
crate::status::remote_connection::remote_connection_status_value(
&app.app_server_target,
Some("1.0.0"),
);
let mut tui = crate::tui::test_support::make_test_tui()?;
app.begin_reconnect();
if edit_offline {
app.handle_tui_event(
&mut tui,
&mut session,
TuiEvent::Key(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT)),
)
.await?;
assert!(!app.chat_widget.has_queued_follow_up_messages());
assert!(
!app.chat_widget
.capture_thread_input_state()
.unwrap()
.recovered_queue
);
}
let stale_picker_refresh = app.agent_navigation.begin_picker_refresh(id).unwrap();
app.last_subagent_backfill_attempt = Some(id);
app.rate_limit_refresh_state
.start(
RateLimitRefreshOrigin::Recovery,
&mut app.rate_limit_hard_stop_generation,
)
.unwrap();
let before_disconnect = Instant::now() - Duration::from_secs(/*secs*/ 300);
app.recap.note_focus_lost(before_disconnect);
for _ in 0..3 {
app.recap
.note_turn_finished(&TurnStatus::Completed, before_disconnect);
}
app.recap
.schedule_check(id, app.app_event_tx.clone(), Instant::now());
let old_sender = app.app_event_tx.clone();
let connected = reconnect(
app.app_server_target.clone(),
app.config.clone(),
Some(id),
/*remote_cwd*/ None,
transport,
ReconnectPresentation::Conversation,
)
.await?;
let paused = call_mcp().send().await?.text().await?;
assert!(
paused.contains("TUI is reconnecting; tool was not sent"),
"{paused}"
);
app.finish_reconnect(&mut tui, &mut session, &mut events, connected)
.await?;
assert!(!app.reconnect.offline);
assert_eq!(app.last_subagent_backfill_attempt, None);
assert!(
app.rate_limit_refresh_state
.start(
RateLimitRefreshOrigin::Recovery,
&mut app.rate_limit_hard_stop_generation
)
.is_some()
);
assert!(app.agent_navigation.begin_picker_refresh(id).is_some());
assert!(
!app.agent_navigation
.finish_picker_refresh(id, stale_picker_refresh)
);
let mut deferred = Vec::new();
tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async {
loop {
let event = events.recv().await.expect("rebound recap event");
if matches!(event, AppEvent::CheckRecap { thread_id } if thread_id == id) {
break;
}
deferred.push(event);
}
})
.await?;
for event in deferred {
app.app_event_tx.send(event);
}
let ready = call_mcp().send().await?.text().await?;
assert!(ready.contains("threads"), "{ready}");
assert_eq!(
app.chat_widget.remote_connection,
crate::status::remote_connection::remote_connection_status_value(
&app.app_server_target,
Some("2.0.0")
)
);
assert!(old_sender.app_event_tx.is_closed());
assert!(session.task_tools_available(id));
assert_eq!(app.current_displayed_thread_id(), Some(id));
assert_eq!(
app.chat_widget.effective_collaboration_mode(),
expected_mode
);
assert_eq!(
app.chat_widget.composer_text_with_pending(),
if edit_offline {
"old queued input"
} else {
"kept draft"
}
);
let history = drain_history(&mut app, &mut tui, &mut session, &mut events).await?;
assert_eq!(history.matches("only once").count(), 1);
if recovered_queue {
assert!(app.chat_widget.has_queued_follow_up_messages());
assert!(!app.chat_widget.maybe_send_next_queued_input());
if edit_offline {
app.handle_tui_event(
&mut tui,
&mut session,
TuiEvent::Key(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT)),
)
.await?;
assert_eq!(
app.chat_widget.composer_text_with_pending(),
"unacknowledged prompt"
);
assert!(ops.try_recv().is_err());
} else {
assert_snapshot!(
"restored_conversation",
render_bottom_popup(&app.chat_widget, /*width*/ 80)
);
}
} else {
app.chat_widget
.handle_key_event(KeyEvent::new(KeyCode::Char('u'), KeyModifiers::CONTROL));
app.chat_widget
.restore_user_message_to_composer("fresh follow-up".into());
app.chat_widget
.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
while let Ok(event) = events.try_recv() {
app.handle_event(&mut tui, &mut session, event).await?;
}
}
session.shutdown().await?;
let methods = server.await??;
assert_eq!(
methods
.iter()
.filter(|method| *method == "turn/start")
.count(),
usize::from(!recovered_queue)
);
}
Ok(())
}
#[tokio::test]
async fn reconnect_exhaustion_and_unknown_initial_thread_stay_offline() -> Result<()> {
let (mut app, _, _) = make_test_app_with_channels().await;
let listener = TcpListener::bind("127.0.0.1:0").await?;
app.app_server_target = AppServerTarget::Remote {
endpoint: crate::resolve_remote_addr(&format!("ws://{}", listener.local_addr()?))?,
};
drop(listener);
tokio::time::pause();
let start = tokio::time::Instant::now();
for id in [Some(ThreadId::new()), None] {
assert!(
reconnect(
app.app_server_target.clone(),
app.config.clone(),
id,
/*remote_cwd*/ None,
crate::dynamic_tools_mcp::ThreadToolTransport::Dynamic,
ReconnectPresentation::Conversation
)
.await
.is_err()
);
}
assert!((15..=65).contains(&start.elapsed().as_secs()));
app.begin_reconnect();
app.chat_widget.reconnect_failed();
assert_snapshot!(
"reconnect_failed",
render_bottom_popup(&app.chat_widget, /*width*/ 80)
);
tokio::time::resume();
Ok(())
}
#[tokio::test]
async fn reconnect_allows_slow_hydration_but_bounds_a_stalled_server() -> Result<()> {
for delay in [15, 150] {
let (mut app, mut events, _) = make_test_app_with_channels().await;
app.config.model = Some("gpt-test".into());
let listener = TcpListener::bind("127.0.0.1:0").await?;
let endpoint = crate::RemoteAppServerEndpoint::WebSocket {
websocket_url: format!("ws://{}", listener.local_addr()?),
auth_token: None,
};
let id = ThreadId::new();
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await?;
serve_reconnect_requests(
tokio_tungstenite::accept_async(stream).await?,
move |request| async move {
assert_eq!(request.method, "thread/resume");
tokio::time::pause();
tokio::time::sleep(Duration::from_secs(delay)).await;
tokio::time::resume();
Some(json!({"error": {"code": -32600, "message": "thread no longer exists"}}))
},
)
.await
});
let start = tokio::time::Instant::now();
let result = reconnect(
AppServerTarget::Remote {
endpoint: endpoint.clone(),
},
app.config.clone(),
Some(id),
/*remote_cwd*/ None,
crate::dynamic_tools_mcp::ThreadToolTransport::Dynamic,
ReconnectPresentation::Conversation,
)
.await;
let elapsed = start.elapsed().as_secs();
if delay == 15 {
app.app_server_target = AppServerTarget::Remote { endpoint };
app.active_thread_id = Some(id);
app.primary_thread_id = Some(id);
let cached = test_thread_session(id, app.config.cwd.to_path_buf());
app.ensure_thread_channel(id)
.store
.lock()
.await
.set_session(cached, Vec::new());
app.chat_widget
.restore_user_message_to_composer("unavailable draft".into());
app.begin_reconnect();
let mut session = crate::start_embedded_app_server_for_picker(&app.config).await?;
let mut tui = crate::tui::test_support::make_test_tui()?;
app.finish_reconnect(&mut tui, &mut session, &mut events, result?)
.await?;
assert!(app.thread_unavailable(id));
app.handle_tui_event(
&mut tui,
&mut session,
TuiEvent::Key(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)),
)
.await?;
assert_eq!(
app.chat_widget.composer_text_with_pending(),
"unavailable draft"
);
let history = drain_history(&mut app, &mut tui, &mut session, &mut events).await?;
let history = &history[history.find("• This conversation is unavailable").unwrap()..];
assert_snapshot!(
"unavailable_conversation",
format!(
"{history}\n{}",
render_bottom_popup(&app.chat_widget, /*width*/ 80)
)
.replace(
&test_path_buf("/tmp/project").display().to_string(),
"/tmp/project"
)
);
session.shutdown().await?;
assert!((15..120).contains(&elapsed), "{elapsed}");
let methods = server.await??;
assert_eq!(
methods
.iter()
.filter(|method| *method == "initialize")
.count(),
1
);
} else {
tokio::time::resume();
assert!(result.is_err());
assert_eq!(elapsed, 120);
server.abort();
}
}
Ok(())
}
pub(super) async fn drain_history(
app: &mut App,
tui: &mut tui::Tui,
session: &mut AppServerSession,
events: &mut mpsc::UnboundedReceiver<AppEvent>,
) -> Result<String> {
while let Ok(event) = events.try_recv() {
assert!(!matches!(
event,
AppEvent::CodexOp(AppCommand::UserTurn { .. })
));
if matches!(
event,
AppEvent::InsertHistoryCell(_)
| AppEvent::BeginThreadSwitchHistoryReplayBuffer
| AppEvent::EndInitialHistoryReplayBuffer
) {
app.handle_event(tui, session, event).await?;
}
}
Ok(lines_to_single_string(
&app.transcript_cells
.iter()
.flat_map(|cell| cell.display_lines(/*width*/ 80))
.collect::<Vec<_>>(),
))
}
@@ -264,7 +264,7 @@ where
methods.push(request.method.clone());
let request_id = request.id.clone();
let response = match request.method.as_str() {
"initialize" => Some(json!({"result": {"userAgent": "reconnect-test"}})),
"initialize" => Some(json!({"result": {"userAgent": "reconnect-test/2.0.0"}})),
"account/read" => {
Some(json!({"result": {"account": null, "requiresOpenaiAuth": false}}))
}
@@ -601,7 +601,7 @@ async fn archive_current_thread_reports_success_only_after_archiving() -> Result
#[tokio::test]
async fn local_daemon_registers_approval_gated_mcp_tools_for_both_start_paths() -> Result<()> {
let (mut app, mut events, _ops) = make_test_app_with_channels().await;
let (mut app, events, _ops) = make_test_app_with_channels().await;
let codex_home = tempdir()?;
app.config.codex_home = codex_home.path().to_path_buf().abs();
app.config.sqlite = SqliteConfig::new_for_testing(codex_home.path().abs());
@@ -612,7 +612,7 @@ async fn local_daemon_registers_approval_gated_mcp_tools_for_both_start_paths()
codex_home.path().join("config.toml"),
"web_search = \"disabled\"\n",
)?;
let (mut app_server, requests, proxy) = start_recording_app_server(
let (mut app_server, mut requests, mut proxy) = start_recording_app_server(
&app.config,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ None,
@@ -776,6 +776,41 @@ async fn local_daemon_registers_approval_gated_mcp_tools_for_both_start_paths()
}
}))
};
let transport = app_server.thread_tool_transport();
let crate::dynamic_tools_mcp::ThreadToolTransport::Mcp(tool_server) = &transport else {
panic!("expected the daemon task-tool bridge");
};
tool_server.suspend();
let paused = call_tool(0, "list_threads", serde_json::json!({}))
.send()
.await?
.text()
.await?;
assert!(
paused.contains("TUI is reconnecting; tool was not sent"),
"{paused}"
);
let (replacement, replacement_requests, replacement_proxy) = start_recording_app_server(
&app.config,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ None,
)
.await?;
let (new_tx, new_rx) = mpsc::unbounded_channel();
let new_sender = AppEventSender::new(new_tx);
drop(events);
let mut events = new_rx;
assert!(app.app_event_tx.app_event_tx.is_closed());
tool_server.reconnect(replacement.request_handle(), new_sender);
let previous = std::mem::replace(
&mut app_server,
replacement.with_thread_tool_transport(transport),
);
previous.shutdown().await?;
proxy.await??;
requests = replacement_requests;
proxy = replacement_proxy;
// The same MCP URL and credentials now use the new connection and event receiver.
let response = call_tool(1, "list_threads", serde_json::json!({}))
.send()
.await?;
@@ -824,9 +859,13 @@ async fn local_daemon_registers_approval_gated_mcp_tools_for_both_start_paths()
child["config"]["mcp_servers.codex_tui"],
starts[0]["config"]["mcp_servers.codex_tui"]
);
let forked = call_tool(3, "fork_thread", serde_json::json!({"threadId": thread_id}))
.send()
.await?;
let forked = call_tool(
3,
"fork_thread",
serde_json::json!({"threadId": delegation_source}),
)
.send()
.await?;
assert!(forked.status().is_success());
let forked = recorded_params(&requests, "thread/fork")
.pop()
@@ -0,0 +1,10 @@
---
source: tui/src/app/tests/active_reconnect_tests.rs
expression: "render_bottom_popup(&app.chat_widget, 80)"
---
• Reconnect failed — check the endpoint, then relaunch (0s)
› Ask Codex to do anything
ctrl+c quit
@@ -0,0 +1,11 @@
---
source: tui/src/app/tests/active_reconnect_tests.rs
expression: "render_bottom_popup(&app.chat_widget, 80)"
---
• Queued follow-up inputs
↳ old queued input
⌥ + ↑ edit last queued message
› kept draft
gpt-test default Plan mode (shift+tab to cycle)
@@ -0,0 +1,9 @@
---
source: tui/src/app/tests/active_reconnect_tests.rs
expression: "format!(\"{history}\\n{}\", render_bottom_popup(&app.chat_widget, 80))"
---
• This conversation is unavailable. Its cached transcript and draft remain here; input is paused. Open the agent picker or return to the parent to continue.
• Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused.
› unavailable draft
gpt-test default · /tmp/project
@@ -2,8 +2,8 @@
source: tui/src/app/tests/disconnect_tests.rs
expression: "format!(\"{}\\n{}\",\nlines_to_single_string(&app.transcript_cells.last().unwrap().display_lines(80)),\nrender_bottom_popup(&app.chat_widget, 80))"
---
■ Connection lost.
• Connection lost — copy your draft, then relaunch (0s)
■ Connection lost. Attempting to reconnect…
• Reconnecting to app-server… (0s)
• Queued follow-up inputs
↳ queued before disconnect
+16 -1
View File
@@ -443,6 +443,12 @@ impl App {
thread_id: ThreadId,
op: AppCommand,
) -> Result<()> {
if self.thread_unavailable(thread_id) {
self.chat_widget.add_error_message(
"This conversation is unavailable; no operation was sent.".into(),
);
return Ok(());
}
if self.chat_widget.rejects_misalignment_policy_op(&op) {
return Ok(());
}
@@ -1546,7 +1552,7 @@ impl App {
pub(super) fn replay_thread_snapshot(
&mut self,
snapshot: ThreadEventSnapshot,
mut snapshot: ThreadEventSnapshot,
resume_restored_queue: bool,
) {
let request_changes = snapshot
@@ -1591,6 +1597,12 @@ impl App {
self.chat_widget.handle_thread_session(session);
}
}
let recovered_input = snapshot
.input_state
.as_ref()
.is_some_and(|input| input.recovered_queue)
.then(|| snapshot.input_state.take())
.flatten();
self.chat_widget.restore_thread_input_state(
snapshot.input_state,
ThreadInputStateRestoreMode {
@@ -1616,6 +1628,9 @@ impl App {
self.app_event_tx
.send(AppEvent::EndInitialHistoryReplayBuffer);
}
if recovered_input.is_some() {
self.chat_widget.restore_reconnected_input(recovered_input);
}
self.chat_widget
.set_queue_autosend_suppressed(/*suppressed*/ false);
self.chat_widget
+12
View File
@@ -464,6 +464,13 @@ impl AppServerSession {
}
}
pub(crate) fn with_thread_tool_transport(mut self, transport: ThreadToolTransport) -> Self {
if let ThreadToolTransport::Mcp(server) = transport {
self.dynamic_tool_mcp = Some(server);
}
self
}
pub(crate) fn with_remote_cwd_override(mut self, remote_cwd_override: Option<PathBuf>) -> Self {
self.remote_cwd_override = remote_cwd_override;
self
@@ -481,6 +488,11 @@ impl AppServerSession {
matches!(&self.client, AppServerClient::InProcess(_))
}
/// Carry capabilities that may exist only in memory when the optional cache is unwritable.
pub(crate) fn inherit_task_tool_capabilities(&mut self, previous: &Self) {
self.task_tool_threads.extend(&previous.task_tool_threads);
}
pub(crate) fn task_tools_available(&self, thread_id: ThreadId) -> bool {
self.task_tool_threads.contains(&thread_id)
|| self
+47 -2
View File
@@ -13,9 +13,47 @@ impl ChatWidget {
self.bottom_pane.ensure_status_indicator();
self.bottom_pane
.set_interrupt_hint_visible(/*visible*/ false);
self.set_status_header("Connection lost — copy your draft, then relaunch".to_string());
self.set_status_header("Reconnecting to app-server…".to_string());
self.set_footer_hint_override(Some(vec![("ctrl+c".into(), "quit".into())]));
self.add_error_message("Connection lost.".into());
self.add_error_message("Connection lost. Attempting to reconnect…".into());
}
/// Restore local input only after replay, which can otherwise move interrupted queues into the draft.
pub(crate) fn restore_reconnected_input(&mut self, input: Option<ThreadInputState>) {
let running = self.turn_lifecycle.agent_turn_running;
if let Some(mut input) = input {
// Its acceptance is unknown. Keep a local copy for manual recovery without
// comparing against partial history or automatically submitting it again.
if input.user_turn_pending_start
&& let Some(prompt) = input.safety_buffering_prompt.take()
{
input.queued_user_messages.push_front(prompt.into());
input
.queued_user_message_history_records
.push_front(UserMessageHistoryRecord::UserMessageText);
input.recovered_queue = true;
}
input.current_collaboration_mode = self.current_collaboration_mode.clone();
// Resume supplies model/effort, but not the user's selected collaboration mode.
if let Some(mask) = input.active_collaboration_mask.as_mut() {
mask.model = Some(self.current_model().to_string());
mask.reasoning_effort = Some(self.effective_reasoning_effort());
}
self.restore_thread_input_state(
Some(input),
ThreadInputStateRestoreMode {
preserve_in_flight_turn: false,
},
);
}
self.turn_lifecycle.restore_running(running, Instant::now());
self.update_task_running_state();
}
pub(crate) fn pause_unavailable_thread(&mut self) {
self.turn_lifecycle
.restore_running(/*running*/ false, Instant::now());
self.update_task_running_state();
}
pub(crate) fn handle_disconnected_view_key(&mut self, key: KeyEvent) {
@@ -41,4 +79,11 @@ impl ChatWidget {
self.bottom_pane.handle_disconnected_key(key);
}
}
pub(crate) fn reconnect_failed(&mut self) {
self.set_status_header("Reconnect failed — check the endpoint, then relaunch".into());
self.add_error_message(
"Automatic reconnect could not restore this session. Your draft is still editable. Copy it before quitting with Ctrl+C, then reconnect with the same command.".into(),
);
}
}
@@ -2413,3 +2413,40 @@ async fn interrupt_prepends_queued_messages_before_existing_composer_text() {
let _ = drain_insert_history(&mut rx);
}
#[tokio::test]
async fn reconnect_holds_only_recovered_input_until_manually_edited() {
for (recovered, pending_start) in [
(None, false),
(Some("review this old input"), false),
(Some("unacknowledged prompt"), true),
] {
let (mut chat, _rx, _ops) = make_chatwidget_manual(/*model_override*/ None).await;
if let Some(text) = recovered {
if pending_start {
chat.input_queue.user_turn_pending_start = true;
chat.safety_buffering_prompt = Some(UserMessage::from(text));
} else {
chat.input_queue
.queued_user_messages
.push_back(UserMessage::from(text).into());
}
}
chat.pause_for_disconnect();
let input = chat.capture_thread_input_state();
let (mut chat, _rx, mut ops) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
chat.restore_reconnected_input(input);
chat.set_queue_autosend_suppressed(/*suppressed*/ false);
if let Some(text) = recovered {
assert!(!chat.maybe_send_next_queued_input());
assert_eq!(chat.pop_latest_queued_composer_state().unwrap().text, text);
assert_no_submit_op(&mut ops);
}
chat.input_queue
.queued_user_messages
.push_back(UserMessage::from("new follow-up").into());
assert!(chat.maybe_send_next_queued_input());
assert_matches!(next_submit_op(&mut ops), Op::UserTurn { .. });
}
}
+1
View File
@@ -616,6 +616,7 @@ async fn execute_inner(
thread_start_params.sandbox = None;
None
} else {
thread_start_params.permissions = None;
thread_start_params.sandbox = Some(match &source.sandbox {
SandboxPolicy::DangerFullAccess => SandboxMode::DangerFullAccess,
SandboxPolicy::ReadOnly { .. } => SandboxMode::ReadOnly,
+34 -6
View File
@@ -43,6 +43,8 @@ use serde_json::json;
use std::borrow::Cow;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::PoisonError;
use std::sync::RwLock;
use tokio::net::TcpListener;
use tokio::sync::broadcast;
use tokio::task::JoinHandle;
@@ -79,12 +81,29 @@ impl ThreadToolTransport {
}
}
type ToolConnection = Arc<RwLock<Option<(AppServerRequestHandle, AppEventSender)>>>;
pub(crate) struct DynamicToolMcpServer {
connection: ToolConnection,
config: Value,
task: JoinHandle<()>,
}
impl DynamicToolMcpServer {
pub(crate) fn suspend(&self) {
*self
.connection
.write()
.unwrap_or_else(PoisonError::into_inner) = None;
}
pub(crate) fn reconnect(&self, handle: AppServerRequestHandle, events: AppEventSender) {
*self
.connection
.write()
.unwrap_or_else(PoisonError::into_inner) = Some((handle, events));
}
pub(crate) async fn start(
request_handle: AppServerRequestHandle,
mut thread_start_params: ThreadStartParams,
@@ -122,10 +141,10 @@ impl DynamicToolMcpServer {
if let Some(overrides) = thread_start_params.config.as_mut() {
overrides.remove("web_search");
}
let connection = Arc::new(RwLock::new(Some((request_handle, app_event_tx))));
let handler = DynamicToolMcpHandler {
request_handle,
connection: Arc::clone(&connection),
thread_start_params,
app_event_tx,
status_updates,
server_config: server_config.clone(),
};
@@ -147,6 +166,7 @@ impl DynamicToolMcpServer {
}
});
Ok(Self {
connection,
config: server_config,
task,
})
@@ -177,9 +197,8 @@ async fn require_authorization(
#[derive(Clone)]
struct DynamicToolMcpHandler {
request_handle: AppServerRequestHandle,
connection: ToolConnection,
thread_start_params: ThreadStartParams,
app_event_tx: AppEventSender,
status_updates: broadcast::Sender<ThreadStatusChangedNotification>,
server_config: Value,
}
@@ -265,12 +284,21 @@ impl ServerHandler for DynamicToolMcpHandler {
format!("mcp_servers.{}", dynamic_tools::NAMESPACE),
self.server_config.clone(),
);
// Snapshot once: an in-flight call must never switch connections or replay a mutation.
let (request_handle, app_event_tx) = self
.connection
.read()
.unwrap_or_else(PoisonError::into_inner)
.clone()
.ok_or_else(|| {
McpError::internal_error("TUI is reconnecting; tool was not sent", None)
})?;
let response = dynamic_tools::execute(
self.request_handle.clone(),
request_handle,
params,
thread_start_params,
self.status_updates.subscribe(),
Some(&self.app_event_tx),
Some(&app_event_tx),
)
.await;
let content = response
+32 -6
View File
@@ -1,4 +1,4 @@
//! Drives disconnect handling through the real binary and terminal event loop.
//! Drives automatic reconnect through the real binary and terminal event loop.
use super::focus_palette::PtyCodex;
use super::focus_palette::write_test_config;
@@ -15,7 +15,7 @@ use tokio::net::UnixListener;
use tokio_tungstenite::tungstenite::Message;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn disconnect_preserves_editable_draft_in_terminal() -> Result<()> {
async fn automatic_reconnect_restores_draft_and_routes_new_notifications() -> Result<()> {
let repo_root = codex_utils_cargo_bin::repo_root()?;
// macOS's default temporary directory leaves too little room for the control socket path.
let codex_home = tempfile::tempdir_in("/tmp")?;
@@ -24,9 +24,11 @@ async fn disconnect_preserves_editable_draft_in_terminal() -> Result<()> {
std::fs::create_dir_all(socket.parent().unwrap())?;
let listener = UnixListener::bind(socket.as_path())?;
let (disconnect_tx, mut disconnect_rx) = tokio::sync::oneshot::channel();
let (restore_tx, restore_rx) = tokio::sync::oneshot::channel();
let server_cwd = repo_root.clone();
let server = tokio::spawn(async move {
let mut methods = Vec::new();
let mut restore_rx = Some(restore_rx);
let id = "00000000-0000-0000-0000-000000000001";
let thread = json!({
"id": id, "sessionId": id, "preview": "", "ephemeral": false,
@@ -34,7 +36,7 @@ async fn disconnect_preserves_editable_draft_in_terminal() -> Result<()> {
"status": {"type": "active", "activeFlags": []}, "cwd": server_cwd,
"cliVersion": "0.0.0", "source": "cli", "turns": [{"id": "running", "items": [], "status": "inProgress", "error": null}]
});
for connection in 0..1 {
for connection in 0..2 {
let mut socket = loop {
let (stream, _) = listener.accept().await?;
// Startup probes the default daemon socket before opening its WebSocket.
@@ -57,6 +59,9 @@ async fn disconnect_preserves_editable_draft_in_terminal() -> Result<()> {
continue;
};
methods.push(request.method.clone());
if connection == 1 && request.method == "initialize" {
restore_rx.take().unwrap().await?;
}
let result = match request.method.as_str() {
"initialize" => json!({"userAgent": "reconnect-pty"}),
"account/read" => {
@@ -92,6 +97,22 @@ async fn disconnect_preserves_editable_draft_in_terminal() -> Result<()> {
.into(),
))
.await?;
if connection == 1 && request.method == "thread/resume" {
// Keep the recovered turn running: its output must appear without waiting
// for turn/completed or rebuilding the transcript.
socket
.send(Message::Text(
json!({
"method": "item/agentMessage/delta", "params": {
"threadId": id, "turnId": "running", "itemId": "live-item",
"delta": "fresh-notification-after-reconnect\n"
}
})
.to_string()
.into(),
))
.await?;
}
}
}
Ok::<_, anyhow::Error>(methods)
@@ -99,11 +120,13 @@ async fn disconnect_preserves_editable_draft_in_terminal() -> Result<()> {
let mut terminal = PtyCodex::start(&repo_root, codex_home)?;
terminal.wait_for_startup()?;
let mut disconnect_tx = Some(disconnect_tx);
let mut restore_tx = Some(restore_tx);
for expected in [
"gpt-5.6-terra",
"preserved-draft",
"Connection lost",
"Reconnecting",
"preserved-draft!",
"fresh-notification-after-reconnect",
] {
let deadline = Instant::now() + Duration::from_secs(/*secs*/ 30);
while !terminal.screen_contains(expected) && Instant::now() < deadline {
@@ -119,7 +142,10 @@ async fn disconnect_preserves_editable_draft_in_terminal() -> Result<()> {
"preserved-draft" => {
disconnect_tx.take().unwrap().send(()).unwrap();
}
"Connection lost" => terminal.write_input(b"!")?,
"Reconnecting" => terminal.write_input(b"!")?,
"preserved-draft!" => {
restore_tx.take().unwrap().send(()).unwrap();
}
_ => {}
}
}
@@ -134,7 +160,7 @@ async fn disconnect_preserves_editable_draft_in_terminal() -> Result<()> {
.iter()
.filter(|method| *method == "thread/resume")
.count(),
0
1
);
assert!(!methods.iter().any(|method| method == "turn/start"));
Ok(())