Fix resume state restoration

Resume now preserves streamed assistant deltas in active thread snapshots so the latest response is not truncated after rejoin. Persist per-thread service_tier alongside model reasoning metadata so resumed sessions restore their own fast/standard mode instead of inheriting the current session settings.

Co-authored-by: Open Codex <hff582580@gmail.com>

Author: LEON-gittech

codex-rs/app-server-protocol/src/protocol/thread_history.rs

@@ -23,6 +23,7 @@ use crate::protocol::v2::UserInput;
 use crate::protocol::v2::WebSearchAction;
 use codex_protocol::items::parse_hook_prompt_message;
 use codex_protocol::models::MessagePhase;
+use codex_protocol::protocol::AgentMessageContentDeltaEvent;
 use codex_protocol::protocol::AgentReasoningEvent;
 use codex_protocol::protocol::AgentReasoningRawContentEvent;
 use codex_protocol::protocol::AgentStatus;
@@ -55,6 +56,7 @@ use codex_protocol::protocol::ViewImageToolCallEvent;
 use codex_protocol::protocol::WebSearchBeginEvent;
 use codex_protocol::protocol::WebSearchEndEvent;
 use std::collections::HashMap;
+use std::collections::HashSet;
 use tracing::warn;
 use uuid::Uuid;
 
@@ -97,6 +99,7 @@ pub struct ThreadHistoryBuilder {
     next_item_index: i64,
     current_rollout_index: usize,
     next_rollout_index: usize,
+    delta_agent_message_item_ids: HashSet<String>,
 }
 
 impl Default for ThreadHistoryBuilder {
@@ -113,6 +116,7 @@ impl ThreadHistoryBuilder {
             next_item_index: 1,
             current_rollout_index: 0,
             next_rollout_index: 0,
+            delta_agent_message_item_ids: HashSet::new(),
         }
     }
 
@@ -171,6 +175,9 @@ impl ThreadHistoryBuilder {
     pub fn handle_event(&mut self, event: &EventMsg) {
         match event {
             EventMsg::UserMessage(payload) => self.handle_user_message(payload),
+            EventMsg::AgentMessageContentDelta(payload) => {
+                self.handle_agent_message_content_delta(payload)
+            }
             EventMsg::AgentMessage(payload) => self.handle_agent_message(
                 payload.message.clone(),
                 payload.phase.clone(),
@@ -300,6 +307,28 @@ impl ThreadHistoryBuilder {
             return;
         }
 
+        let delta_backed_last_id = self.ensure_turn().items.last().and_then(|item| {
+            if matches!(item, ThreadItem::AgentMessage { .. }) {
+                Some(item.id().to_string())
+            } else {
+                None
+            }
+        });
+        if let Some(id) = delta_backed_last_id
+            && self.delta_agent_message_item_ids.remove(&id)
+            && let Some(ThreadItem::AgentMessage {
+                text: existing_text,
+                phase: existing_phase,
+                memory_citation: existing_memory_citation,
+                ..
+            }) = self.ensure_turn().items.last_mut()
+        {
+            *existing_text = text;
+            *existing_phase = phase;
+            *existing_memory_citation = memory_citation;
+            return;
+        }
+
         let id = self.next_item_id();
         self.ensure_turn().items.push(ThreadItem::AgentMessage {
             id,
@@ -309,6 +338,52 @@ impl ThreadHistoryBuilder {
         });
     }
 
+    fn handle_agent_message_content_delta(&mut self, payload: &AgentMessageContentDeltaEvent) {
+        if payload.delta.is_empty() {
+            return;
+        }
+
+        self.delta_agent_message_item_ids
+            .insert(payload.item_id.clone());
+
+        let append_delta = |items: &mut Vec<ThreadItem>| {
+            if let Some(ThreadItem::AgentMessage { text, .. }) =
+                items.iter_mut().find(|item| item.id() == payload.item_id)
+            {
+                text.push_str(&payload.delta);
+                return;
+            }
+
+            items.push(ThreadItem::AgentMessage {
+                id: payload.item_id.clone(),
+                text: payload.delta.clone(),
+                phase: None,
+                memory_citation: None,
+            });
+        };
+
+        if let Some(turn) = self.current_turn.as_mut()
+            && turn.id == payload.turn_id
+        {
+            append_delta(&mut turn.items);
+            return;
+        }
+
+        if let Some(turn) = self
+            .turns
+            .iter_mut()
+            .find(|turn| turn.id == payload.turn_id)
+        {
+            append_delta(&mut turn.items);
+            return;
+        }
+
+        warn!(
+            item_id = payload.item_id,
+            "dropping agent message delta for unknown turn id `{}`", payload.turn_id
+        );
+    }
+
     fn handle_agent_reasoning(&mut self, payload: &AgentReasoningEvent) {
         if payload.text.is_empty() {
             return;
@@ -1254,6 +1329,7 @@ mod tests {
     use codex_protocol::models::MessagePhase as CoreMessagePhase;
     use codex_protocol::models::WebSearchAction as CoreWebSearchAction;
     use codex_protocol::parse_command::ParsedCommand;
+    use codex_protocol::protocol::AgentMessageContentDeltaEvent;
     use codex_protocol::protocol::AgentMessageEvent;
     use codex_protocol::protocol::AgentReasoningEvent;
     use codex_protocol::protocol::AgentReasoningRawContentEvent;
@@ -1401,7 +1477,7 @@ mod tests {
                 local_images: Vec::new(),
             }),
             EventMsg::ItemStarted(ItemStartedEvent {
-                thread_id,
+                thread_id: thread_id.clone(),
                 turn_id: turn_id.to_string(),
                 item: CoreTurnItem::UserMessage(CoreUserMessageItem {
                     id: "user-item-id".to_string(),
@@ -1462,6 +1538,86 @@ mod tests {
         );
     }
 
+    #[test]
+    fn active_turn_snapshot_accumulates_agent_message_deltas() {
+        let thread_id = ThreadId::new().to_string();
+        let mut builder = ThreadHistoryBuilder::new();
+
+        builder.handle_event(&EventMsg::TurnStarted(TurnStartedEvent {
+            turn_id: "turn-1".to_string(),
+            started_at: None,
+            model_context_window: None,
+            collaboration_mode_kind: Default::default(),
+        }));
+        builder.handle_event(&EventMsg::AgentMessageContentDelta(
+            AgentMessageContentDeltaEvent {
+                thread_id: thread_id.clone(),
+                turn_id: "turn-1".to_string(),
+                item_id: "msg-1".to_string(),
+                delta: "partial ".to_string(),
+            },
+        ));
+        builder.handle_event(&EventMsg::AgentMessageContentDelta(
+            AgentMessageContentDeltaEvent {
+                thread_id: thread_id.clone(),
+                turn_id: "turn-1".to_string(),
+                item_id: "msg-1".to_string(),
+                delta: "reply".to_string(),
+            },
+        ));
+
+        let turn = builder
+            .active_turn_snapshot()
+            .expect("active turn snapshot");
+        assert_eq!(
+            turn.items,
+            vec![ThreadItem::AgentMessage {
+                id: "msg-1".to_string(),
+                text: "partial reply".to_string(),
+                phase: None,
+                memory_citation: None,
+            }]
+        );
+    }
+
+    #[test]
+    fn final_agent_message_replaces_delta_backed_message() {
+        let thread_id = ThreadId::new().to_string();
+        let mut builder = ThreadHistoryBuilder::new();
+
+        builder.handle_event(&EventMsg::TurnStarted(TurnStartedEvent {
+            turn_id: "turn-1".to_string(),
+            started_at: None,
+            model_context_window: None,
+            collaboration_mode_kind: Default::default(),
+        }));
+        builder.handle_event(&EventMsg::AgentMessageContentDelta(
+            AgentMessageContentDeltaEvent {
+                thread_id: thread_id.clone(),
+                turn_id: "turn-1".to_string(),
+                item_id: "msg-1".to_string(),
+                delta: "streamed prefix".to_string(),
+            },
+        ));
+        builder.handle_event(&EventMsg::AgentMessage(AgentMessageEvent {
+            message: "Final reply".to_string(),
+            phase: Some(CoreMessagePhase::FinalAnswer),
+            memory_citation: None,
+        }));
+
+        let turns = builder.finish();
+        assert_eq!(turns.len(), 1);
+        assert_eq!(
+            turns[0].items,
+            vec![ThreadItem::AgentMessage {
+                id: "msg-1".to_string(),
+                text: "Final reply".to_string(),
+                phase: Some(MessagePhase::FinalAnswer),
+                memory_citation: None,
+            }]
+        );
+    }
+
     #[test]
     fn replays_image_generation_end_events_into_turn_history() {
         let items = vec![

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

@@ -152,18 +152,22 @@ fn merge_persisted_resume_metadata(
     typesafe_overrides: &mut ConfigOverrides,
     persisted_metadata: &ThreadMetadata,
 ) {
-    if has_model_resume_override(request_overrides.as_ref(), typesafe_overrides) {
-        return;
+    if !has_model_resume_override(request_overrides.as_ref(), typesafe_overrides) {
+        typesafe_overrides.model = persisted_metadata.model.clone();
+        typesafe_overrides.model_provider = Some(persisted_metadata.model_provider.clone());
+
+        if let Some(reasoning_effort) = persisted_metadata.reasoning_effort {
+            request_overrides.get_or_insert_with(HashMap::new).insert(
+                "model_reasoning_effort".to_string(),
+                serde_json::Value::String(reasoning_effort.to_string()),
+            );
+        }
     }
 
-    typesafe_overrides.model = persisted_metadata.model.clone();
-    typesafe_overrides.model_provider = Some(persisted_metadata.model_provider.clone());
-
-    if let Some(reasoning_effort) = persisted_metadata.reasoning_effort {
-        request_overrides.get_or_insert_with(HashMap::new).insert(
-            "model_reasoning_effort".to_string(),
-            serde_json::Value::String(reasoning_effort.to_string()),
-        );
+    if typesafe_overrides.service_tier.is_none()
+        && let Some(service_tier) = &persisted_metadata.service_tier
+    {
+        typesafe_overrides.service_tier = Some(service_tier.clone());
     }
 }
 
@@ -196,7 +200,6 @@ fn has_model_resume_override(
     typesafe_overrides: &ConfigOverrides,
 ) -> bool {
     typesafe_overrides.model.is_some()
-        || typesafe_overrides.model_provider.is_some()
         || request_overrides.is_some_and(|overrides| overrides.contains_key("model"))
         || request_overrides
             .is_some_and(|overrides| overrides.contains_key("model_reasoning_effort"))