Deduplicate retained instructions against Guardian transcripts (#48109)

## Why

Synchronous Guardian reviews can include the same user instruction in both the retained instructions section and the conversation transcript. Avoid this duplication while preserving complete authorization evidence and its original acceptance order.

## What changed

- Carry retained source metadata into transcript rendering. Omit a retained instruction when the transcript delivers its complete matching source revision.
- Label qualifying transcript entries with retained source order and protect them from budget truncation. Keep separate retained instructions when source metadata is missing or incomplete.
- Track delivery of source-order guidance so it remains available even when all retained instructions are delivered through the transcript. Invalidate that delivery proof when messages are shortened.

## Testing

Add coverage for complete-source deduplication, missing or incomplete metadata, budget protection, guidance redelivery, and source metadata preservation across resume. Update Guardian review assertions and snapshots to verify instructions are delivered once.

GitOrigin-RevId: 20ad0304501fa49f1596c4cfaf3ef89d64815753

codex-rs/core/src/compact_remote_v2.rs

@@ -1110,7 +1110,31 @@ mod tests {
             ),
         };
 
-        let truncated = truncate_without_metadata(vec![item], /*max_tokens*/ 3);
+        let source = codex_history::RetainedSource {
+            id: codex_history::RetainedSourceId {
+                message_id: "original".to_owned(),
+                turn_id: "parent".to_owned(),
+                role: codex_history::RetainedSourceRole::User,
+            },
+            revision: codex_protocol::ResponseItemId::from_server("revision-1".to_owned()),
+            complete: true,
+        };
+        let mut metadata = CodexHarnessMetadata {
+            retained_source: Some(source.clone()),
+            guardian_sources: vec![source],
+            ..Default::default()
+        };
+        let truncated = truncate_message_text_to_token_budget(
+            ResponseItemEnvelope {
+                item,
+                metadata: Some(metadata.clone()),
+            },
+            /*max_tokens*/ 3,
+        )
+        .unwrap();
+        metadata.mark_retained_sources_incomplete();
+        assert_eq!(truncated.metadata, Some(metadata));
+        let truncated = vec![truncated.item];
 
         assert_eq!(
             truncated,

codex-rs/core/src/context_manager/history.rs

@@ -141,12 +141,21 @@ impl ConversationHistorySnapshot for SharedConversationHistory {
     }
 
     fn review_items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
-        if self.guardian_review_mode == GuardianContextMode::Legacy
-            && let Some(history) = &self.review_history
-        {
-            return history.items();
+        Box::new(self.review_items_with_sources().map(|(item, _)| item))
+    }
+
+    fn review_items_with_sources(
+        &self,
+    ) -> Box<dyn Iterator<Item = (&ResponseItem, Option<&codex_history::RetainedSource>)> + Send + '_>
+    {
+        if self.guardian_review_mode == GuardianContextMode::Legacy {
+            let items = self
+                .review_history
+                .as_ref()
+                .map_or_else(|| self.items(), SectionHistory::items);
+            return Box::new(items.map(|item| (item, None)));
         }
-        self.items()
+        Box::new(self.items_with_sources())
     }
 
     fn review_history_version(&self) -> u64 {
@@ -168,12 +177,27 @@ impl ConversationHistorySnapshot for SharedConversationHistory {
     }
 
     fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
-        Box::new(
-            self.items
-                .iter()
-                .map(|envelope| &envelope.item)
-                .filter(|item| !is_guardian_context_message(item)),
-        )
+        Box::new(self.items_with_sources().map(|(item, _)| item))
+    }
+}
+
+impl SharedConversationHistory {
+    fn items_with_sources(
+        &self,
+    ) -> impl Iterator<Item = (&ResponseItem, Option<&codex_history::RetainedSource>)> + Send + '_
+    {
+        self.items
+            .iter()
+            .filter(|envelope| !is_guardian_context_message(&envelope.item))
+            .map(|envelope| {
+                (
+                    &envelope.item,
+                    envelope
+                        .metadata
+                        .as_ref()
+                        .and_then(|metadata| metadata.retained_source.as_ref()),
+                )
+            })
     }
 }
 

codex-rs/core/src/guardian/prompt.rs

@@ -287,23 +287,37 @@ impl SectionHistory for GuardianReviewHistory<'_> {
     }
 
     fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
-        self.0.review_items()
+        Box::new(self.items_with_sources().map(|(item, _)| item))
+    }
+
+    fn items_with_sources(
+        &self,
+    ) -> Box<dyn Iterator<Item = (&ResponseItem, Option<&codex_history::RetainedSource>)> + Send + '_>
+    {
+        self.0.review_items_with_sources()
     }
 }
 
 struct FilteredGuardianHistory<'a>(&'a dyn SectionHistory);
 
 impl SectionHistory for FilteredGuardianHistory<'_> {
+    fn items_with_sources(
+        &self,
+    ) -> Box<dyn Iterator<Item = (&ResponseItem, Option<&codex_history::RetainedSource>)> + Send + '_>
+    {
+        Box::new(
+            self.0
+                .items_with_sources()
+                .filter(|(item, _)| !is_guardian_context_message(item)),
+        )
+    }
+
     fn retained_context(&self) -> Option<&codex_history::RetainedContext> {
         self.0.retained_context()
     }
 
     fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
-        Box::new(
-            self.0
-                .items()
-                .filter(|item| !is_guardian_context_message(item)),
-        )
+        Box::new(self.items_with_sources().map(|(item, _)| item))
     }
 }
 

codex-rs/core/src/guardian/tests.rs

@@ -1054,6 +1054,7 @@ fn collect_guardian_transcript_entries_skips_contextual_user_messages() {
             kind: ConversationTranscriptEntryKind::ProtectedAssistant,
             text: "hello".to_string(),
             original_bytes: "hello".len(),
+            retained_source: None,
         }
     );
 }
@@ -1109,6 +1110,7 @@ fn collect_guardian_transcript_entries_includes_recent_tool_calls_and_output() {
             kind: ConversationTranscriptEntryKind::ToolCall("tool read_file call".to_string()),
             text: "{\"path\":\"README.md\"}".to_string(),
             original_bytes: "{\"path\":\"README.md\"}".len(),
+            retained_source: None,
         }
     );
     assert_eq!(
@@ -1117,6 +1119,7 @@ fn collect_guardian_transcript_entries_includes_recent_tool_calls_and_output() {
             kind: ConversationTranscriptEntryKind::ToolOutput("tool read_file result".to_string()),
             text: "repo is public".to_string(),
             original_bytes: "repo is public".len(),
+            retained_source: None,
         }
     );
     if let ResponseItem::FunctionCall { namespace, .. } = &mut items[1] {
@@ -1145,6 +1148,7 @@ fn collect_guardian_transcript_entries_includes_recent_tool_calls_and_output() {
                 ),
                 text: guardian_truncate_text(&oversized_result, token_cap).0,
                 original_bytes: oversized_result.len(),
+                retained_source: None,
             }
         );
         assert_eq!(entries.len(), 4);

codex-rs/core/src/session/rollout_reconstruction_tests.rs

@@ -75,6 +75,18 @@ async fn recorded_questions_share_queued_input_order_across_resume() {
             ],
         )
         .await;
+    let sources = |items: &[ResponseItemEnvelope]| {
+        items
+            .iter()
+            .map(|item| {
+                item.metadata
+                    .as_ref()
+                    .and_then(|metadata| metadata.retained_source.clone())
+            })
+            .collect::<Vec<_>>()
+    };
+    let original_sources = sources(session.clone_history().await.annotated_items());
+    assert!(original_sources.iter().any(Option::is_some));
     let saved = session
         .clone_history()
         .await
@@ -91,6 +103,7 @@ async fn recorded_questions_share_queued_input_order_across_resume() {
         }))
         .await;
     let history = session.clone_history().await;
+    assert_eq!(sources(history.annotated_items()), original_sources);
     assert_eq!(
         history
             .annotated_items()