[app-server] improve thread list and resume rpc paths

codex-rs/app-server/src/request_processors.rs

@@ -530,8 +530,11 @@ use crate::thread_state::ConnectionCapabilities;
 use crate::thread_state::ThreadListenerCommand;
 use crate::thread_state::ThreadState;
 use crate::thread_state::ThreadStateManager;
+use token_usage_replay::TokenUsageTurnOwner;
+use token_usage_replay::latest_token_usage_turn_id_from_owner;
 use token_usage_replay::latest_token_usage_turn_id_from_rollout_items;
 use token_usage_replay::send_thread_token_usage_update_to_connection;
+use token_usage_replay::token_usage_turn_owner_for_rollout_item;
 
 fn resolve_request_cwd(cwd: Option<PathBuf>) -> Result<Option<AbsolutePathBuf>, JSONRPCErrorError> {
     cwd.map(|cwd| {
@@ -606,11 +609,23 @@ pub(crate) use self::thread_summary::thread_settings_from_config_snapshot;
 pub(crate) use self::thread_summary::thread_settings_from_core_snapshot;
 
 pub(crate) fn build_api_turns_from_rollout_items(items: &[RolloutItem]) -> Vec<Turn> {
+    build_api_turns_from_rollout_items_with_token_usage_owner(items).0
+}
+
+fn build_api_turns_from_rollout_items_with_token_usage_owner(
+    items: &[RolloutItem],
+) -> (Vec<Turn>, Option<TokenUsageTurnOwner>) {
     let mut builder = ThreadHistoryBuilder::new();
+    let mut token_usage_turn_owner = None;
+
     for item in items {
+        if let Some(owner) = token_usage_turn_owner_for_rollout_item(&builder, item) {
+            token_usage_turn_owner = Some(owner);
+        }
         if is_persisted_rollout_item(item) {
             builder.handle_rollout_item(item);
         }
     }
-    builder.finish()
+
+    (builder.finish(), token_usage_turn_owner)
 }

codex-rs/app-server/src/request_processors/thread_lifecycle.rs

@@ -558,13 +558,15 @@ pub(super) async fn handle_pending_thread_resume_request(
     let request_id = pending.request_id;
     let connection_id = request_id.connection_id;
     let mut thread = pending.thread_summary;
-    if pending.include_turns {
+    let token_usage_turn_id = if pending.include_turns {
         populate_thread_turns_from_history(
             &mut thread,
             &pending.history_items,
             active_turn.as_ref(),
-        );
-    }
+        )
+    } else {
+        None
+    };
 
     let thread_status = thread_watch_manager
         .loaded_status_for_thread(&thread.id)
@@ -667,10 +669,6 @@ pub(super) async fn handle_pending_thread_resume_request(
     // Match cold resume: metadata-only resume should attach the listener without
     // paying the cost of turn reconstruction for historical usage replay.
     if let Some(token_usage_thread) = token_usage_thread {
-        let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items(
-            &pending.history_items,
-            token_usage_thread.turns.as_slice(),
-        );
         // Rejoining a loaded thread has the same UI contract as a cold resume, but
         // uses the live conversation state instead of reconstructing a new session.
         send_thread_token_usage_update_to_connection(
@@ -742,12 +740,16 @@ pub(crate) fn populate_thread_turns_from_history(
     thread: &mut Thread,
     items: &[RolloutItem],
     active_turn: Option<&Turn>,
-) {
-    let mut turns = build_api_turns_from_rollout_items(items);
+) -> Option<String> {
+    let (mut turns, token_usage_turn_owner) =
+        build_api_turns_from_rollout_items_with_token_usage_owner(items);
     if let Some(active_turn) = active_turn {
         merge_turn_history_with_active_turn(&mut turns, active_turn.clone());
     }
+    let token_usage_turn_id =
+        latest_token_usage_turn_id_from_owner(token_usage_turn_owner, turns.as_slice());
     thread.turns = turns;
+    token_usage_turn_id
 }
 
 pub(super) async fn resolve_pending_server_request(