Preserve retained answers across steer rollbacks (#42298)

## Why

A steer creates a new instruction boundary without creating a new turn ID. Retained `request_user_input` answers that share that turn ID must not all be discarded when only the later steer is rolled back.

## What changed

- Associate retained answers with their originating function call and instruction boundary.
- Keep answers whose source calls survive rollback, including through compaction, resume, and rollout migration.
- Remove answers whose source calls belong to the rolled-back instruction.

## Testing

- Cover live rollback and resume with answers before and after a steer.
- Cover rollout migration when delayed answers share a turn ID but originate from different calls.

GitOrigin-RevId: 225e7eb7b6a3ae5a5e7fa6fcb80cb7d259db90d0

codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs

@@ -5,6 +5,7 @@
 //! surviving turn after a newer turn has started. This planner keeps compact per-record ownership
 //! metadata for SQLite visibility, then combines it with `rollback_replay`'s cold-resume answer
 //! before the writer makes its second streaming pass.
+//! Retained answers follow their source calls, not later steers sharing the same turn ID.
 
 use std::collections::HashMap;
 
@@ -86,7 +87,7 @@ pub(super) struct RollbackPlanner {
     pending_user_response: Option<PendingUserResponse>,
     pending_delivery_boundary: Option<usize>,
     turn_boundaries: HashMap<String, usize>,
-    retained_fact_sources: Vec<(usize, String)>,
+    call_boundaries: HashMap<(String, String), Option<usize>>,
     compactions: Vec<CompactionFrame>,
     model_replay: ModelReplayPlanner,
 }
@@ -103,7 +104,7 @@ impl RollbackPlanner {
             pending_user_response: None,
             pending_delivery_boundary: None,
             turn_boundaries: HashMap::new(),
-            retained_fact_sources: Vec::new(),
+            call_boundaries: HashMap::new(),
             compactions: Vec::new(),
             model_replay: ModelReplayPlanner::new(),
         }
@@ -154,6 +155,14 @@ impl RollbackPlanner {
                     // no later turn to attach it to.
                     self.pending_context_records.push(index);
                 }
+                if let ResponseItem::FunctionCall { call_id, .. } = &response.item
+                    && let Some(turn_id) = response.turn_id().or(self.active_turn_id.as_deref())
+                {
+                    self.call_boundaries.insert(
+                        (turn_id.to_owned(), call_id.clone()),
+                        self.record_boundaries[index],
+                    );
+                }
             }
             RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => {
                 self.apply_rollback(rollback.num_turns)?;
@@ -228,9 +237,16 @@ impl RollbackPlanner {
             RolloutItem::RetainedContext(codex_rollout::RetainedContextEvent::VerifiedAnswer(
                 answer,
             )) => {
-                self.assign_targeted_record(index, Some(&answer.turn_id));
-                self.retained_fact_sources
-                    .push((index, answer.turn_id.clone()));
+                let source = (answer.turn_id.clone(), answer.call_id.clone());
+                // A late answer still belongs to its call's instruction boundary, even
+                // if the same running turn has since received another user steer.
+                if let Some(boundary) = self.call_boundaries.get(&source) {
+                    self.record_boundaries[index] = *boundary;
+                } else {
+                    self.assign_targeted_record(index, Some(&answer.turn_id));
+                }
+                self.call_boundaries
+                    .insert(source, self.record_boundaries[index]);
             }
             RolloutItem::SecurityRiskScore(_) => self.record_boundaries[index] = None,
         }
@@ -245,7 +261,7 @@ impl RollbackPlanner {
             compactions,
             model_replay,
             turn_boundaries,
-            retained_fact_sources,
+            call_boundaries,
             ..
         } = self;
         let replay_anchor = model_replay.finish().empty_replacement_history_compaction;
@@ -253,14 +269,6 @@ impl RollbackPlanner {
             .iter()
             .filter(|(_, boundary)| !boundary_alive[**boundary])
             .map(|(turn_id, _)| turn_id.as_str())
-            .chain(
-                retained_fact_sources
-                    .iter()
-                    .filter(|(index, _)| {
-                        record_boundaries[*index].is_some_and(|boundary| !boundary_alive[boundary])
-                    })
-                    .map(|(_, turn_id)| turn_id.as_str()),
-            )
             .collect::<Vec<_>>();
         let compacted_items = compactions
             .into_iter()
@@ -268,7 +276,14 @@ impl RollbackPlanner {
                 // Migration removes rollback markers. Checkpoints must therefore carry
                 // the same surviving facts as their standalone source events.
                 if let Some(context) = &mut frame.item.retained_context {
-                    context.remove_turns(&removed_turns);
+                    context.retain_answers(|answer| {
+                        call_boundaries
+                            .get(&(answer.turn_id.clone(), answer.call_id.clone()))
+                            .map_or_else(
+                                || !removed_turns.contains(&answer.turn_id.as_str()),
+                                |boundary| boundary.is_none_or(|boundary| boundary_alive[boundary]),
+                            )
+                    });
                 }
                 if Some(frame.record_index) == replay_anchor {
                     frame.item.replacement_history = Some(Vec::new());

codex-rs/thread-store/src/local/rollout_migration_tests.rs

@@ -1160,6 +1160,79 @@ async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() {
     );
 }
 
+#[tokio::test]
+async fn migration_preserves_answers_before_a_rolled_back_steer() {
+    let home = TempDir::new().expect("create Codex home");
+    let thread_id = ThreadId::new();
+    let mut items = vec![started("shared-turn")];
+    let mut answers = Vec::new();
+    let RolloutItem::Compacted(mut checkpoint) = compacted(vec![
+        input_response_message("user", "before-steer"),
+        input_response_message("user", "after-steer"),
+    ]) else {
+        unreachable!()
+    };
+    checkpoint.retained_context = Some(Default::default());
+    for call_id in ["before-steer", "after-steer"] {
+        items.push(user_message(call_id));
+        let mut call: ResponseItem = serde_json::from_value(json!({
+            "type": "function_call", "call_id": call_id,
+            "name": "request_user_input", "arguments": "{}"
+        }))
+        .expect("request_user_input call");
+        call.set_turn_id_if_missing("shared-turn");
+        items.push(rollout_response_item(call));
+        let answer: codex_rollout::RetainedContextEvent = serde_json::from_value(json!({
+            "type": "verified_answer", "turn_id": "shared-turn", "call_id": call_id,
+            "questions": [{"question": "Publish?", "answer": "Only privately."}]
+        }))
+        .expect("verified answer");
+        checkpoint
+            .retained_context
+            .as_mut()
+            .expect("retained context")
+            .record(&answer);
+        answers.push(answer);
+    }
+    // Delayed delivery must use the original call boundary, not the turn's latest steer.
+    items.extend(answers.iter().cloned().map(RolloutItem::RetainedContext));
+    items.push(completed("shared-turn"));
+    items.push(RolloutItem::Compacted(checkpoint));
+    items.push(RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
+        ThreadRolledBackEvent { num_turns: 1 },
+    )));
+    let path = write_rollout(home.path(), thread_id, SessionSource::Cli, items);
+    let store = indexed_store(home.path()).await;
+    store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("migrate rollback of a steer");
+    let migrated = read_rollout(&path);
+    let events = migrated
+        .iter()
+        .filter_map(|line| match &line.item {
+            RolloutItem::RetainedContext(event) => Some(event.clone()),
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(events, answers[..1]);
+    let checkpoint = migrated
+        .iter()
+        .find_map(|line| match &line.item {
+            RolloutItem::Compacted(checkpoint) => checkpoint.retained_context.as_ref(),
+            _ => None,
+        })
+        .expect("retained checkpoint");
+    assert_eq!(
+        checkpoint
+            .verified_answers()
+            .cloned()
+            .map(codex_rollout::RetainedContextEvent::VerifiedAnswer)
+            .collect::<Vec<_>>(),
+        answers[..1]
+    );
+}
+
 #[tokio::test]
 async fn migration_preserves_reverse_replay_anchor_after_pre_compaction_rollback() {
     let home = TempDir::new().expect("create Codex home");