From 60d78332fc6adbf7f4fd000ede9eabe8d52f4e86 Mon Sep 17 00:00:00 2001 From: ioanbudea-OAI Date: Tue, 20 Oct 2026 14:34:45 +0000 Subject: [PATCH] Measure Guardian approval blocking time per turn (#55591) ## What changed Record `codex.guardian.turn.blocking_ms` when a turn completes or aborts, with an `outcome` label. Count overlapping approval waits once, exclude idle gaps, and split pending reviews across the active turns they span. Use each turn's telemetry so samples retain the servicing turn's model attribution. ## Testing Add unit coverage for overlapping waits, idle gaps, late turn completion, and duplicate stops. Extend integration coverage to verify attribution across turns and model changes for yielded command reviews and MCP elicitation, including aborted turns. GitOrigin-RevId: a3e6a0c32a06fe58ea93119ecd941076100fb82c --- codex-rs/core/src/tasks/lifecycle.rs | 3 + .../tests/suite/guardian_mcp_elicitation.rs | 66 +++++++- codex-rs/core/tests/suite/guardian_review.rs | 154 +++++++++++++++++- .../ext/guardian-reviewer/src/blocking.rs | 149 +++++++++++++++++ .../guardian-reviewer/src/blocking_tests.rs | 68 ++++++++ codex-rs/ext/guardian-reviewer/src/lib.rs | 3 + codex-rs/ext/guardian-reviewer/src/routing.rs | 4 + .../ext/guardian-v2/src/sync_reviewer/mod.rs | 41 ++++- 8 files changed, 473 insertions(+), 15 deletions(-) create mode 100644 codex-rs/ext/guardian-reviewer/src/blocking.rs create mode 100644 codex-rs/ext/guardian-reviewer/src/blocking_tests.rs 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; + }) } }