Add retained conversation support to Guardian async sampling (#49792)

## What changed

- Add request preparation that preserves committed classifier history and appends transcript deltas, filtering instructions and images already delivered. Account for the complete input budget before transferring history; rejected requests leave the checkpoint and cursor intact.
- Add retained sampling that publishes the early score while collecting completed assistant output and encrypted reasoning for reuse. Preserve item IDs and input metadata, and discard unusable history without retracting an already published score.
- Keep the existing snapshot sampling path and share input preparation and execution with the new retained path. Retained sampling is not yet wired into the active backend.

## Testing

Add coverage for transcript and image deduplication, immutable history prefixes, budget boundaries, rejected preparation, WebSocket output retention, and retention failures that preserve the early score.

GitOrigin-RevId: 0fa33db8b8ea4483eb64d75e207690f1d349123b

codex-rs/ext/guardian-reviewer/src/conversation.rs

@@ -1,5 +1,6 @@
 //! Tracks reviewed transcript progress separately from committed conversation history.
 //! Hosts serialize reviews and checkpoint commits, and supply opaque completed history.
+//! An exclusive continuation can take the history without advancing reviewed progress.
 
 use codex_guardian_context::TranscriptCursor;
 
@@ -79,6 +80,12 @@ impl<H> ConversationState<H> {
     pub fn snapshot(&self) -> Option<&ConversationCheckpoint<H>> {
         self.checkpoint.as_ref()
     }
+
+    /// Transfers committed history to its exclusive owner, leaving progress unchanged.
+    /// The host must commit a completed replacement before this state can be reused.
+    pub fn take_history(&mut self) -> Option<H> {
+        self.checkpoint.take().map(|checkpoint| checkpoint.history)
+    }
 }
 
 #[cfg(test)]

codex-rs/ext/guardian-reviewer/src/conversation_tests.rs

@@ -30,4 +30,13 @@ fn fork_keeps_committed_history_and_cursor_while_next_review_is_uncommitted() {
         (history, fork.cursor(), fork.completed_review_count()),
         (vec!["first review", "second review"], Some(second), 2),
     );
+    assert_eq!(
+        (
+            state.take_history(),
+            state.cursor(),
+            state.completed_review_count(),
+        ),
+        (Some(vec!["first review", "second review"]), Some(second), 2),
+    );
+    assert!(state.snapshot().is_none());
 }

codex-rs/ext/guardian-v2/src/async_scorer/mod.rs

@@ -8,6 +8,9 @@ mod extension;
 mod metrics;
 mod observation;
 mod parent_compaction;
+// The treatment backend wires this builder into sampling in the next PR.
+#[allow(dead_code)]
+mod request;
 mod sampler;
 mod score;
 mod startup;

codex-rs/ext/guardian-v2/src/async_scorer/request.rs

@@ -0,0 +1,97 @@
+//! Admits classifier requests before moving committed history, without advancing the cursor.
+//! The complete immutable prefix, setup and new evidence share the classifier input limit.
+//! Instruction delivery metadata stays with history; previously delivered evidence is
+//! omitted from new input before budgeting, using the same instruction filtering as sync.
+
+use codex_guardian_context::TranscriptCursor;
+use codex_guardian_context::TruncationObservation;
+use codex_guardian_context::estimate_input_tokens;
+use codex_guardian_reviewer::ConversationCheckpoint;
+use codex_guardian_reviewer::ConversationState;
+use codex_history::ResponseItemEnvelope;
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ResponseItem;
+
+use super::sampler::LunaSampler;
+use super::sampler::LunaSamplingRequest;
+use super::transcript::CollectedTranscript;
+
+impl LunaSamplingRequest {
+    /// Takes committed history only after admission; rejection leaves the checkpoint intact.
+    pub(super) fn prepare_retained(
+        &self,
+        evidence: &CollectedTranscript,
+        conversation: &mut ConversationState<Vec<ResponseItemEnvelope>>,
+        max_input_tokens: usize,
+    ) -> Option<PreparedRequest<'_>> {
+        let cursor = conversation.cursor();
+        if conversation.snapshot().is_some() != cursor.is_some() {
+            return None;
+        }
+        let (_, next_cursor) = evidence.select(cursor);
+        if cursor.is_some_and(|cursor| {
+            cursor.parent_history_version != next_cursor.parent_history_version
+                || cursor.transcript_entry_count > next_cursor.transcript_entry_count
+        }) {
+            return None;
+        }
+        let (mut context, _) = evidence.compose(cursor).ok()?;
+        context.retain_new_instructions(
+            conversation
+                .snapshot()
+                .map(|snapshot| snapshot.history().as_slice())
+                .unwrap_or_default(),
+        );
+        if let Some(snapshot) = conversation.snapshot() {
+            context.retain_images(|image, _| {
+                !snapshot.history().iter().any(|item| match &item.item {
+                    ResponseItem::Message { content, .. } => content.iter().any(|item| {
+                        matches!(item, ContentItem::InputImage { image: retained, .. } if retained == image)
+                    }),
+                    _ => false,
+                })
+            });
+        }
+        let truncations = std::mem::take(&mut context.truncations);
+        let history = conversation.snapshot().map(ConversationCheckpoint::history);
+        let existing_context_tokens = history.map_or(0, |history| {
+            history
+                .iter()
+                .map(|item| estimate_input_tokens(&item.item))
+                .fold(0usize, usize::saturating_add)
+        });
+        // A continuation already owns its setup. Normalize only the new suffix while
+        // borrowing the immutable prefix for accounting; do not clone or rewrite it.
+        let (mut input, new_input_tokens) = LunaSampler::prepare_input(
+            self,
+            history.map(|_| Vec::new()),
+            context.into_annotated_messages(),
+        );
+        let input_tokens = existing_context_tokens.saturating_add(new_input_tokens);
+        if input_tokens > max_input_tokens.saturating_sub(/*rhs*/ 256) {
+            return None;
+        }
+        if let Some(mut history) = conversation.take_history() {
+            history.append(&mut input);
+            input = history;
+        }
+        Some(PreparedRequest {
+            sampling: self,
+            input,
+            cursor: next_cursor,
+            truncations,
+            existing_context_tokens,
+            input_tokens,
+        })
+    }
+}
+
+/// Admitted input and its token count, tied to the sampling settings used to prepare it.
+pub(super) struct PreparedRequest<'a> {
+    pub(super) sampling: &'a LunaSamplingRequest,
+    pub(super) input: Vec<ResponseItemEnvelope>,
+    pub(super) cursor: TranscriptCursor,
+    pub(super) truncations: Vec<TruncationObservation>,
+    pub(super) existing_context_tokens: usize,
+    pub(super) input_tokens: usize,
+}

codex-rs/ext/guardian-v2/src/async_scorer/request_tests.rs

@@ -0,0 +1,248 @@
+use super::*;
+use crate::async_scorer::transcript::CollectedTranscript;
+use crate::async_scorer::transcript::ContextInput;
+use crate::async_scorer::transcript::TranscriptConfig;
+use codex_extension_api::ConversationHistorySnapshot;
+use codex_guardian_context::ContextTarget;
+use codex_guardian_context::PlannedAction;
+use codex_guardian_context::PlannedActionKind;
+use codex_guardian_context::PreviousReviews;
+use codex_guardian_reviewer::ConversationState;
+use codex_history::ResponseItemEnvelope;
+use codex_protocol::models::ImageDetail;
+use codex_protocol::models::ImageReference;
+use pretty_assertions::assert_eq;
+
+const INPUT_BUDGET: usize = 128_000;
+
+struct History {
+    items: Vec<ResponseItem>,
+    version: u64,
+}
+
+impl ConversationHistorySnapshot for History {
+    fn history_version(&self) -> u64 {
+        self.version
+    }
+    fn user_message_revision(&self) -> u64 {
+        0
+    }
+    fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
+        Box::new(self.items.iter())
+    }
+}
+
+fn collect(history: &History, images: &[ContentItem]) -> CollectedTranscript {
+    let action = PlannedAction {
+        json: r#"{"tool":"read_file","path":"README.md"}"#.to_owned(),
+        tool_descriptions: None,
+        kind: PlannedActionKind::Command,
+        reason: None,
+    };
+    let reviews =
+        PreviousReviews::try_from_fragments(vec!["Approved reading README.md only.".to_owned()])
+            .unwrap();
+    TranscriptConfig::default()
+        .collect_context(ContextInput {
+            target: ContextTarget::Async,
+            history,
+            root_conversation: &[],
+            trusted_user_answers: &["Do not write files.".to_owned()],
+            planned_action: Some(&action),
+            permissions: None,
+            previous_reviews: Some(&reviews),
+            trusted_tool: None,
+            trusted_skill_paths: &[],
+            node_repl_images: Some(images),
+        })
+        .unwrap()
+}
+
+#[test]
+fn preparation_preserves_committed_prefix_and_selects_only_new_entries_and_images() {
+    let mut images = [
+        ImageReference::Inline {
+            image_url: "data:image/png;base64,Zmlyc3Q=".to_owned(),
+        },
+        ImageReference::File {
+            file_id: "file-screenshot".to_owned(),
+        },
+        ImageReference::Inline {
+            image_url: "data:image/png;base64,c2Vjb25k".to_owned(),
+        },
+    ]
+    .map(|image| ContentItem::InputImage {
+        image,
+        detail: None,
+    });
+    let input_images = |input: &[ResponseItemEnvelope]| {
+        input
+            .iter()
+            .flat_map(|item| match &item.item {
+                ResponseItem::Message { content, .. } => content.as_slice(),
+                _ => &[],
+            })
+            .filter(|item| matches!(item, ContentItem::InputImage { .. }))
+            .cloned()
+            .collect::<Vec<_>>()
+    };
+    let mut history = History {
+        items: vec![responses::user_message_item("Inspect README.md.")],
+        version: 1,
+    };
+    if let ResponseItem::Message { content, .. } = &mut history.items[0] {
+        content.push(images[1].clone());
+    }
+    let mut state = ConversationState::default();
+    let config = sample_request("parent");
+    let first = config
+        .prepare_retained(&collect(&history, &images[..1]), &mut state, INPUT_BUDGET)
+        .unwrap();
+    assert_eq!(
+        input_images(&first.input),
+        vec![images[1].clone(), images[0].clone()]
+    );
+    assert!(first.input.iter().all(|item| item.id().is_some()));
+    let mut committed = first.input;
+    // Keep the actual returned item, including the server ID and metadata.
+    let output: ResponseItem =
+        serde_json::from_value(ev_assistant_message("msg_classifier_a", "low")["item"].clone())
+            .unwrap();
+    committed.push(output.into());
+    state.complete_review(first.cursor);
+    state.commit_snapshot(committed.clone());
+    history
+        .items
+        .push(responses::user_message_item("Inspect LICENSE next."));
+    // Source detail can differ from the normalized image already in classifier history.
+    if let ContentItem::InputImage { detail, .. } = &mut images[0] {
+        *detail = Some(ImageDetail::High);
+    }
+    let evidence = collect(&history, &[images[0].clone(), images[2].clone()]);
+    // Fits three distinct images, but not another copy of the two retained images.
+    let second = config
+        .prepare_retained(&evidence, &mut state, /*max_input_tokens*/ 40_000)
+        .unwrap();
+    assert_eq!(&second.input[..committed.len()], committed);
+    assert!(state.snapshot().is_none());
+    assert_eq!(
+        second.input_tokens,
+        second
+            .input
+            .iter()
+            .map(|item| codex_guardian_context::estimate_input_tokens(&item.item))
+            .fold(0usize, usize::saturating_add)
+    );
+    assert_eq!(
+        input_images(&second.input[committed.len()..]),
+        vec![images[2].clone()]
+    );
+    assert_eq!(second.cursor.transcript_entry_count, 2);
+    let delta = serde_json::to_string(
+        &second.input[committed.len()..]
+            .iter()
+            .map(|item| &item.item)
+            .collect::<Vec<_>>(),
+    )
+    .unwrap();
+    assert!(delta.contains("TRANSCRIPT DELTA START"));
+    assert!(delta.contains("[2] user: Inspect LICENSE next."));
+    assert!(!delta.contains("[1] user: Inspect README.md."));
+    assert!(delta.contains("Do not write files."));
+    assert!(delta.contains("read_file"));
+    assert!(
+        matches!(&second.input[committed.len()].item, ResponseItem::Message { role, .. } if role == "user")
+    );
+    assert!(second.input[committed.len()..].iter().any(
+        |item| matches!(&item.item, ResponseItem::Message { role, .. } if role == "developer")
+    ));
+    // A rebuilt conversation must receive the current images again.
+    let rebuilt = config
+        .prepare_retained(&evidence, &mut ConversationState::default(), INPUT_BUDGET)
+        .unwrap();
+    assert_eq!(input_images(&rebuilt.input), input_images(&second.input));
+}
+
+#[test]
+fn rejected_preparation_preserves_committed_history_and_cursor() {
+    let mut history = History {
+        items: vec![responses::user_message_item("Inspect README.md.")],
+        version: 1,
+    };
+    let mut state = ConversationState::default();
+    let config = sample_request("parent");
+    let first = config
+        .prepare_retained(&collect(&history, &[]), &mut state, INPUT_BUDGET)
+        .unwrap();
+    let limit = first.input_tokens + 256;
+    let mut committed = first.input;
+    committed.push(responses::user_message_item(&"output ".repeat(/*n*/ 400)).into());
+    state.complete_review(first.cursor);
+    state.commit_snapshot(committed.clone());
+    assert!(
+        config
+            .prepare_retained(&collect(&history, &[]), &mut state, limit)
+            .is_none()
+    );
+    assert_eq!(state.cursor(), Some(first.cursor));
+    assert_eq!(state.snapshot().unwrap().history(), &committed);
+    config
+        .prepare_retained(
+            &collect(&history, &[]),
+            &mut ConversationState::default(),
+            limit,
+        )
+        .unwrap();
+    history.version += 1;
+    assert!(
+        config
+            .prepare_retained(&collect(&history, &[]), &mut state, INPUT_BUDGET)
+            .is_none()
+    );
+    state.reset_transcript();
+    assert!(
+        config
+            .prepare_retained(&collect(&history, &[]), &mut state, INPUT_BUDGET)
+            .is_none()
+    );
+    assert_eq!(state.snapshot().unwrap().history(), &committed);
+}
+
+#[test]
+fn complete_budget_counts_setup_checkpoint_images_and_trusted_evidence() {
+    let history = History {
+        items: vec![responses::user_message_item("Inspect README.md.")],
+        version: 1,
+    };
+    let mut config = sample_request("parent");
+    config.parent_compaction = Some(ResponseItem::ContextCompaction {
+        id: Some(ResponseItemId::from_server("cmp_parent".to_owned())),
+        encrypted_content: Some("opaque checkpoint".repeat(/*n*/ 100)),
+        internal_chat_message_metadata_passthrough: None,
+    });
+    let images = [ContentItem::InputImage {
+        image: ImageReference::Inline {
+            image_url: "data:image/png;base64,aGVsbG8=".to_owned(),
+        },
+        detail: None,
+    }];
+    let mut state = ConversationState::default();
+    let evidence = collect(&history, &images);
+    let request = config
+        .prepare_retained(&evidence, &mut state, INPUT_BUDGET)
+        .unwrap();
+    assert_eq!(
+        Some(&request.input[2].item),
+        config.parent_compaction.as_ref()
+    );
+    assert!(request.input_tokens > 10_000);
+    assert!(
+        config
+            .prepare_retained(&evidence, &mut state, request.input_tokens + 255)
+            .is_none()
+    );
+    // The complete input fits exactly at the reserved-output boundary.
+    config
+        .prepare_retained(&evidence, &mut state, request.input_tokens + 256)
+        .unwrap();
+}

codex-rs/ext/guardian-v2/src/async_scorer/retained_sampling_tests.rs

@@ -0,0 +1,151 @@
+use super::*;
+use crate::async_scorer::request::PreparedRequest;
+use codex_guardian_context::TranscriptCursor;
+use codex_history::ResponseItemEnvelope;
+use pretty_assertions::assert_eq;
+
+// The app-server approval matrix covers successful HTTP retention and both experiment arms.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn retained_websocket_requests_preserve_model_output_and_ids() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    let events = vec![
+        ev_output_text_delta("low"),
+        json!({"type":"response.output_item.done", "item":{"type":"reasoning", "id":"rs_score", "summary":[], "encrypted_content":"opaque reasoning"}}),
+        ev_assistant_message("msg_score", "low"),
+        ev_completed("resp_score"),
+    ];
+    let ws =
+        responses::start_websocket_server(vec![Vec::new(), vec![events.clone(), events]]).await;
+    let sampler = connect_sampler(sampler_config(format!(
+        "http://{}/v1",
+        ws.uri().trim_start_matches("ws://")
+    )))
+    .await?;
+    let mut committed = sample_request("parent").input;
+    let mut output = Vec::new();
+    for _ in 0..2 {
+        let (ready, score) = tokio::sync::oneshot::channel();
+        let mut request = sample_request("parent");
+        request.input = committed.clone();
+        let input_tokens = request
+            .input
+            .iter()
+            .map(codex_guardian_context::estimate_input_tokens)
+            .fold(0usize, usize::saturating_add);
+        let prepared = PreparedRequest {
+            sampling: &request,
+            input: request
+                .input
+                .iter()
+                .cloned()
+                .map(ResponseItemEnvelope::new)
+                .collect(),
+            input_tokens,
+            existing_context_tokens: 0,
+            cursor: TranscriptCursor {
+                parent_history_version: 0,
+                transcript_entry_count: 0,
+            },
+            truncations: Vec::new(),
+        };
+        let (completed, score) = tokio::join!(sampler.sample_retained(prepared, ready), score);
+        assert_eq!(score??, "low");
+        let completed = completed
+            .expect("completed model history")
+            .into_iter()
+            .map(ResponseItemEnvelope::into_item)
+            .collect::<Vec<_>>();
+        assert_eq!(&completed[..committed.len()], committed);
+        output = completed[committed.len()..].to_vec();
+        committed = completed;
+        committed.push(responses::user_message_item("Next action."));
+    }
+    assert!(
+        matches!(&output[..], [ResponseItem::Reasoning { encrypted_content: Some(_), .. }, ResponseItem::Message { id: Some(id), role, .. }] if id.as_str() == "msg_score" && role == "assistant")
+    );
+    let requests = ws.connections().into_iter().flatten().collect::<Vec<_>>();
+    assert_eq!(requests.len(), 2);
+    let first = requests[0].body_json();
+    assert_eq!(first["include"], json!(["reasoning.encrypted_content"]));
+    let second = requests[1].body_json();
+    assert_eq!(second["input"][0], first["input"][0]);
+    assert_eq!(
+        json!(&second["input"].as_array().unwrap()[1..1 + output.len()]),
+        serde_json::to_value(&output)?
+    );
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn retention_overflow_or_incomplete_stream_keeps_early_score() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    for (limit, output, completed) in [
+        (
+            512,
+            vec![ev_assistant_message(
+                "msg_large",
+                &"large ".repeat(/*n*/ 1_000),
+            )],
+            true,
+        ),
+        (
+            64_000,
+            vec![
+                json!({"type":"response.output_item.done", "item":{"type":"reasoning", "id":"rs_large", "summary":[], "encrypted_content":"x".repeat(/*n*/ 48_000)}}),
+                ev_assistant_message("msg_score", "low"),
+            ],
+            true,
+        ),
+        (
+            512,
+            vec![ev_assistant_message("msg_score", "lowhigh")],
+            true,
+        ),
+        (
+            512,
+            vec![
+                json!({"type":"response.output_item.done", "item":{"type":"message", "role":"developer", "content":[{"type":"output_text", "text":"low"}]}}),
+            ],
+            true,
+        ),
+        (512, vec![], false),
+    ] {
+        let mut events = vec![ev_output_text_delta("low")];
+        events.extend(output);
+        if completed {
+            events.push(ev_completed("score"));
+        }
+        let server = responses::start_mock_server().await;
+        responses::mount_sse_once(&server, responses::sse(events)).await;
+        let mut config = sampler_config(server.uri());
+        config.max_input_tokens = limit;
+        let sampler = LunaSampler::new(config);
+        let (ready, score) = tokio::sync::oneshot::channel();
+        let request = sample_request("parent");
+        let input_tokens = request
+            .input
+            .iter()
+            .map(codex_guardian_context::estimate_input_tokens)
+            .fold(0usize, usize::saturating_add);
+        let prepared = PreparedRequest {
+            sampling: &request,
+            input: request
+                .input
+                .iter()
+                .cloned()
+                .map(ResponseItemEnvelope::new)
+                .collect(),
+            input_tokens,
+            existing_context_tokens: 0,
+            cursor: TranscriptCursor {
+                parent_history_version: 0,
+                transcript_entry_count: 0,
+            },
+            truncations: Vec::new(),
+        };
+        let (history, score) = tokio::join!(sampler.sample_retained(prepared, ready), score);
+        assert_eq!(score??, "low");
+        assert_eq!(history, None);
+    }
+    Ok(())
+}

codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs

@@ -4,6 +4,7 @@
 mod connection_pool;
 mod execution;
 
+use super::request::PreparedRequest;
 use connection_pool::ConnectionPool;
 use std::collections::VecDeque;
 use std::sync::Arc;
@@ -17,6 +18,7 @@ use codex_api::ReasoningContext;
 use codex_api::ResponsesApiRequest;
 use codex_context_fragments::RenderedFragment;
 use codex_extension_api::ExtensionMetrics;
+use codex_history::ResponseItemEnvelope;
 use codex_http_client::HttpClientFactory;
 use codex_login::AgentIdentityAuthPolicy;
 use codex_model_provider::SharedModelProvider;
@@ -152,10 +154,8 @@ impl LunaSampler {
         }
     }
 
-    /// Sends one tool-less classification request using an available transport.
-    pub async fn sample(&self, request: LunaSamplingRequest) -> Result<String, LunaSamplerError> {
-        let auth_owner_generation = self
-            .config
+    fn auth_owner_generation(&self) -> Option<u64> {
+        self.config
             .provider
             .auth_manager()
             .filter(|_| self.config.provider.info().auth.is_none())
@@ -164,31 +164,35 @@ impl LunaSampler {
                     .auth_change_state_receiver()
                     .borrow()
                     .owner_generation
-            });
-        if request.parent_compaction.is_some()
-            && !self.supports_parent_compaction(request.parent_compaction_hash.as_deref())
-        {
-            return Err(LunaSamplerError::IncompatibleCompaction);
-        }
-        // A classification is its own inference turn; retries keep that identity.
-        let turn_id = Uuid::now_v7().to_string();
-        let parent_response_id = request.parent_response_id;
-        let parent_turn_id = request.parent_turn_id;
-        let root_turn_id = request.root_turn_id;
-        let mut input = vec![
-            ResponseItem::AdditionalTools {
-                id: None,
-                role: "developer".to_owned(),
-                tools: Vec::new(),
-            },
-            ResponseItem::from(request.instructions),
-        ];
-        if let Some(parent_compaction) = request.parent_compaction {
-            input.push(parent_compaction);
-        }
-        let mut evidence = request.input;
+            })
+    }
+
+    // Normalize new evidence while preserving the retained prefix and its item IDs.
+    pub(super) fn prepare_input(
+        request: &LunaSamplingRequest,
+        history: Option<Vec<ResponseItemEnvelope>>,
+        mut evidence: Vec<ResponseItemEnvelope>,
+    ) -> (Vec<ResponseItemEnvelope>, usize) {
+        let mut input = history.unwrap_or_else(|| {
+            let mut input = vec![
+                ResponseItemEnvelope::new(ResponseItem::AdditionalTools {
+                    id: None,
+                    role: "developer".to_owned(),
+                    tools: Vec::new(),
+                }),
+                ResponseItemEnvelope::new(ResponseItem::from(request.instructions.clone())),
+            ];
+            input.extend(
+                request
+                    .parent_compaction
+                    .clone()
+                    .map(ResponseItemEnvelope::new),
+            );
+            input
+        });
+        // Normalize only new evidence; never rewrite the committed prefix.
         for item in &mut evidence {
-            if let ResponseItem::Message { content, .. } = item {
+            if let ResponseItem::Message { content, .. } = &mut item.item {
                 for content in content {
                     if let ContentItem::InputImage { detail, .. } = content {
                         *detail = None;
@@ -205,10 +209,74 @@ impl LunaSampler {
                 item.set_id(Some(ResponseItemId::new(prefix)));
             }
         }
-        let total_tokens = input
+        let tokens = input
             .iter()
-            .map(codex_guardian_context::estimate_input_tokens)
+            .map(|item| codex_guardian_context::estimate_input_tokens(&item.item))
             .fold(0usize, usize::saturating_add);
+        (input, tokens)
+    }
+
+    fn execution(
+        &self,
+        request: &LunaSamplingRequest,
+        input: Vec<ResponseItem>,
+        auth_owner_generation: Option<u64>,
+    ) -> execution::SamplingExecution {
+        let api_request = ResponsesApiRequest {
+            model: MODEL.to_owned(),
+            instructions: String::new(),
+            input,
+            tools: None,
+            tool_choice: "none".to_owned(),
+            parallel_tool_calls: false,
+            reasoning: Some(Reasoning {
+                effort: Some(request.reasoning_effort.clone()),
+                summary: None,
+                context: Some(ReasoningContext::AllTurns),
+            }),
+            store: false,
+            stream: true,
+            stream_options: None,
+            include: Vec::new(),
+            service_tier: None,
+            prompt_cache_key: Some(format!("guardian-v2:{}", self.config.thread_id)),
+            text: None,
+            client_metadata: None,
+            access_programs: None,
+        };
+        execution::SamplingExecution {
+            auth_owner_generation,
+            config: Arc::clone(&self.config),
+            connections: Arc::clone(&self.connections),
+            request: api_request,
+            // Retries keep the classification's inference turn identity.
+            turn_id: Uuid::now_v7().to_string(),
+            parent_response_id: request.parent_response_id.clone(),
+            parent_turn_id: request.parent_turn_id.clone(),
+            root_turn_id: request.root_turn_id.clone(),
+        }
+    }
+
+    /// Sends one tool-less classification request using an available transport.
+    pub async fn sample(
+        &self,
+        mut request: LunaSamplingRequest,
+    ) -> Result<String, LunaSamplerError> {
+        let auth_owner_generation = self.auth_owner_generation();
+        if request.parent_compaction.is_some()
+            && !self.supports_parent_compaction(request.parent_compaction_hash.as_deref())
+        {
+            return Err(LunaSamplerError::IncompatibleCompaction);
+        }
+        let evidence = std::mem::take(&mut request.input)
+            .into_iter()
+            .map(ResponseItemEnvelope::new)
+            .collect();
+        let (input, total_tokens) = Self::prepare_input(&request, /*history*/ None, evidence);
+        let input = input
+            .into_iter()
+            .map(ResponseItemEnvelope::into_item)
+            .collect();
         if let Some(metrics) = self.config.metrics.as_deref() {
             for (component, tokens) in [
                 ("existing_context", 0),
@@ -227,30 +295,8 @@ impl LunaSampler {
         if total_tokens > self.config.max_input_tokens.saturating_sub(/*rhs*/ 256) {
             return Err(LunaSamplerError::InputTooLarge);
         }
-        let request = ResponsesApiRequest {
-            model: MODEL.to_owned(),
-            instructions: String::new(),
-            input,
-            tools: None,
-            tool_choice: "none".to_owned(),
-            parallel_tool_calls: false,
-            reasoning: Some(Reasoning {
-                effort: Some(request.reasoning_effort),
-                summary: None,
-                context: Some(ReasoningContext::AllTurns),
-            }),
-            store: false,
-            stream: true,
-            stream_options: None,
-            include: Vec::new(),
-            service_tier: None,
-            prompt_cache_key: Some(format!("guardian-v2:{}", self.config.thread_id)),
-            text: None,
-            client_metadata: None,
-            access_programs: None,
-        };
         let (supersede, superseded) = oneshot::channel();
-        let scored = Arc::new(AtomicBool::new(false));
+        let scored = Arc::new(AtomicBool::new(/*v*/ false));
         {
             let mut active_requests = self
                 .active_requests
@@ -271,18 +317,50 @@ impl LunaSampler {
                 scored: Arc::clone(&scored),
             });
         }
-        execution::SamplingExecution {
-            auth_owner_generation,
-            config: Arc::clone(&self.config),
-            connections: Arc::clone(&self.connections),
-            request,
-            turn_id,
-            parent_response_id,
-            parent_turn_id,
-            root_turn_id,
-        }
-        .run(superseded, scored)
-        .await
+        self.execution(&request, input, auth_owner_generation)
+            .run(execution::SamplingMode::Snapshot { superseded, scored })
+            .await
+    }
+
+    /// Publishes the early score and returns only validated, completed history.
+    #[cfg_attr(not(test), allow(dead_code))]
+    pub(super) async fn sample_retained(
+        &self,
+        prepared: PreparedRequest<'_>,
+        ready: oneshot::Sender<Result<String, LunaSamplerError>>,
+    ) -> Option<Vec<ResponseItemEnvelope>> {
+        let auth_owner_generation = self.auth_owner_generation();
+        let mut completion = execution::RetainedCompletion {
+            ready: Some(ready),
+            output: Some(Vec::new()),
+            early_score: None,
+            remaining_tokens: self
+                .config
+                .max_input_tokens
+                .saturating_sub(/*rhs*/ 256)
+                .saturating_sub(prepared.input_tokens),
+        };
+        // Transport sees model items; delivery proof follows the unchanged input prefix.
+        let (input, metadata): (Vec<_>, Vec<_>) = prepared
+            .input
+            .into_iter()
+            .map(|envelope| (envelope.item, envelope.metadata))
+            .unzip();
+        let mut execution = self.execution(prepared.sampling, input, auth_owner_generation);
+        execution.request.include = vec!["reasoning.encrypted_content".to_owned()];
+        let result = execution
+            .run(execution::SamplingMode::Retained(&mut completion))
+            .await;
+        execution.request.input.extend(completion.finish(result)?);
+        Some(
+            execution
+                .request
+                .input
+                .into_iter()
+                .zip(metadata.into_iter().chain(std::iter::repeat(None)))
+                .map(|(item, metadata)| ResponseItemEnvelope { item, metadata })
+                .collect(),
+        )
     }
 }
 

codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs

@@ -1,5 +1,6 @@
 //! Executes one prepared classifier request with the existing retry and streaming rules.
-//! The first output returns immediately; a detached drain preserves reuse and token accounting.
+//! Snapshot output returns immediately with a detached drain. Retained requests publish
+//! the same early result but keep ownership until completion to capture reusable output.
 //! Requests and retries stop when the account owner that started the classification changes.
 
 use super::CLASSIFICATION_TOKEN_USAGE_METRIC;
@@ -41,6 +42,81 @@ pub(super) struct SamplingExecution {
     pub(super) root_turn_id: Option<String>,
 }
 
+pub(super) struct RetainedCompletion {
+    pub(super) ready: Option<oneshot::Sender<Result<String, LunaSamplerError>>>,
+    pub(super) output: Option<Vec<ResponseItem>>,
+    pub(super) remaining_tokens: usize,
+    pub(super) early_score: Option<String>,
+}
+
+/// Snapshot admission owns supersession; retained admission owns the response through completion.
+pub(super) enum SamplingMode<'a> {
+    Snapshot {
+        superseded: oneshot::Receiver<()>,
+        scored: Arc<AtomicBool>,
+    },
+    Retained(&'a mut RetainedCompletion),
+}
+
+impl SamplingMode<'_> {
+    async fn superseded(&mut self) {
+        match self {
+            Self::Snapshot { superseded, .. } => {
+                let _ = superseded.await;
+            }
+            Self::Retained(_) => std::future::pending().await,
+        }
+    }
+
+    fn early_result_sent(&self) -> bool {
+        matches!(self, Self::Retained(completion) if completion.ready.is_none())
+    }
+}
+
+impl RetainedCompletion {
+    fn record(&mut self, item: ResponseItem) {
+        let Some(output) = self.output.as_mut() else {
+            return;
+        };
+        let tokens = codex_guardian_context::estimate_input_tokens(&item);
+        let valid_item = match &item {
+            ResponseItem::Message { role, content, .. } => {
+                role == "assistant"
+                    && content
+                        .iter()
+                        .all(|part| matches!(part, ContentItem::OutputText { .. }))
+            }
+            ResponseItem::Reasoning {
+                encrypted_content: Some(content),
+                ..
+            } => !content.is_empty(),
+            _ => false,
+        };
+        if tokens > self.remaining_tokens || tokens > 10_000 || !valid_item {
+            // Retention failure does not invalidate an already published score.
+            self.output = None;
+        } else {
+            self.remaining_tokens -= tokens;
+            output.push(item);
+        }
+    }
+
+    /// Publishes any remaining result and returns only history matching the completed score.
+    pub(super) fn finish(
+        mut self,
+        result: Result<String, LunaSamplerError>,
+    ) -> Option<Vec<ResponseItem>> {
+        let reusable = result.as_ref().is_ok_and(|score| {
+            matches!(score.as_str(), "high" | "low")
+                && self.early_score.as_ref().is_none_or(|early| early == score)
+        });
+        if let Some(ready) = self.ready.take() {
+            let _ = ready.send(result);
+        }
+        self.output.filter(|output| reusable && !output.is_empty())
+    }
+}
+
 impl SamplingExecution {
     async fn retry_after_failure(
         &self,
@@ -110,9 +186,8 @@ impl SamplingExecution {
     }
 
     pub(super) async fn run(
-        mut self,
-        mut superseded: oneshot::Receiver<()>,
-        scored: Arc<AtomicBool>,
+        &mut self,
+        mut mode: SamplingMode<'_>,
     ) -> Result<String, LunaSamplerError> {
         let auth_changes = self
             .config
@@ -154,12 +229,20 @@ impl SamplingExecution {
             .provider
             .auth_manager()
             .map(|manager| manager.unauthorized_recovery());
+        let retention_budget = match &mode {
+            SamplingMode::Snapshot { .. } => 0,
+            SamplingMode::Retained(completion) => completion.remaining_tokens,
+        };
         'retry: loop {
             ensure_account_owner()?;
+            if let SamplingMode::Retained(completion) = &mut mode {
+                completion.output = Some(Vec::new());
+                completion.remaining_tokens = retention_budget;
+            }
             let lease = match tokio::select! {
                 biased;
                 _ = &mut owner_changed => return Err(account_changed_error()),
-                _ = &mut superseded => return Err(LunaSamplerError::Superseded),
+                _ = mode.superseded() => return Err(LunaSamplerError::Superseded),
                 lease = self.connections.lease() => lease,
             } {
                 Ok(lease) => lease,
@@ -213,7 +296,7 @@ impl SamplingExecution {
             let mut stream = match tokio::select! {
                 biased;
                 _ = &mut owner_changed => return Err(account_changed_error()),
-                _ = &mut superseded => return Err(LunaSamplerError::Superseded),
+                _ = mode.superseded() => return Err(LunaSamplerError::Superseded),
                 stream = lease.stream_request(&self.request) => stream,
             } {
                 Ok(stream) => stream,
@@ -233,8 +316,9 @@ impl SamplingExecution {
             while let Some(event) = tokio::select! {
                 biased;
                 _ = &mut owner_changed => return Err(account_changed_error()),
-                _ = &mut superseded => {
-                    return if scored.load(Ordering::Relaxed) && !output.is_empty() {
+                _ = mode.superseded() => {
+                    let scored = matches!(&mode, SamplingMode::Snapshot { scored, .. } if scored.load(Ordering::Relaxed));
+                    return if scored && !output.is_empty() {
                         Ok(output)
                     } else {
                         Err(LunaSamplerError::Superseded)
@@ -247,6 +331,9 @@ impl SamplingExecution {
                     Ok(event) => event,
                     Err(error) => {
                         let error = LunaSamplerError::Api(error);
+                        if mode.early_result_sent() {
+                            return Err(error);
+                        }
                         if self
                             .retry_after_failure(&error, &mut auth_recovery, &mut retries)
                             .await
@@ -258,6 +345,9 @@ impl SamplingExecution {
                 };
                 match event {
                     ResponseEvent::OutputTextDelta(delta) => {
+                        if mode.early_result_sent() {
+                            continue;
+                        }
                         if delta.is_empty() {
                             continue;
                         }
@@ -267,6 +357,16 @@ impl SamplingExecution {
                         // The first output token is the complete classification.
                         // Later output cannot revise that decision; drain it only
                         // to preserve connection reuse and token accounting.
+                        let (mut superseded, scored) = match mode {
+                            SamplingMode::Snapshot { superseded, scored } => (superseded, scored),
+                            SamplingMode::Retained(ref mut retained) => {
+                                retained.early_score = Some(delta.clone());
+                                if let Some(ready) = retained.ready.take() {
+                                    let _ = ready.send(Ok(delta));
+                                }
+                                continue;
+                            }
+                        };
                         scored.store(true, Ordering::Relaxed);
                         let mut remaining_events = stream.rx_event;
                         let metrics = self.config.metrics.clone();
@@ -292,12 +392,17 @@ impl SamplingExecution {
                         });
                         return Ok(delta);
                     }
-                    ResponseEvent::OutputItemDone(ResponseItem::Message {
-                        role, content, ..
-                    }) if role == "assistant" => {
-                        for item in content {
-                            if let ContentItem::OutputText { text } = item {
-                                output.push_str(&text);
+                    ResponseEvent::OutputItemDone(item) => {
+                        if let SamplingMode::Retained(completion) = &mut mode {
+                            completion.record(item.clone());
+                        }
+                        if let ResponseItem::Message { role, content, .. } = item
+                            && role == "assistant"
+                        {
+                            for item in content {
+                                if let ContentItem::OutputText { text } = item {
+                                    output.push_str(&text);
+                                }
                             }
                         }
                     }
@@ -314,7 +419,9 @@ impl SamplingExecution {
                 if output.len() > MAX_OUTPUT_BYTES {
                     return Err(LunaSamplerError::OutputTooLarge);
                 }
-                if !output.is_empty() {
+                if !output.is_empty()
+                    && let SamplingMode::Snapshot { scored, .. } = &mode
+                {
                     scored.store(true, Ordering::Relaxed);
                 }
             }

codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs

@@ -49,6 +49,10 @@ use super::LunaSamplerError;
 use super::LunaSamplingRequest;
 use super::MAX_CONCURRENT_REQUESTS;
 
+#[path = "request_tests.rs"]
+mod request;
+#[path = "retained_sampling_tests.rs"]
+mod retained;
 #[path = "sampler_routing_tests.rs"]
 mod routing;
 

codex-rs/ext/guardian-v2/src/async_scorer/transcript.rs

@@ -1,6 +1,7 @@
 use codex_extension_api::ConversationHistorySnapshot;
 use codex_extension_api::ResponseItem;
 pub(crate) use codex_features::GuardianV2TranscriptSource as TranscriptSource;
+use codex_guardian_context::CollectedContext;
 use codex_guardian_context::ComposedContext;
 use codex_guardian_context::ContextPresentation;
 use codex_guardian_context::ContextProfile;
@@ -15,9 +16,12 @@ use codex_guardian_context::PreviousReviews;
 use codex_guardian_context::SectionError;
 use codex_guardian_context::SectionHistory;
 use codex_guardian_context::SectionInput;
+use codex_guardian_context::TranscriptCursor;
 use codex_guardian_context::TranscriptEntryLimits;
 use codex_guardian_context::TranscriptImageInput;
+use codex_guardian_context::TranscriptMode;
 use codex_guardian_context::TranscriptRetentionConfig;
+use codex_guardian_context::TranscriptSelection;
 use codex_guardian_context::TrustedTool;
 use codex_guardian_context::default_registry;
 
@@ -84,6 +88,23 @@ impl TranscriptConfig {
         &self,
         input: ContextInput<'_>,
     ) -> Result<RenderedContext, SectionError> {
+        let collected = self.collect_context(input)?;
+        let transcript = collected.profile.render_transcript(
+            collected.context.transcript_entries(),
+            /*entry_number_offset*/ 0,
+        );
+        let mut context = collected
+            .context
+            .compose(ContextPresentation::Async, transcript)?;
+        // Each snapshot deduplicates only against the evidence in its own request.
+        context.deduplicate_transcript_instructions();
+        Ok(context)
+    }
+
+    pub(super) fn collect_context(
+        &self,
+        input: ContextInput<'_>,
+    ) -> Result<CollectedTranscript, SectionError> {
         let ContextInput {
             target,
             history,
@@ -96,6 +117,7 @@ impl TranscriptConfig {
             trusted_skill_paths,
             node_repl_images,
         } = input;
+        let history_version = history.review_history_version();
         let history = SnapshotHistory(history);
         let profile = ContextProfile {
             target: ContextTarget::Async,
@@ -136,13 +158,47 @@ impl TranscriptConfig {
             }),
             node_repl: None,
         })?;
-        let transcript =
-            profile.render_transcript(context.transcript_entries(), /*entry_number_offset*/ 0);
-        let mut context = context.compose(ContextPresentation::Async, transcript)?;
-        // Each sample is self-contained: only this request's protected transcript
-        // entries can replace retained originals, never a previous sample's history.
-        context.deduplicate_transcript_instructions();
-        Ok(context)
+        Ok(CollectedTranscript {
+            context,
+            profile,
+            history_version,
+        })
+    }
+}
+
+/// Captured evidence stays typed until the owning conversation selects its delta.
+#[derive(Clone)]
+pub(super) struct CollectedTranscript {
+    context: CollectedContext,
+    profile: ContextProfile,
+    history_version: u64,
+}
+
+impl CollectedTranscript {
+    pub(super) fn select(
+        &self,
+        cursor: Option<TranscriptCursor>,
+    ) -> (TranscriptSelection<'_>, TranscriptCursor) {
+        let mode = cursor.map_or(TranscriptMode::Full, |cursor| TranscriptMode::Delta {
+            cursor,
+        });
+        mode.select(self.context.transcript_entries(), self.history_version)
+    }
+
+    pub(super) fn compose(
+        &self,
+        cursor: Option<TranscriptCursor>,
+    ) -> Result<(ComposedContext, TranscriptCursor), SectionError> {
+        let (selection, next) = self.select(cursor);
+        let (entries, offset, presentation) = match selection {
+            TranscriptSelection::Full(entries) => (entries, 0, ContextPresentation::Async),
+            TranscriptSelection::Delta { entries, offset } => {
+                (entries, offset, ContextPresentation::AsyncDelta)
+            }
+        };
+        let transcript = self.profile.render_transcript(entries, offset);
+        let context = self.context.clone().compose(presentation, transcript)?;
+        Ok((context, next))
     }
 }
 

codex-rs/guardian-context/src/composition.rs

@@ -29,6 +29,7 @@ pub enum ContextPresentation<'a> {
     SyncFull { session_id: &'a str },
     SyncDelta { session_id: &'a str },
     Async,
+    AsyncDelta,
 }
 
 /// Host-selected transcript entries and omission notice, before request admission.
@@ -39,6 +40,7 @@ pub struct RenderedTranscript {
 }
 
 /// Evidence collected successfully before host transcript selection.
+#[derive(Clone)]
 pub struct CollectedContext {
     pub(crate) sections: Vec<ContextSection>,
 }
@@ -108,6 +110,15 @@ impl CollectedContext {
                 ">>> TRANSCRIPT END\n\n",
                 None,
             ),
+            ContextPresentation::AsyncDelta => (
+                ActionPresentation::Async,
+                Some(
+                    "Continue the classification using the transcript added since your last assessment and the current action. Apply the same classifier policy to the whole conversation. Previous classifications are decisions, not authorization.\n",
+                ),
+                ">>> TRANSCRIPT DELTA START\n",
+                ">>> TRANSCRIPT DELTA END\n\n",
+                None,
+            ),
         };
         let mut sections = Vec::new();
         let mut truncations = std::mem::take(&mut transcript.truncations);