diff --git a/codex-rs/core/src/tasks/lifecycle.rs b/codex-rs/core/src/tasks/lifecycle.rs index afecaede5a..07a2c850a8 100644 --- a/codex-rs/core/src/tasks/lifecycle.rs +++ b/codex-rs/core/src/tasks/lifecycle.rs @@ -21,6 +21,9 @@ impl Session { ) { let metadata: Arc = turn_context.turn_metadata_state.clone(); turn_context.extension_data.insert(metadata); + turn_context + .extension_data + .insert(turn_context.session_telemetry.clone()); let collaboration_mode = turn_context.collaboration_mode(); for contributor in self.services.extensions.turn_lifecycle_contributors() { if contributor.turn_start_phase(&self.services.thread_extension_data) != phase { diff --git a/codex-rs/core/tests/suite/guardian_mcp_elicitation.rs b/codex-rs/core/tests/suite/guardian_mcp_elicitation.rs index ab6bf48c1f..1193fe6648 100644 --- a/codex-rs/core/tests/suite/guardian_mcp_elicitation.rs +++ b/codex-rs/core/tests/suite/guardian_mcp_elicitation.rs @@ -8,6 +8,8 @@ use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::McpServerContribution; use codex_extension_api::McpServerContributionContext; use codex_extension_api::McpServerContributor; +use codex_otel::MetricsClient; +use codex_otel::MetricsConfig; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::items::McpToolCallStatus; use codex_protocol::items::TurnItem; @@ -27,9 +29,12 @@ use core_test_support::test_codex::local_requests; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use core_test_support::wait_for_mcp_server; +use opentelemetry_sdk::metrics::data::AggregatedMetrics; +use opentelemetry_sdk::metrics::data::MetricData; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; +use std::collections::BTreeMap; use std::sync::Arc; use std::time::Duration; use test_case::test_case; @@ -955,6 +960,10 @@ async fn yielded_code_mode_elicitation_keeps_live_invocation_metadata() -> Resul } }))?; let mut builder = test_codex() + .with_model_info_override("gpt-5.2", |model| { + model.tool_mode = Some(codex_protocol::openai_models::ToolMode::CodeMode); + model.experimental_supported_tools = vec!["test_sync_tool".to_string()]; + }) .with_model_info_override("gpt-5.4", |model| { model.tool_mode = Some(codex_protocol::openai_models::ToolMode::CodeMode); model.experimental_supported_tools = vec!["test_sync_tool".to_string()]; @@ -968,7 +977,16 @@ async fn yielded_code_mode_elicitation_keeps_live_invocation_metadata() -> Resul config.approvals_reviewer = ApprovalsReviewer::User; config.mcp_servers.set(mcp_servers).unwrap(); }); - let test = builder.build_with_auto_env(&server).await?; + let mut test = builder.build_with_auto_env(&server).await?; + let metrics = MetricsClient::new( + MetricsConfig::in_memory("test", "test", "1", Default::default()).with_runtime_reader(), + )?; + let mut options = test.start_thread_options().await; + options.history_mode = Some(codex_protocol::protocol::ThreadHistoryMode::Legacy); + options.thread_extension_init.insert(metrics.clone()); + let thread = test.thread_manager.start_thread(options).await?; + test.codex.shutdown_and_wait().await?; + test.codex = thread.thread; wait_for_mcp_server(&test.codex, "node_repl").await?; let barrier = r#"await tools.test_sync_tool({barrier: { id: "elicitation-origin", participants: 2, timeout_ms: 60000 @@ -1043,10 +1061,16 @@ async fn yielded_code_mode_elicitation_keeps_live_invocation_metadata() -> Resul ) .await; test.codex - .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { - text: "Run B and wait for A.".into(), - text_elements: Vec::new(), - }])) + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "Run B and wait for A.".into(), + text_elements: Vec::new(), + }]) + .with_thread_settings(ThreadSettingsOverrides { + model: Some("gpt-5.2".into()), + ..Default::default() + }), + ) .await?; let mut invocation_id = None; let mut assessment_target = None; @@ -1081,5 +1105,37 @@ async fn yielded_code_mode_elicitation_keeps_live_invocation_metadata() -> Resul .count(), 1 ); + test.codex.shutdown_and_wait().await?; + let mut samples = guardian_blocking_samples(&metrics, "model")?; + assert_eq!(samples.remove("gpt-5.4"), Some(0.0)); + assert!(samples.remove("gpt-5.2").unwrap() > 0.0); + assert!(samples.is_empty()); Ok(()) } + +/// Reads per-turn samples shared by command and MCP approval regressions. +pub(super) fn guardian_blocking_samples( + metrics: &MetricsClient, + label: &str, +) -> Result> { + let snapshot = metrics.snapshot()?; + let metric = snapshot + .scope_metrics() + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + .find(|metric| metric.name() == "codex.guardian.turn.blocking_ms") + .context("blocking histogram")?; + let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() else { + panic!("expected blocking histogram"); + }; + histogram + .data_points() + .map(|point| { + let attribute = point + .attributes() + .find(|tag| tag.key.as_str() == label) + .context("blocking metric label")?; + assert_eq!(point.count(), 1); + Ok((attribute.value.as_str().to_string(), point.sum())) + }) + .collect() +} diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index bd31aa799e..a0ef8c25b1 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -26,6 +26,8 @@ use codex_features::Feature; use codex_history::RolloutItem; use codex_login::CodexAuth; use codex_models_manager::manager::RefreshStrategy; +use codex_otel::MetricsClient; +use codex_otel::MetricsConfig; use codex_protocol::ThreadId; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::config_types::WindowsSandboxLevel; @@ -95,6 +97,7 @@ use wiremock::http::Method; use wiremock::matchers::method; use wiremock::matchers::path_regex; +use super::guardian_mcp_elicitation::guardian_blocking_samples; use super::network_approval::guardian_parent_catalog; use super::rmcp_client::remote_aware_environment_id; use super::rmcp_client::remote_aware_stdio_server_bin; @@ -2806,7 +2809,16 @@ async fn yielded_code_mode_denials_interrupt_the_servicing_turn() -> Result<()> config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); config.approvals_reviewer = ApprovalsReviewer::AutoReview; }); - let test = builder.build_with_auto_env(&server).await?; + let mut test = builder.build_with_auto_env(&server).await?; + let metrics = MetricsClient::new( + MetricsConfig::in_memory("test", "test", "1", Default::default()).with_runtime_reader(), + )?; + let mut options = test.start_thread_options().await; + options.history_mode = Some(codex_protocol::protocol::ThreadHistoryMode::Legacy); + options.thread_extension_init.insert(metrics.clone()); + let thread = test.thread_manager.start_thread(options).await?; + test.codex.shutdown_and_wait().await?; + test.codex = thread.thread; let barrier = r#"await tools.test_sync_tool({barrier: { id: "guardian-origin", participants: 2, timeout_ms: 60000 }});"#; @@ -2950,5 +2962,145 @@ for (let index = 0; index < 6; index++) {{ 1 ); } + test.codex.shutdown_and_wait().await?; + let mut samples = guardian_blocking_samples(&metrics, "outcome")?; + assert_eq!(samples.remove("completed"), Some(0.0)); + assert!(samples.remove("aborted").unwrap() > 0.0); + assert!(samples.is_empty()); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_blocking_follows_a_pending_review_across_turns() -> Result<()> { + skip_if_no_network!(Ok(())); + skip_if_sandbox!(Ok(())); + + let code = r#"// @exec: {"yield_time_ms": 1} +text(await tools.exec_command({ + cmd: "echo spanning-review", + sandbox_permissions: "require_escalated", + justification: "Keep a Guardian review pending across turns." +}));"#; + let (finish_a, a_gate) = tokio::sync::oneshot::channel(); + let (finish_review, review_gate) = tokio::sync::oneshot::channel(); + let chunk = |gate, events| StreamingSseChunk { + gate, + body: sse(events), + }; + let (streaming, _) = start_streaming_sse_server(vec![ + vec![ + chunk( + /*gate*/ None, + vec![ + ev_response_created("a"), + ev_custom_tool_call("cell-a", "exec", code), + ], + ), + chunk(Some(a_gate), vec![ev_completed("a")]), + ], + vec![chunk( + Some(review_gate), + vec![ + ev_response_created("review"), + ev_assistant_message("assessment", r#"{"outcome":"deny"}"#), + ev_completed("review"), + ], + )], + vec![chunk(/*gate*/ None, vec![ev_completed("a-finished")])], + vec![chunk( + /*gate*/ None, + vec![ + ev_function_call("wait-a", "wait", r#"{"cell_id":"1","yield_time_ms":60000}"#), + ev_completed("b"), + ], + )], + vec![chunk(/*gate*/ None, vec![ev_completed("b-finished")])], + ]) + .await; + let server = start_mock_server().await; + let base_url = format!("{}/v1", streaming.uri()); + let mut builder = test_codex().with_config(move |config| { + config.features.enable(Feature::CodeMode).unwrap(); + config + .features + .disable(Feature::EnableRequestCompression) + .unwrap(); + config.model_provider.base_url = Some(base_url); + config.model_provider.supports_websockets = false; + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + }); + for model in ["gpt-5.2", "gpt-5.4"] { + builder = builder.with_model_info_override(model, |info| { + info.tool_mode = Some(codex_protocol::openai_models::ToolMode::CodeMode); + }); + } + let mut test = builder + .with_model("gpt-5.4") + .build_with_auto_env(&server) + .await?; + let metrics = MetricsClient::new( + MetricsConfig::in_memory("test", "test", "1", Default::default()).with_runtime_reader(), + )?; + let mut options = test.start_thread_options().await; + options.history_mode = Some(codex_protocol::protocol::ThreadHistoryMode::Legacy); + options.thread_extension_init.insert(metrics.clone()); + let thread = test.thread_manager.start_thread(options).await?; + test.codex.shutdown_and_wait().await?; + test.codex = thread.thread; + + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Start A and leave its cell running.".into(), + text_elements: Vec::new(), + }])) + .await?; + // The review has reached its backend before A is allowed to finish. + streaming.wait_for_request_count(/*count*/ 2).await; + finish_a.send(()).unwrap(); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let requests = streaming.requests().await; + let yielded: Value = serde_json::from_slice(&requests[2])?; + let output = yielded["input"] + .as_array() + .unwrap() + .iter() + .find(|item| item["call_id"] == "cell-a" && item["type"] == "custom_tool_call_output") + .unwrap(); + assert!( + output["output"] + .to_string() + .contains("Script running with cell ID 1"), + "{output}" + ); + + test.codex + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "Wait for A's cell.".into(), + text_elements: Vec::new(), + }]) + .with_thread_settings(ThreadSettingsOverrides { + model: Some("gpt-5.2".into()), + ..Default::default() + }), + ) + .await?; + // B is active while the same review is still blocked on its response gate. + streaming.wait_for_request_count(/*count*/ 4).await; + finish_review.send(()).unwrap(); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.codex.shutdown_and_wait().await?; + let samples = guardian_blocking_samples(&metrics, "model")?; + assert_eq!(samples.len(), 2); + assert!(samples["gpt-5.4"] > 0.0); + assert!(samples["gpt-5.2"] > 0.0); + streaming.shutdown().await; Ok(()) } diff --git a/codex-rs/ext/guardian-reviewer/src/blocking.rs b/codex-rs/ext/guardian-reviewer/src/blocking.rs new file mode 100644 index 0000000000..0daecf7f40 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/blocking.rs @@ -0,0 +1,149 @@ +//! Splits the union of thread-owned approval waits across active turns. +//! Turn completion exports one sample; pending reviews survive idle gaps and turn changes. +//! Saved durations are removed when their turn finishes or its handle drops. + +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::PoisonError; +use std::sync::Weak; +use std::time::Duration; +use std::time::Instant; + +/// Tracks pending approvals independently of the turn currently servicing them. +#[derive(Default)] +pub struct ReviewBlocking { + state: Mutex, +} + +#[derive(Default)] +struct State { + interval: Interval, + turn: Option, + completed: HashMap, +} + +/// Identifies a turn's sample and removes it when the turn is released. +pub struct ReviewBlockingTurn { + id: String, + tracker: Weak, + telemetry: Arc, +} + +impl State { + fn start_turn(&mut self, turn_id: &str, now: Instant) { + if let Some(previous) = self.turn.replace(turn_id.to_owned()) { + self.completed + .insert(previous, self.interval.finish_turn(now)); + } + self.interval.start_turn(now); + } + + fn finish_turn(&mut self, turn_id: &str, now: Instant) -> Option { + if self.turn.as_deref() == Some(turn_id) { + self.turn = None; + Some(self.interval.finish_turn(now)) + } else { + self.completed.remove(turn_id) + } + } +} + +#[derive(Default)] +struct Interval { + active: usize, + since: Option, + elapsed: Duration, +} + +impl Interval { + fn start_turn(&mut self, now: Instant) { + self.elapsed = Duration::ZERO; + self.since = (self.active > 0).then_some(now); + } + + fn finish_turn(&mut self, now: Instant) -> Duration { + if let Some(since) = self.since.take() { + self.elapsed += now.saturating_duration_since(since); + } + std::mem::take(&mut self.elapsed) + } + + fn enter(&mut self, now: Instant) { + if self.active == 0 { + self.since = Some(now); + } + self.active += 1; + } + fn leave(&mut self, now: Instant) { + self.active = self.active.saturating_sub(1); + if self.active == 0 + && let Some(since) = self.since.take() + { + self.elapsed += now.saturating_duration_since(since); + } + } +} + +impl ReviewBlocking { + pub fn start_turn( + self: &Arc, + turn_id: &str, + telemetry: Arc, + ) -> ReviewBlockingTurn { + self.state + .lock() + .unwrap_or_else(PoisonError::into_inner) + .start_turn(turn_id, Instant::now()); + ReviewBlockingTurn { + id: turn_id.to_owned(), + tracker: Arc::downgrade(self), + telemetry, + } + } + + pub fn finish_turn(&self, turn: &ReviewBlockingTurn, outcome: &'static str) { + let elapsed = self + .state + .lock() + .unwrap_or_else(PoisonError::into_inner) + .finish_turn(&turn.id, Instant::now()); + if let Some(elapsed) = elapsed { + turn.telemetry.record_duration( + "codex.guardian.turn.blocking_ms", + elapsed, + &[("outcome", outcome)], + ); + } + } + + pub(crate) fn enter(self: &Arc) -> BlockingGuard { + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + state.interval.enter(Instant::now()); + BlockingGuard(Arc::clone(self)) + } +} + +impl Drop for ReviewBlockingTurn { + fn drop(&mut self) { + if let Some(tracker) = self.tracker.upgrade() { + tracker + .state + .lock() + .unwrap_or_else(PoisonError::into_inner) + .finish_turn(&self.id, Instant::now()); + } + } +} + +pub(crate) struct BlockingGuard(Arc); +impl Drop for BlockingGuard { + fn drop(&mut self) { + let mut state = self.0.state.lock().unwrap_or_else(PoisonError::into_inner); + state.interval.leave(Instant::now()); + } +} + +#[cfg(test)] +#[path = "blocking_tests.rs"] +mod tests; diff --git a/codex-rs/ext/guardian-reviewer/src/blocking_tests.rs b/codex-rs/ext/guardian-reviewer/src/blocking_tests.rs new file mode 100644 index 0000000000..0d0a6553f4 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/blocking_tests.rs @@ -0,0 +1,68 @@ +//! Checks overlapping approvals and turn boundaries with an explicit clock. + +use super::*; +use pretty_assertions::assert_eq; + +#[test] +fn overlapping_reviews_span_turns_without_charging_idle_gaps() { + let start = Instant::now(); + let mut interval = Interval::default(); + interval.start_turn(start); + interval.enter(start); + interval.enter(start + Duration::from_millis(10)); + interval.leave(start + Duration::from_millis(30)); + assert_eq!(interval.elapsed, Duration::ZERO); + assert_eq!( + interval.finish_turn(start + Duration::from_millis(40)), + Duration::from_millis(40) + ); + interval.start_turn(start + Duration::from_millis(60)); + interval.leave(start + Duration::from_millis(70)); + assert_eq!( + interval.finish_turn(start + Duration::from_millis(80)), + Duration::from_millis(10) + ); + // A review that starts while idle contributes only after the next turn starts. + interval.enter(start + Duration::from_millis(100)); + interval.start_turn(start + Duration::from_millis(200)); + interval.leave(start + Duration::from_millis(210)); + assert_eq!( + interval.finish_turn(start + Duration::from_millis(220)), + Duration::from_millis(10) + ); + interval.start_turn(start + Duration::from_millis(230)); + interval.enter(start + Duration::from_millis(240)); + interval.leave(start + Duration::from_millis(250)); + interval.enter(start + Duration::from_millis(270)); + interval.leave(start + Duration::from_millis(280)); + assert_eq!( + interval.finish_turn(start + Duration::from_millis(300)), + Duration::from_millis(20) + ); + interval.start_turn(start + Duration::from_millis(310)); + assert_eq!( + interval.finish_turn(start + Duration::from_millis(320)), + Duration::ZERO + ); +} + +#[test] +fn late_turn_completion_preserves_both_samples_and_ignores_duplicate_stops() { + let start = Instant::now(); + let mut state = State::default(); + state.start_turn("a", start); + state.interval.enter(start + Duration::from_millis(10)); + state.start_turn("b", start + Duration::from_millis(30)); + assert_eq!( + state.finish_turn("a", start + Duration::from_millis(50)), + Some(Duration::from_millis(20)) + ); + assert_eq!( + state.finish_turn("a", start + Duration::from_millis(60)), + None + ); + assert_eq!( + state.finish_turn("b", start + Duration::from_millis(70)), + Some(Duration::from_millis(40)) + ); +} diff --git a/codex-rs/ext/guardian-reviewer/src/lib.rs b/codex-rs/ext/guardian-reviewer/src/lib.rs index 173401024a..94b078f738 100644 --- a/codex-rs/ext/guardian-reviewer/src/lib.rs +++ b/codex-rs/ext/guardian-reviewer/src/lib.rs @@ -3,6 +3,9 @@ //! The host supplies review attempts and enforces the resulting decision on the bound action. mod assessment; +mod blocking; +pub use blocking::ReviewBlocking; +pub use blocking::ReviewBlockingTurn; mod circuit_breaker; mod completion; mod conversation; diff --git a/codex-rs/ext/guardian-reviewer/src/routing.rs b/codex-rs/ext/guardian-reviewer/src/routing.rs index c33aa1ae7e..28d97774ba 100644 --- a/codex-rs/ext/guardian-reviewer/src/routing.rs +++ b/codex-rs/ext/guardian-reviewer/src/routing.rs @@ -71,6 +71,10 @@ impl ReviewRequest<'_, H> { .telemetry .start_timer("codex.guardian.decision.duration_ms", &[]) .ok(); + let _blocking = self + .thread_store + .get::() + .map(|blocking| blocking.enter()); let runtime = self.thread_store.get::(); let _task = runtime.as_ref().map(|runtime| runtime.tasks.token()); if runtime diff --git a/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs b/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs index 0aba3d653b..e6fb25fdca 100644 --- a/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs +++ b/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs @@ -22,6 +22,8 @@ use codex_extension_api::TurnAbortInput; use codex_extension_api::TurnLifecycleContributor; use codex_extension_api::TurnStartInput; use codex_extension_api::TurnStopInput; +use codex_guardian_reviewer::ReviewBlocking; +use codex_guardian_reviewer::ReviewBlockingTurn; use codex_guardian_reviewer::ReviewDenials; use codex_guardian_reviewer::ReviewerPool; use codex_guardian_reviewer::ReviewerTasks; @@ -58,6 +60,7 @@ impl ThreadLifecycleContributor for GuardianExtension { let manager = self.thread_manager.clone(); let parent_thread_id = input.thread_store.level_id().to_owned(); let runtime = input.thread_store.get_or_init(ReviewerTasks::default); + input.thread_store.get_or_init(ReviewBlocking::default); input.thread_store.get_or_init(|| { ReviewerPool::::new( Arc::clone(&runtime), @@ -228,19 +231,39 @@ impl ThreadLifecycleContributor for GuardianExtension { impl TurnLifecycleContributor for GuardianExtension { fn on_turn_start<'a>(&'a self, input: TurnStartInput<'a>) -> ExtensionFuture<'a, ()> { - Box::pin(ReviewDenials::clear_turn(input.thread_store, input.turn_id)) + Box::pin(async move { + if input.thread_store.get::().is_some() + && let Some(telemetry) = input.turn_store.get() + { + let tracker = input.thread_store.get_or_init(ReviewBlocking::default); + input + .turn_store + .get_or_init(|| tracker.start_turn(input.turn_id, telemetry)); + } + ReviewDenials::clear_turn(input.thread_store, input.turn_id).await; + }) } fn on_turn_stop<'a>(&'a self, input: TurnStopInput<'a>) -> ExtensionFuture<'a, ()> { - Box::pin(ReviewDenials::clear_turn( - input.thread_store, - input.turn_store.level_id(), - )) + Box::pin(async move { + if let (Some(tracker), Some(turn)) = ( + input.thread_store.get::(), + input.turn_store.get::(), + ) { + tracker.finish_turn(&turn, "completed"); + } + ReviewDenials::clear_turn(input.thread_store, input.turn_store.level_id()).await; + }) } fn on_turn_abort<'a>(&'a self, input: TurnAbortInput<'a>) -> ExtensionFuture<'a, ()> { - Box::pin(ReviewDenials::clear_turn( - input.thread_store, - input.turn_store.level_id(), - )) + Box::pin(async move { + if let (Some(tracker), Some(turn)) = ( + input.thread_store.get::(), + input.turn_store.get::(), + ) { + tracker.finish_turn(&turn, "aborted"); + } + ReviewDenials::clear_turn(input.thread_store, input.turn_store.level_id()).await; + }) } }