From 3fe8c6376fdc228a8f00074acb6c2b6dc0a1fa57 Mon Sep 17 00:00:00 2001 From: anais Date: Thu, 18 Jun 2026 16:48:36 -0700 Subject: [PATCH] fix codex fork resume review feedback --- .../request_processors/thread_processor.rs | 44 +++++++++++-------- .../thread_processor_tests.rs | 32 ++++++++++++-- codex-rs/core/src/thread_manager.rs | 2 + 3 files changed, 57 insertions(+), 21 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index ae58cde722..e5cb3662dc 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2523,6 +2523,10 @@ impl ThreadRequestProcessor { let Some(rollout_path) = stored_thread.rollout_path else { return Ok(None); }; + match codex_rollout::read_session_meta_line(rollout_path.as_path()).await { + Ok(meta_line) if meta_line.meta.id == thread_id => {} + _ => return Ok(None), + } try_build_recent_not_loaded_turns_page_from_rollout_path( rollout_path.as_path(), loaded_status, @@ -3623,7 +3627,10 @@ impl ThreadRequestProcessor { "`permissions` cannot be combined with `sandbox`", )); } - let optimized_source = if !include_turns && !ephemeral { + let optimized_source = if !include_turns + && !ephemeral + && self.thread_store.as_any().is::() + { self.read_stored_thread_for_resume_with_compacted_history(&thread_id, path.as_ref()) .await? } else { @@ -3886,6 +3893,17 @@ impl ThreadRequestProcessor { let notif = thread_started_notification(thread); let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); + log_listener_attach_result( + self.ensure_conversation_listener( + thread_id, + connection_id, + /*raw_events_enabled*/ false, + ) + .await, + thread_id, + connection_id, + "thread", + ); self.outgoing.send_response(request_id, response).await; // `excludeTurns` is the cheap fork path, so skip restored usage replay // instead of rebuilding history only to attribute a historical update. @@ -3912,22 +3930,6 @@ impl ThreadRequestProcessor { .await; } - tokio::task::yield_now().await; - // Auto-attach after the response is queued. The attach is still - // performed before the follow-up notification, but it no longer sits on - // the fork response latency path. - log_listener_attach_result( - self.ensure_conversation_listener( - thread_id, - connection_id, - /*raw_events_enabled*/ false, - ) - .await, - thread_id, - connection_id, - "thread", - ); - self.outgoing .send_server_notification(ServerNotification::ThreadStarted(notif)) .await; @@ -4321,6 +4323,7 @@ fn read_compacted_resume_suffix_items( ) -> std::io::Result>> { let mut suffix_items = Vec::new(); let mut found_checkpoint = false; + let mut found_checkpoint_turn_context = false; let mut should_fallback = false; read_rollout_lines_from_end(path, byte_len, |line| { @@ -4341,15 +4344,20 @@ fn read_compacted_resume_suffix_items( return Ok(false); } let is_checkpoint = rollout_item_is_reconstruction_checkpoint(&rollout_line.item); + let is_turn_context = matches!(&rollout_line.item, RolloutItem::TurnContext(_)); suffix_items.push(rollout_line.item); if is_checkpoint { found_checkpoint = true; + return Ok(true); + } + if found_checkpoint && is_turn_context { + found_checkpoint_turn_context = true; return Ok(false); } Ok(true) })?; - if should_fallback || !found_checkpoint { + if should_fallback || !found_checkpoint || !found_checkpoint_turn_context { return Ok(None); } diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index 00d2ea523b..bf749e1907 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -361,6 +361,29 @@ mod thread_processor_behavior_tests { )) } + fn compacted_resume_test_turn_context() -> RolloutItem { + RolloutItem::TurnContext(codex_protocol::protocol::TurnContextItem { + turn_id: Some("turn-1".to_string()), + cwd: test_path_buf("/tmp/codex-compacted-resume").abs(), + workspace_roots: None, + current_date: None, + timezone: None, + approval_policy: AskForApproval::Never, + sandbox_policy: codex_protocol::protocol::SandboxPolicy::new_read_only_policy(), + permission_profile: None, + network: None, + file_system_sandbox_policy: None, + model: "test-model".to_string(), + comp_hash: None, + personality: None, + collaboration_mode: None, + multi_agent_version: None, + realtime_active: None, + effort: None, + summary: codex_protocol::config_types::ReasoningSummary::Auto, + }) + } + fn compacted_resume_test_message( role: &str, text: &str, @@ -379,6 +402,7 @@ mod thread_processor_behavior_tests { role: role.to_string(), content, phase: None, + metadata: None, } } @@ -426,6 +450,7 @@ mod thread_processor_behavior_tests { RolloutItem::ResponseItem(compacted_resume_test_message("user", "first user response")), compacted_resume_test_compaction("older checkpoint", /*window_id*/ 1), compacted_resume_test_user_event("middle user event"), + compacted_resume_test_turn_context(), compacted_resume_test_compaction("latest checkpoint", /*window_id*/ 2), compacted_resume_test_agent_event(&large_tail_message), ])?; @@ -435,19 +460,20 @@ mod thread_processor_behavior_tests { let items = read_compacted_resume_rollout_items(path.as_path(), thread_id, byte_len)? .expect("compacted resume items"); - assert_eq!(items.len(), 5); + assert_eq!(items.len(), 6); assert!(matches!(items[0], RolloutItem::SessionMeta(_))); assert!(matches!( items[1], RolloutItem::EventMsg(EventMsg::UserMessage(_)) )); assert!(matches!(items[2], RolloutItem::ResponseItem(_))); + assert!(matches!(items[3], RolloutItem::TurnContext(_))); assert_eq!( - compacted_resume_test_replacement_text(&items[3]), + compacted_resume_test_replacement_text(&items[4]), Some("latest checkpoint") ); assert!(matches!( - &items[4], + &items[5], RolloutItem::EventMsg(EventMsg::AgentMessage(event)) if event.message == large_tail_message )); diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 7b5489b7a4..98a81415a6 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -961,6 +961,7 @@ impl ThreadManager { .await } + #[allow(clippy::too_many_arguments)] pub async fn fork_thread_from_history_with_initial_rollout_copy( &self, snapshot: S, @@ -986,6 +987,7 @@ impl ThreadManager { .await } + #[allow(clippy::too_many_arguments)] async fn fork_thread_with_initial_history( &self, snapshot: ForkSnapshot,