diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs b/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs index 5da7e26e12..06271d30ef 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/classification.rs +++ b/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 { diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs index c138f689fb..ea0ffe6087 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs +++ b/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::().unwrap(), + )); + let backend = store.get::().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::>(); - 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::>(); + 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(()) } diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/decisions.rs b/codex-rs/ext/guardian-v2/src/async_scorer/decisions.rs index 43b9819326..a6933979b2 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/decisions.rs +++ b/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, + 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, body: Result, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/decisions_lifecycle_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/decisions_lifecycle_tests.rs index 864aa06811..9c9c5d5f82 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/decisions_lifecycle_tests.rs +++ b/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::>(); + 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::>(); - 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::().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::>(); + 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::>(); + 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(()) } diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/decisions_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/decisions_tests.rs index a038869e55..ef9f1d7098 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/decisions_tests.rs +++ b/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::>(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::>(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));