fix codex fork resume review feedback

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::<LocalThreadStore>()
+        {
             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<Option<Vec<RolloutItem>>> {
     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);
     }
 

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
         ));

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<S>(
         &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,