Preserve verified answers across history compaction (#42293)

## Why

Compaction replaces the model history window, but host-verified user answers may
still be needed for the rest of the owning thread.

## What changed

- Add bounded, model-invisible retained context for verified
  `request_user_input` answers.
- Persist retained context in rollout events and compaction checkpoints, and
  restore it when a thread resumes.
- Remove answers when their source turns are rolled back, and prevent
  parent-local answers from being inherited by child forks.

## Testing

Add lifecycle coverage for deduplication and size limits, compaction and resume,
rollback, rollout migration, and child forks.

GitOrigin-RevId: c32900d3e8120a9899b71f0576683acd67a4b898

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

@@ -759,6 +759,7 @@ fn compacted(message: &str, replacement_history: Option<Vec<ResponseItem>>) -> R
         message: message.to_string(),
         replacement_history: replacement_history
             .map(|items| items.into_iter().map(Into::into).collect()),
+        retained_context: None,
         guardian_history: None,
         mcp_resource_origins: None,
         window_number: Some(1),

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

@@ -294,6 +294,7 @@ impl LegacyRolloutCanonicalizer {
             | RolloutItem::TurnContext(_)
             | RolloutItem::TokenUsageRecord(_)
             | RolloutItem::RealtimeItem(_)
+            | RolloutItem::RetainedContext(_)
             | RolloutItem::SecurityRiskScore(_)
             | RolloutItem::WorldState(_)) => {
                 self.write_item(writer, &timestamp, item).await?;

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

@@ -86,6 +86,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)>,
     compactions: Vec<CompactionFrame>,
     model_replay: ModelReplayPlanner,
 }
@@ -102,6 +103,7 @@ impl RollbackPlanner {
             pending_user_response: None,
             pending_delivery_boundary: None,
             turn_boundaries: HashMap::new(),
+            retained_fact_sources: Vec::new(),
             compactions: Vec::new(),
             model_replay: ModelReplayPlanner::new(),
         }
@@ -223,6 +225,13 @@ impl RollbackPlanner {
                 self.assign_targeted_record(index, Some(record.turn_id.as_str()));
             }
             RolloutItem::WorldState(_) | RolloutItem::RealtimeItem(_) => {}
+            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()));
+            }
             RolloutItem::SecurityRiskScore(_) => self.record_boundaries[index] = None,
         }
 
@@ -235,12 +244,32 @@ impl RollbackPlanner {
             boundary_alive,
             compactions,
             model_replay,
+            turn_boundaries,
+            retained_fact_sources,
             ..
         } = self;
         let replay_anchor = model_replay.finish().empty_replacement_history_compaction;
+        let removed_turns = turn_boundaries
+            .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()
             .filter_map(|mut frame| {
+                // 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);
+                }
                 if Some(frame.record_index) == replay_anchor {
                     frame.item.replacement_history = Some(Vec::new());
                     frame.item.mcp_resource_origins = None;

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

@@ -77,6 +77,7 @@ impl ModelReplayPlanner {
             | RolloutItem::InterAgentCommunicationMetadata { .. }
             | RolloutItem::RealtimeItem(_)
             | RolloutItem::TokenUsageRecord(_)
+            | RolloutItem::RetainedContext(_)
             | RolloutItem::SecurityRiskScore(_)
             | RolloutItem::WorldState(_) => return,
         };

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

@@ -231,6 +231,7 @@ fn compacted(replacement_history: Vec<ResponseItem>) -> RolloutItem {
     RolloutItem::Compacted(CompactedItem {
         message: "checkpoint".to_string(),
         replacement_history: Some(replacement_history.into_iter().map(Into::into).collect()),
+        retained_context: None,
         guardian_history: None,
         mcp_resource_origins: None,
         window_number: Some(1),
@@ -1076,6 +1077,23 @@ async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() {
         turns: vec!["keep-before-compaction".to_string()],
         current_turn_id: Some("keep-before-compaction".to_string()),
     });
+    let answer_event = serde_json::from_value(serde_json::json!({
+        "type": "verified_answer", "turn_id": "keep-before-compaction", "call_id": "ask-1",
+        "questions": [{"question": "Publish?", "answer": "Only privately."}]
+    }))
+    .expect("verified answer fixture");
+    let wake_answer_event = serde_json::from_value(serde_json::json!({
+        "type": "verified_answer", "turn_id": "empty-answer-turn", "call_id": "ask-2",
+        "questions": [{"question": "Continue?", "answer": "Keep it private."}]
+    }))
+    .expect("empty-turn answer fixture");
+    checkpoint.retained_context = Some(Default::default());
+    let context = checkpoint
+        .retained_context
+        .as_mut()
+        .expect("retained context");
+    context.record(&answer_event);
+    context.record(&wake_answer_event);
     let path = write_rollout(
         home.path(),
         thread_id,
@@ -1083,7 +1101,11 @@ async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() {
         vec![
             started("keep-before-compaction"),
             user_message("old question"),
+            RolloutItem::RetainedContext(answer_event),
             completed("keep-before-compaction"),
+            started("empty-answer-turn"),
+            RolloutItem::RetainedContext(wake_answer_event),
+            completed("empty-answer-turn"),
             RolloutItem::Compacted(checkpoint),
             started("remove-after-compaction"),
             user_message("new question"),
@@ -1113,7 +1135,13 @@ async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() {
         .await
         .expect("read rollback-through-compaction turns");
     assert_eq!(turns.turns.len(), 1);
-    let checkpoint = read_rollout(&path)
+    let migrated = read_rollout(&path);
+    assert!(
+        !migrated
+            .iter()
+            .any(|line| matches!(line.item, RolloutItem::RetainedContext(_)))
+    );
+    let checkpoint = migrated
         .into_iter()
         .find_map(|line| match line.item {
             RolloutItem::Compacted(item) => Some(item),
@@ -1122,6 +1150,14 @@ async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() {
         .expect("retained compaction");
     assert_eq!(checkpoint.replacement_history, Some(Vec::new()));
     assert_eq!(checkpoint.mcp_resource_origins, None);
+    assert_eq!(
+        checkpoint
+            .retained_context
+            .expect("retained context checkpoint")
+            .verified_answers()
+            .count(),
+        0
+    );
 }
 
 #[tokio::test]
@@ -1410,6 +1446,7 @@ async fn migration_compacts_subagent_prefix_and_does_not_project_it() {
             RolloutItem::Compacted(CompactedItem {
                 message: "superseded checkpoint".repeat(1024),
                 replacement_history: Some(Vec::new()),
+                retained_context: None,
                 guardian_history: None,
                 mcp_resource_origins: None,
                 window_number: Some(1),
@@ -1433,6 +1470,7 @@ async fn migration_compacts_subagent_prefix_and_does_not_project_it() {
                     }
                     .into(),
                 ]),
+                retained_context: None,
                 guardian_history: None,
                 mcp_resource_origins: None,
                 window_number: Some(2),