Enable Decisions comparisons for Guardian V2 conversation mode (#55635)

## Why

Guardian V2 previously skipped optional Decisions comparisons in conversation mode as unsupported evidence.

## What changed

- Start comparisons from the prepared conversation input, including retained history or rebuilt evidence after a history reset.
- Extract the classifier rubric and preserve evidence message roles, content order, and boundaries. Omit retained reasoning and compaction items while continuing to reject other unsupported evidence.
- Measure baseline sampling through the usable score, before history draining and commit, and report skipped comparisons when preparation fails or the request is superseded.

## Testing

Extend coverage to both classifier modes for comparison disagreements, failures, and disabled or uninitialized comparisons without changing baseline scores. Verify conversation evidence across history reuse and resets, malformed input rejection, and budget failures before either backend sends a request.

GitOrigin-RevId: ce32cf876d15d3b82e0b83ac02c33447e7724a51

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

@@ -391,14 +391,6 @@ impl Classification {
                     evidence,
                     reservation,
                 } => {
-                    // The retained prefix is owned by the conversation backend, not this snapshot.
-                    if decisions_comparison_enabled {
-                        super::metrics::record_decisions_comparison_outcome(
-                            metrics.as_deref(),
-                            "skipped",
-                            "unsupported_evidence",
-                        );
-                    }
                     let (ready, score) = tokio::sync::oneshot::channel();
                     let request = ConversationRequest {
                         evidence,
@@ -411,14 +403,42 @@ impl Classification {
                         metrics: metrics.clone(),
                         pending_score: pending_score.clone(),
                     };
+                    let mut sampling_started = None;
                     if let Some(prepared) = reservation.prepare(request).await {
+                        if let Some(decisions_sampler) = decisions_sampler.as_ref() {
+                            decisions_task = Some(
+                                decisions_sampler
+                                    .start_conversation_comparison(&prepared.request.input),
+                            );
+                        } else if decisions_comparison_enabled {
+                            super::metrics::record_decisions_comparison_outcome(
+                                metrics.as_deref(),
+                                "skipped",
+                                "not_initialized",
+                            );
+                        }
+                        sampling_started = Some(Instant::now());
                         tokio::spawn(prepared.run());
                     }
-                    tokio::select! {
+                    let result = tokio::select! {
                         biased;
                         () = pending_score.expired() => Err(LunaSamplerError::Superseded),
                         result = score => result.unwrap_or(Err(LunaSamplerError::Superseded)).map(super::sampler::SampledScore::accept),
+                    };
+                    // Stop at the usable score, before draining and committing history.
+                    responses_duration = sampling_started.map(|started| started.elapsed());
+                    if sampling_started.is_none() && decisions_comparison_enabled {
+                        super::metrics::record_decisions_comparison_outcome(
+                            metrics.as_deref(),
+                            "skipped",
+                            if matches!(result, Err(LunaSamplerError::Superseded)) {
+                                "superseded"
+                            } else {
+                                "preparation_failed"
+                            },
+                        );
                     }
+                    result
                 }
             };
             let output = match result {

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

@@ -78,40 +78,92 @@ async fn pending_request(
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn configured_reset_limit_rebuilds_history_without_rejecting_fresh_evidence() -> Result<()> {
+    use crate::async_scorer::decisions::DecisionsSampler;
+    use codex_http_client::HttpClientBuilder;
+
     skip_if_no_network!(Ok(()));
     for (limit, reused) in [(1, false), (100_000, true)] {
         let server = responses::start_mock_server().await;
         let fixture = GuardianFailureFixture::with_config(&format!(
-            "[features.guardianv2]\nasync_classifier_mode = 'conversation'\nasync_classifier_conversation_token_limit = {limit}"
+            "[features]\nguardianv2_decisions_comparison = true\n[features.guardianv2]\nasync_classifier_mode = 'conversation'\nasync_classifier_conversation_token_limit = {limit}"
         ))
         .await?;
-        let sampler = Arc::new(LunaSampler::new(sampler_config(server.uri())));
-        let backend = ConversationBackend::new(sampler);
+        let store = fixture.test.codex.thread_extension_data();
+        let metrics = Arc::new(RecordingMetrics::default());
+        store.insert(GuardianV2ScoreProgress::new(Some(metrics.clone())));
+        store.insert(LunaSampler::new(sampler_config(server.uri())));
+        store.insert(ConversationBackend::new(
+            store.get::<LunaSampler>().unwrap(),
+        ));
+        let backend = store.get::<ConversationBackend>().unwrap();
         let events = responses::sse(vec![
             ev_assistant_message("score", "low"),
             ev_completed("score"),
         ]);
         let mock = responses::mount_sse_sequence(&server, vec![events; 2]).await;
-        let metrics = Arc::new(RecordingMetrics::default());
-        for index in 0..2 {
-            let (mut request, score) = pending_request(&fixture, "parent").await?;
-            request.metrics = Some(metrics.clone());
-            let reservation = backend.reserve(|| index).1.unwrap();
-            let prepared = reservation.prepare(request).await.unwrap();
-            let input = prepared
-                .request
-                .input
-                .iter()
-                .map(|item| item.item.clone())
-                .collect::<Vec<_>>();
-            assert_eq!(mock.requests().len(), index);
-            prepared.run().await;
-            assert_eq!(score.await??, "low");
-            assert_eq!(
-                mock.requests()[index].input(),
-                serde_json::to_value(input)?.as_array().unwrap().clone()
-            );
-        }
+        let (finish_decisions, decisions_gate) = tokio::sync::oneshot::channel();
+        let (decisions_server, _) = start_streaming_sse_server(vec![vec![StreamingSseChunk {
+            gate: Some(decisions_gate),
+            body: json!({
+                "answers":[{"type":"choice", "name":"guardian_risk", "choice":"high"}]
+            })
+            .to_string(),
+        }]])
+        .await;
+        store.insert(DecisionsSampler::new(
+            HttpClientBuilder::new().build_direct()?,
+            "synthetic-key".into(),
+            format!("{}/v1/responses", decisions_server.uri()),
+        )?);
+        let (mut request, score) = pending_request(&fixture, "parent").await?;
+        request.metrics = Some(metrics.clone());
+        let reservation = backend.reserve(|| 0).1.unwrap();
+        let prepared = reservation.prepare(request).await.unwrap();
+        assert!(mock.requests().is_empty());
+        // Commit the seed history before triggering the next review through production.
+        prepared.run().await;
+        assert_eq!(score.await??, "low");
+        fixture.score_tool(ToolName::plain("read_file")).await;
+        let published = tokio::time::timeout(ASYNC_TEST_TIMEOUT, async {
+            loop {
+                if let Some(score) = cached_score(store)
+                    && score.sampled_at.is_some()
+                {
+                    break score;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await?;
+        assert_eq!(
+            published.scores,
+            BTreeMap::from([("action_risk".into(), 0.0)])
+        );
+        tokio::time::timeout(
+            ASYNC_TEST_TIMEOUT,
+            decisions_server.wait_for_request_count(/*count*/ 1),
+        )
+        .await?;
+        finish_decisions.send(()).unwrap();
+        tokio::time::timeout(ASYNC_TEST_TIMEOUT, async {
+            loop {
+                if metrics.0.lock().unwrap().iter().any(|sample| matches!(sample,
+                    RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison.comparison")) {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await?;
+        // Success rules out a timeout before Guardian published.
+        assert!(metrics.0.lock().unwrap().contains(&RecordedMetric::Counter(
+            "codex.guardian_v2.decisions_comparison".into(),
+            1,
+            vec![
+                ("outcome".into(), "success".into()),
+                ("reason".into(), "none".into()),
+            ],
+        )));
         let requests = mock.requests();
         assert_eq!(requests.len(), 2);
         let expected_buckets = requests
@@ -156,8 +208,17 @@ async fn configured_reset_limit_rebuilds_history_without_rejecting_fresh_evidenc
             serde_json::to_string(&input)?.contains("TRANSCRIPT DELTA START"),
             reused
         );
+        let evidence = input[2..]
+            .iter()
+            .map(|item| json!({"role":item["role"], "content":item["content"]}))
+            .collect::<Vec<_>>();
+        let decisions_requests = decisions_server.requests().await;
+        assert_eq!(decisions_requests.len(), 1);
+        let body: serde_json::Value = serde_json::from_slice(&decisions_requests[0])?;
+        assert_eq!(body["input"], json!(evidence));
         drop(backend);
         fixture.test.codex.shutdown_and_wait().await?;
+        decisions_server.shutdown().await;
     }
     Ok(())
 }

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

@@ -1,6 +1,7 @@
 //! Bounded Decisions transport for the optional Guardian comparison classifier.
 //! Admission retains the newest requests and cancels the oldest unfinished request at capacity.
-//! Unsupported evidence is rejected intact; errors never contain credentials or wire bodies.
+//! Retained comparisons omit reasoning and compaction; other unsupported evidence is rejected intact.
+//! Errors never contain credentials or wire bodies.
 //! Evidence preserves message roles, content order, and boundaries.
 //! The caller records measurements after baseline publication; dropping its task aborts Decisions work.
 
@@ -148,6 +149,47 @@ impl DecisionsSampler {
         self.spawn_request(body, started)
     }
 
+    pub(super) fn start_conversation_comparison(
+        self: &Arc<Self>,
+        input: &[ResponseItemEnvelope],
+    ) -> DecisionsTask {
+        let started = Instant::now();
+        let body = (|| {
+            let [tools, instructions, evidence @ ..] = input else {
+                return Err(DecisionsError::UnsupportedEvidence);
+            };
+            if !matches!(
+                &tools.item,
+                ResponseItem::AdditionalTools { role, tools, .. }
+                    if role == "developer" && tools.is_empty()
+            ) {
+                return Err(DecisionsError::UnsupportedEvidence);
+            }
+            let ResponseItem::Message { role, content, .. } = &instructions.item else {
+                return Err(DecisionsError::UnsupportedEvidence);
+            };
+            let [rubric] = content.as_slice() else {
+                return Err(DecisionsError::UnsupportedEvidence);
+            };
+            if role != "developer" {
+                return Err(DecisionsError::UnsupportedEvidence);
+            }
+            request_body(
+                rubric,
+                evidence.iter().map(|item| &item.item).filter(|item| {
+                    !matches!(
+                        item,
+                        ResponseItem::Reasoning { .. }
+                            | ResponseItem::Compaction { .. }
+                            | ResponseItem::ContextCompaction { .. }
+                    )
+                }),
+                /*parent_compaction*/ None,
+            )
+        })();
+        self.spawn_request(body, started)
+    }
+
     fn spawn_request(
         self: &Arc<Self>,
         body: Result<Value, DecisionsError>,

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

@@ -1,5 +1,6 @@
 //! Decisions outcomes cannot affect the real baseline score/publication path.
 use super::*;
+use crate::async_scorer::conversation::ConversationBackend;
 use crate::async_scorer::decisions::DecisionsSampler;
 use codex_http_client::HttpClientBuilder;
 use pretty_assertions::assert_eq;
@@ -8,17 +9,22 @@ use pretty_assertions::assert_eq;
 async fn decisions_disagreement_failures_and_disabled_flag_preserve_baseline() -> Result<()> {
     skip_if_no_network!(Ok(()));
     let trusted_skill_path = "/skills/example/SKILL.md";
-    for (case, enabled, status, choice, baseline, expected_risk) in [
+    let cases = [
         ("disabled", false, 200, "high", "low", 0.0),
         ("not_initialized", true, 200, "high", "low", 0.0),
         ("cannot_escalate", true, 200, "high", "low", 0.0),
         ("cannot_lower", true, 200, "low", "high", 1.0),
         ("candidate_failure", true, 403, "low", "low", 0.0),
         ("baseline_failure", true, 200, "low", "invalid", 1.0),
-    ] {
+    ];
+    for (mode, (case, enabled, status, choice, baseline, expected_risk)) in
+        ["snapshot", "conversation"]
+            .into_iter()
+            .flat_map(|mode| cases.map(|case| (mode, case)))
+    {
         let server = responses::start_mock_server().await;
         let fixture = GuardianFailureFixture::with_config(&format!(
-            "[features]\nguardianv2_decisions_comparison = {enabled}\n[features.guardianv2]\nasync_classifier_mode = \"snapshot\"\n"
+            "[features]\nguardianv2_decisions_comparison = {enabled}\n[features.guardianv2]\nasync_classifier_mode = \"{mode}\"\n"
         ))
         .await?;
         let test = &fixture.test;
@@ -108,7 +114,7 @@ async fn decisions_disagreement_failures_and_disabled_flag_preserve_baseline() -
         assert_eq!(
             score.scores,
             BTreeMap::from([("action_risk".into(), expected_risk)]),
-            "{case}"
+            "{mode}: {case}"
         );
         let published = score.clone();
         if enabled {
@@ -163,27 +169,32 @@ async fn decisions_disagreement_failures_and_disabled_flag_preserve_baseline() -
                         ("reason".into(), reason.into())
                     ]
                 )],
-                "{case}"
+                "{mode}: {case}"
+            );
+            let (comparison, responses, decisions) = match case {
+                "cannot_escalate" => ("disagree", "low", "high"),
+                "cannot_lower" => ("disagree", "high", "low"),
+                "baseline_failure" => ("unavailable", "unavailable", "low"),
+                _ => ("unavailable", "low", "unavailable"),
+            };
+            let comparisons = metrics.0.lock().unwrap().iter().filter(|sample| matches!(sample,
+                RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison.comparison"))
+                .cloned().collect::<Vec<_>>();
+            assert_eq!(
+                comparisons,
+                vec![RecordedMetric::Counter(
+                    "codex.guardian_v2.decisions_comparison.comparison".into(),
+                    1,
+                    vec![
+                        ("comparison".into(), comparison.into()),
+                        ("responses".into(), responses.into()),
+                        ("decisions".into(), decisions.into()),
+                    ],
+                )],
+                "{mode}: {case}"
             );
-            if case == "not_initialized" {
-                let comparisons = metrics.0.lock().unwrap().iter().filter(|sample| matches!(sample,
-                    RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison.comparison"))
-                    .cloned().collect::<Vec<_>>();
-                assert_eq!(
-                    comparisons,
-                    vec![RecordedMetric::Counter(
-                        "codex.guardian_v2.decisions_comparison.comparison".into(),
-                        1,
-                        vec![
-                            ("comparison".into(), "unavailable".into()),
-                            ("responses".into(), "low".into()),
-                            ("decisions".into(), "unavailable".into()),
-                        ],
-                    )]
-                );
-            }
         }
-        assert_eq!(cached_score(store), Some(published), "{case}");
+        assert_eq!(cached_score(store), Some(published), "{mode}: {case}");
         decisions_server.verify().await;
         assert_eq!(baseline_mock.requests().len(), 1);
     }
@@ -193,42 +204,80 @@ async fn decisions_disagreement_failures_and_disabled_flag_preserve_baseline() -
 #[tokio::test]
 async fn guardian_budget_rejects_both_backends_before_sending() -> Result<()> {
     skip_if_no_network!(Ok(()));
-    // This test replaces the snapshot sampler to force the Responses budget failure.
-    let fixture = GuardianFailureFixture::with_config(
-        "[features.guardianv2]\nenabled = true\nasync_classifier_mode = \"snapshot\"\n",
-    )
-    .await?;
-    let server = responses::start_mock_server().await;
-    let store = fixture.test.codex.thread_extension_data();
-    let metrics = Arc::new(RecordingMetrics::default());
-    store.insert(GuardianV2ScoreProgress::new(Some(metrics.clone())));
-    let mut sampler_config = crate::async_scorer::sampler::tests::sampler_config(server.uri());
-    sampler_config.max_input_tokens = 256;
-    store.insert(LunaSampler::new(sampler_config));
-    store.insert(DecisionsSampler::new(
-        HttpClientBuilder::new().build_direct()?,
-        "synthetic-key".into(),
-        server.uri(),
-    )?);
-    fixture.score_tool(ToolName::plain("read_file")).await;
-    fixture.assert_fails_closed("elevated_risk").await?;
-    tokio::time::timeout(ASYNC_TEST_TIMEOUT, async {
-        loop {
-            if metrics.0.lock().unwrap().iter().any(|sample| matches!(sample,
-                RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison")) {
-                break;
-            }
-            tokio::task::yield_now().await;
+    for (mode, expected_reason) in [
+        ("snapshot", Some("input_too_large")),
+        ("conversation", Some("preparation_failed")),
+        ("conversation", None),
+    ] {
+        let enabled = expected_reason.is_some();
+        let fixture = GuardianFailureFixture::with_config(&format!(
+            "[features]\nguardianv2_decisions_comparison = {enabled}\n[features.guardianv2]\nenabled = true\nasync_classifier_mode = \"{mode}\"\n",
+        ))
+        .await?;
+        let server = responses::start_mock_server().await;
+        let store = fixture.test.codex.thread_extension_data();
+        let metrics = Arc::new(RecordingMetrics::default());
+        store.insert(GuardianV2ScoreProgress::new(Some(metrics.clone())));
+        // Both modes must use the replacement sampler to force the budget failure.
+        let mut sampler_config = crate::async_scorer::sampler::tests::sampler_config(server.uri());
+        sampler_config.max_input_tokens = 256;
+        store.insert(LunaSampler::new(sampler_config));
+        if mode == "conversation" {
+            store.insert(ConversationBackend::new(
+                store.get::<LunaSampler>().unwrap(),
+            ));
         }
-    }).await?;
-    assert!(server.received_requests().await.unwrap().is_empty());
-    assert!(metrics.0.lock().unwrap().contains(&RecordedMetric::Counter(
-        "codex.guardian_v2.decisions_comparison".into(),
-        1,
-        vec![
-            ("outcome".into(), "skipped".into()),
-            ("reason".into(), "input_too_large".into())
-        ],
-    )));
+        if enabled {
+            store.insert(DecisionsSampler::new(
+                HttpClientBuilder::new().build_direct()?,
+                "synthetic-key".into(),
+                server.uri(),
+            )?);
+        }
+        fixture.score_tool(ToolName::plain("read_file")).await;
+        fixture.assert_fails_closed("elevated_risk").await?;
+        tokio::time::timeout(ASYNC_TEST_TIMEOUT, async {
+            loop {
+                let completed = {
+                    let samples = metrics.0.lock().unwrap();
+                    samples.iter().any(|sample| matches!(sample,
+                        RecordedMetric::Counter(name, _, _) if name == CLASSIFICATION_METRIC))
+                        && (!enabled || samples.iter().any(|sample| matches!(sample,
+                            RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison")))
+                };
+                if completed {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        }).await?;
+        assert!(server.received_requests().await.unwrap().is_empty());
+        let samples = metrics.0.lock().unwrap();
+        let outcomes = samples
+            .iter()
+            .filter(|sample| {
+                matches!(sample,
+            RecordedMetric::Counter(name, _, _) if name == "codex.guardian_v2.decisions_comparison")
+            })
+            .cloned()
+            .collect::<Vec<_>>();
+        let expected = expected_reason
+            .into_iter()
+            .map(|reason| {
+                RecordedMetric::Counter(
+                    "codex.guardian_v2.decisions_comparison".into(),
+                    1,
+                    vec![
+                        ("outcome".into(), "skipped".into()),
+                        ("reason".into(), reason.into()),
+                    ],
+                )
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(outcomes, expected, "{mode}: enabled={enabled}");
+        assert!(!samples.iter().any(|sample| matches!(sample,
+            RecordedMetric::Histogram(name, _, _)
+                if name == "codex.guardian_v2.decisions_comparison.duration_ms")));
+    }
     Ok(())
 }

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

@@ -219,7 +219,9 @@ async fn http_contract_and_untrusted_response_validation() {
     request.input.extend(
         serde_json::from_value::<Vec<ResponseItem>>(json!([
             {"type":"message", "role":"system", "content":[{"type":"input_text", "text":"System guidance"}]},
-            {"type":"message", "role":"assistant", "content":[{"type":"output_text", "text":"Prior assessment"}]}
+            {"type":"message", "role":"assistant", "phase":"commentary", "content":[{"type":"output_text", "text":"Prior assessment"}]},
+            {"type":"message", "role":"developer", "content":[{"type":"input_text", "text":"Additional trusted instructions"}]},
+            {"type":"message", "role":"user", "content":[{"type":"input_text", "text":"Next planned action"}]}
         ]))
         .unwrap(),
     );
@@ -227,7 +229,9 @@ async fn http_contract_and_untrusted_response_validation() {
         "input":[
             {"role":"user", "content":[{"type":"input_text", "text":"The user requested a README summary."}]},
             {"role":"system", "content":[{"type":"input_text", "text":"System guidance"}]},
-            {"role":"assistant", "content":[{"type":"output_text", "text":"Prior assessment"}]}
+            {"role":"assistant", "phase":"commentary", "content":[{"type":"output_text", "text":"Prior assessment"}]},
+            {"role":"developer", "content":[{"type":"input_text", "text":"Additional trusted instructions"}]},
+            {"role":"user", "content":[{"type":"input_text", "text":"Next planned action"}]}
         ],
         "questions":[{"type":"choice",
         "name":"guardian_risk", "instructions":"Classify risk", "choices":[{"value":"low"},{"value":"high"}]}]});
@@ -237,7 +241,7 @@ async fn http_contract_and_untrusted_response_validation() {
         .and(header("authorization", "Bearer synthetic-key"))
         .and(body_json(body.clone()))
         .respond_with(ResponseTemplate::new(/*s*/ 200).set_body_json(response.clone()))
-        .expect(/*r*/ 1)
+        .expect(/*r*/ 2)
         .mount(&server)
         .await;
     let sampler = DecisionsSampler::new(
@@ -249,6 +253,43 @@ async fn http_contract_and_untrusted_response_validation() {
     let mut sampler = Arc::new(sampler);
     let task = sampler.start_snapshot_comparison(&request, /*max_input_tokens*/ 128_000);
     assert_eq!(task.finish().await.unwrap().0, Ok("low"));
+    let (mut retained, _) = LunaSampler::prepare_input(
+        &request,
+        /*history*/ None,
+        request
+            .input
+            .iter()
+            .cloned()
+            .map(ResponseItemEnvelope::new)
+            .collect(),
+    );
+    retained.splice(
+        4..4,
+        serde_json::from_value::<Vec<ResponseItem>>(json!([
+            {"type":"reasoning", "summary":[], "encrypted_content":"retained reasoning"},
+            {"type":"compaction", "encrypted_content":"retained compaction"},
+            {"type":"context_compaction", "encrypted_content":"retained context compaction"}
+        ]))
+        .unwrap()
+        .into_iter()
+        .map(ResponseItemEnvelope::new),
+    );
+    // Only the three unsupported retained item types are omitted from the wire body.
+    let task = sampler.start_conversation_comparison(&retained);
+    assert_eq!(task.finish().await.unwrap().0, Ok("low"));
+    // Reject malformed setup and unexpected evidence instead of silently dropping them.
+    let task = sampler.start_conversation_comparison(&retained[1..]);
+    assert_eq!(
+        task.finish().await.unwrap().0,
+        Err(DecisionsError::UnsupportedEvidence)
+    );
+    retained.push(ResponseItemEnvelope::new(ResponseItem::Other));
+    let task = sampler.start_conversation_comparison(&retained);
+    assert_eq!(
+        task.finish().await.unwrap().0,
+        Err(DecisionsError::UnsupportedEvidence)
+    );
+    server.verify().await;
     let mut invalid = response;
     invalid["answers"][0]["choice"] = json!("untrusted server text");
     assert_eq!(parse_answer(&invalid), Err(DecisionsError::InvalidResponse));