feat(auto-review) short-circuit (#18890)

## Summary
Short circuit the convo if auto-review hits too many denials

## Testing
- [x] Added unit tests

---------

Co-authored-by: Codex <noreply@openai.com>

codex-rs/core/src/guardian/mod.rs

@@ -30,6 +30,8 @@ pub(crate) use review::guardian_rejection_message;
 pub(crate) use review::guardian_timeout_message;
 pub(crate) use review::is_guardian_reviewer_source;
 pub(crate) use review::new_guardian_review_id;
+#[cfg(test)]
+pub(crate) use review::record_guardian_denial_for_test;
 pub(crate) use review::review_approval_request;
 #[cfg(test)]
 pub(crate) use review::review_approval_request_with_cancel;
@@ -40,6 +42,8 @@ pub(crate) use review_session::GuardianReviewSessionManager;
 const GUARDIAN_PREFERRED_MODEL: &str = "codex-auto-review";
 pub(crate) const GUARDIAN_REVIEW_TIMEOUT: Duration = Duration::from_secs(90);
 pub(crate) const GUARDIAN_REVIEWER_NAME: &str = "guardian";
+pub(crate) const MAX_CONSECUTIVE_GUARDIAN_DENIALS_PER_TURN: u32 = 3;
+pub(crate) const MAX_TOTAL_GUARDIAN_DENIALS_PER_TURN: u32 = 10;
 const GUARDIAN_MAX_MESSAGE_TRANSCRIPT_TOKENS: usize = 10_000;
 const GUARDIAN_MAX_TOOL_TRANSCRIPT_TOKENS: usize = 10_000;
 const GUARDIAN_MAX_MESSAGE_ENTRY_TOKENS: usize = 2_000;
@@ -63,6 +67,56 @@ pub(crate) struct GuardianRejection {
     pub(crate) source: GuardianAssessmentDecisionSource,
 }
 
+#[derive(Debug, Default)]
+pub(crate) struct GuardianRejectionCircuitBreaker {
+    turns: std::collections::HashMap<String, GuardianRejectionCircuitBreakerTurn>,
+}
+
+#[derive(Debug, Default)]
+struct GuardianRejectionCircuitBreakerTurn {
+    consecutive_denials: u32,
+    total_denials: u32,
+    interrupt_triggered: bool,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum GuardianRejectionCircuitBreakerAction {
+    Continue,
+    InterruptTurn {
+        consecutive_denials: u32,
+        total_denials: u32,
+    },
+}
+
+impl GuardianRejectionCircuitBreaker {
+    pub(crate) fn clear_turn(&mut self, turn_id: &str) {
+        self.turns.remove(turn_id);
+    }
+
+    pub(crate) fn record_denial(&mut self, turn_id: &str) -> GuardianRejectionCircuitBreakerAction {
+        let turn = self.turns.entry(turn_id.to_string()).or_default();
+        turn.consecutive_denials = turn.consecutive_denials.saturating_add(1);
+        turn.total_denials = turn.total_denials.saturating_add(1);
+        if !turn.interrupt_triggered
+            && (turn.consecutive_denials >= MAX_CONSECUTIVE_GUARDIAN_DENIALS_PER_TURN
+                || turn.total_denials >= MAX_TOTAL_GUARDIAN_DENIALS_PER_TURN)
+        {
+            turn.interrupt_triggered = true;
+            GuardianRejectionCircuitBreakerAction::InterruptTurn {
+                consecutive_denials: turn.consecutive_denials,
+                total_denials: turn.total_denials,
+            }
+        } else {
+            GuardianRejectionCircuitBreakerAction::Continue
+        }
+    }
+
+    pub(crate) fn record_non_denial(&mut self, turn_id: &str) {
+        let turn = self.turns.entry(turn_id.to_string()).or_default();
+        turn.consecutive_denials = 0;
+    }
+}
+
 #[cfg(test)]
 use approval_request::format_guardian_action_pretty;
 #[cfg(test)]

codex-rs/core/src/guardian/review.rs

@@ -17,6 +17,7 @@ use codex_protocol::protocol::GuardianRiskLevel;
 use codex_protocol::protocol::GuardianUserAuthorization;
 use codex_protocol::protocol::ReviewDecision;
 use codex_protocol::protocol::SubAgentSource;
+use codex_protocol::protocol::TurnAbortReason;
 use codex_protocol::protocol::WarningEvent;
 use tokio::sync::oneshot;
 use tokio_util::sync::CancellationToken;
@@ -30,6 +31,7 @@ use super::GuardianApprovalRequest;
 use super::GuardianAssessment;
 use super::GuardianAssessmentOutcome;
 use super::GuardianRejection;
+use super::GuardianRejectionCircuitBreakerAction;
 use super::approval_request::guardian_assessment_action;
 use super::approval_request::guardian_request_target_item_id;
 use super::approval_request::guardian_request_turn_id;
@@ -172,6 +174,64 @@ fn track_guardian_review(
         .track_guardian_review(tracking, result);
 }
 
+async fn record_guardian_non_denial(session: &Arc<Session>, turn_id: &str) {
+    session
+        .services
+        .guardian_rejection_circuit_breaker
+        .lock()
+        .await
+        .record_non_denial(turn_id);
+}
+
+async fn record_guardian_denial(session: &Arc<Session>, turn: &Arc<TurnContext>, turn_id: &str) {
+    let action = session
+        .services
+        .guardian_rejection_circuit_breaker
+        .lock()
+        .await
+        .record_denial(turn_id);
+    let GuardianRejectionCircuitBreakerAction::InterruptTurn {
+        consecutive_denials,
+        total_denials,
+    } = action
+    else {
+        return;
+    };
+
+    if session.turn_context_for_sub_id(turn_id).await.is_none() {
+        return;
+    }
+
+    session
+        .send_event(
+            turn.as_ref(),
+            EventMsg::GuardianWarning(WarningEvent {
+                message: format!(
+                    "Automatic approval review rejected too many approval requests for this turn ({consecutive_denials} consecutive, {total_denials} total); interrupting the turn."
+                ),
+            }),
+        )
+        .await;
+
+    let runtime_handle = session.services.runtime_handle.clone();
+    let session = Arc::clone(session);
+    let turn_id = turn_id.to_string();
+    let _abort_task = runtime_handle.spawn(async move {
+        session
+            .abort_turn_if_active(&turn_id, TurnAbortReason::Interrupted)
+            .await;
+    });
+}
+
+#[cfg(test)]
+pub(crate) async fn record_guardian_denial_for_test(
+    session: &Arc<Session>,
+    turn: &Arc<TurnContext>,
+    turn_id: &str,
+) {
+    record_guardian_denial(session, turn, turn_id).await;
+}
+
 /// This function always fails closed: timeouts, review-session failures, and
 /// parse failures all block execution, but timeouts are still surfaced to the
 /// caller as distinct from explicit guardian denials.
@@ -234,7 +294,7 @@ async fn run_guardian_review(
                 EventMsg::GuardianAssessment(GuardianAssessmentEvent {
                     id: review_id,
                     target_item_id,
-                    turn_id: assessment_turn_id,
+                    turn_id: assessment_turn_id.clone(),
                     status: GuardianAssessmentStatus::Aborted,
                     risk_level: None,
                     user_authorization: None,
@@ -244,6 +304,7 @@ async fn run_guardian_review(
                 }),
             )
             .await;
+        record_guardian_non_denial(&session, &assessment_turn_id).await;
         return ReviewDecision::Abort;
     }
 
@@ -259,7 +320,7 @@ async fn run_guardian_review(
     ))
     .await;
 
-    let assessment = match outcome {
+    let (assessment, count_denial_for_circuit_breaker) = match outcome {
         GuardianReviewOutcome::Completed(assessment) => {
             let approved = matches!(assessment.outcome, GuardianAssessmentOutcome::Allow);
             track_guardian_review(
@@ -284,7 +345,9 @@ async fn run_guardian_review(
                     ..analytics_result
                 },
             );
-            assessment
+            let count_denial_for_circuit_breaker =
+                matches!(assessment.outcome, GuardianAssessmentOutcome::Deny);
+            (assessment, count_denial_for_circuit_breaker)
         }
         GuardianReviewOutcome::Error(error) => match error {
             GuardianReviewError::Timeout => {
@@ -305,7 +368,7 @@ async fn run_guardian_review(
                 session
                     .send_event(
                         turn.as_ref(),
-                        EventMsg::Warning(WarningEvent {
+                        EventMsg::GuardianWarning(WarningEvent {
                             message: rationale.clone(),
                         }),
                     )
@@ -316,7 +379,7 @@ async fn run_guardian_review(
                         EventMsg::GuardianAssessment(GuardianAssessmentEvent {
                             id: review_id,
                             target_item_id,
-                            turn_id: assessment_turn_id,
+                            turn_id: assessment_turn_id.clone(),
                             status: GuardianAssessmentStatus::TimedOut,
                             risk_level: None,
                             user_authorization: None,
@@ -326,6 +389,7 @@ async fn run_guardian_review(
                         }),
                     )
                     .await;
+                record_guardian_non_denial(&session, &assessment_turn_id).await;
                 return ReviewDecision::TimedOut;
             }
             GuardianReviewError::Cancelled => {
@@ -346,7 +410,7 @@ async fn run_guardian_review(
                         EventMsg::GuardianAssessment(GuardianAssessmentEvent {
                             id: review_id,
                             target_item_id,
-                            turn_id: assessment_turn_id,
+                            turn_id: assessment_turn_id.clone(),
                             status: GuardianAssessmentStatus::Aborted,
                             risk_level: None,
                             user_authorization: None,
@@ -356,6 +420,7 @@ async fn run_guardian_review(
                         }),
                     )
                     .await;
+                record_guardian_non_denial(&session, &assessment_turn_id).await;
                 return ReviewDecision::Abort;
             }
             GuardianReviewError::PromptBuild { .. }
@@ -381,12 +446,15 @@ async fn run_guardian_review(
                         ..analytics_result
                     },
                 );
-                GuardianAssessment {
-                    risk_level: GuardianRiskLevel::High,
-                    user_authorization: GuardianUserAuthorization::Unknown,
-                    outcome: GuardianAssessmentOutcome::Deny,
-                    rationale,
-                }
+                (
+                    GuardianAssessment {
+                        risk_level: GuardianRiskLevel::High,
+                        user_authorization: GuardianUserAuthorization::Unknown,
+                        outcome: GuardianAssessmentOutcome::Deny,
+                        rationale,
+                    },
+                    false,
+                )
             }
         },
     };
@@ -410,7 +478,7 @@ async fn run_guardian_review(
     session
         .send_event(
             turn.as_ref(),
-            EventMsg::Warning(WarningEvent { message: warning }),
+            EventMsg::GuardianWarning(WarningEvent { message: warning }),
         )
         .await;
     let status = if approved {
@@ -436,7 +504,7 @@ async fn run_guardian_review(
             EventMsg::GuardianAssessment(GuardianAssessmentEvent {
                 id: review_id,
                 target_item_id,
-                turn_id: assessment_turn_id,
+                turn_id: assessment_turn_id.clone(),
                 status,
                 risk_level: Some(assessment.risk_level),
                 user_authorization: Some(assessment.user_authorization),
@@ -447,6 +515,12 @@ async fn run_guardian_review(
         )
         .await;
 
+    if count_denial_for_circuit_breaker {
+        record_guardian_denial(&session, &turn, &assessment_turn_id).await;
+    } else {
+        record_guardian_non_denial(&session, &assessment_turn_id).await;
+    }
+
     if approved {
         ReviewDecision::Approved
     } else {

codex-rs/core/src/guardian/tests.rs

@@ -67,6 +67,74 @@ fn fixed_guardian_parent_session_id() -> ThreadId {
         .expect("fixed parent session id should be a valid UUID")
 }
 
+#[test]
+fn guardian_rejection_circuit_breaker_interrupts_after_three_consecutive_denials() {
+    let mut circuit_breaker = GuardianRejectionCircuitBreaker::default();
+    assert_eq!(
+        circuit_breaker.record_denial("turn-1"),
+        GuardianRejectionCircuitBreakerAction::Continue
+    );
+    assert_eq!(
+        circuit_breaker.record_denial("turn-1"),
+        GuardianRejectionCircuitBreakerAction::Continue
+    );
+    assert_eq!(
+        circuit_breaker.record_denial("turn-1"),
+        GuardianRejectionCircuitBreakerAction::InterruptTurn {
+            consecutive_denials: 3,
+            total_denials: 3,
+        }
+    );
+    assert_eq!(
+        circuit_breaker.record_denial("turn-1"),
+        GuardianRejectionCircuitBreakerAction::Continue
+    );
+}
+
+#[test]
+fn guardian_rejection_circuit_breaker_resets_consecutive_denials_on_non_denial() {
+    let mut circuit_breaker = GuardianRejectionCircuitBreaker::default();
+    assert_eq!(
+        circuit_breaker.record_denial("turn-1"),
+        GuardianRejectionCircuitBreakerAction::Continue
+    );
+    circuit_breaker.record_non_denial("turn-1");
+    assert_eq!(
+        circuit_breaker.record_denial("turn-1"),
+        GuardianRejectionCircuitBreakerAction::Continue
+    );
+    assert_eq!(
+        circuit_breaker.record_denial("turn-1"),
+        GuardianRejectionCircuitBreakerAction::Continue
+    );
+    assert_eq!(
+        circuit_breaker.record_denial("turn-1"),
+        GuardianRejectionCircuitBreakerAction::InterruptTurn {
+            consecutive_denials: 3,
+            total_denials: 4,
+        }
+    );
+}
+
+#[test]
+fn guardian_rejection_circuit_breaker_interrupts_after_ten_total_denials() {
+    let mut circuit_breaker = GuardianRejectionCircuitBreaker::default();
+    for _ in 0..9 {
+        assert_eq!(
+            circuit_breaker.record_denial("turn-1"),
+            GuardianRejectionCircuitBreakerAction::Continue
+        );
+        circuit_breaker.record_non_denial("turn-1");
+    }
+    assert_eq!(
+        circuit_breaker.record_denial("turn-1"),
+        GuardianRejectionCircuitBreakerAction::InterruptTurn {
+            consecutive_denials: 1,
+            total_denials: 10,
+        }
+    );
+}
+
 async fn guardian_test_session_and_turn(
     server: &wiremock::MockServer,
 ) -> (Arc<Session>, Arc<TurnContext>) {
@@ -739,7 +807,7 @@ async fn cancelled_guardian_review_emits_terminal_abort_without_warning() {
     while let Ok(event) = rx.try_recv() {
         match event.msg {
             EventMsg::GuardianAssessment(event) => guardian_statuses.push(event.status),
-            EventMsg::Warning(event) => warnings.push(event.message),
+            EventMsg::GuardianWarning(event) => warnings.push(event.message),
             _ => {}
         }
     }
@@ -1491,7 +1559,7 @@ async fn guardian_review_surfaces_responses_api_errors_in_rejection_reason() ->
     let mut denial_rationales = Vec::new();
     while let Ok(event) = rx.try_recv() {
         match event.msg {
-            EventMsg::Warning(event) => warnings.push(event.message),
+            EventMsg::GuardianWarning(event) => warnings.push(event.message),
             EventMsg::GuardianAssessment(event)
                 if event.status == GuardianAssessmentStatus::Denied =>
             {

codex-rs/core/src/session/session.rs

@@ -695,6 +695,8 @@ impl Session {
             models_manager: Arc::clone(&models_manager),
             tool_approvals: Mutex::new(ApprovalStore::default()),
             guardian_rejections: Mutex::new(HashMap::new()),
+            guardian_rejection_circuit_breaker: Mutex::new(Default::default()),
+            runtime_handle: tokio::runtime::Handle::current(),
             skills_manager,
             plugins_manager: Arc::clone(&plugins_manager),
             mcp_manager: Arc::clone(&mcp_manager),

codex-rs/core/src/session/tests.rs

@@ -3093,6 +3093,8 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
         models_manager: Arc::clone(&models_manager),
         tool_approvals: Mutex::new(ApprovalStore::default()),
         guardian_rejections: Mutex::new(std::collections::HashMap::new()),
+        guardian_rejection_circuit_breaker: Mutex::new(Default::default()),
+        runtime_handle: tokio::runtime::Handle::current(),
         skills_manager,
         plugins_manager,
         mcp_manager,
@@ -4313,6 +4315,8 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx(
         models_manager: Arc::clone(&models_manager),
         tool_approvals: Mutex::new(ApprovalStore::default()),
         guardian_rejections: Mutex::new(std::collections::HashMap::new()),
+        guardian_rejection_circuit_breaker: Mutex::new(Default::default()),
+        runtime_handle: tokio::runtime::Handle::current(),
         skills_manager,
         plugins_manager,
         mcp_manager,
@@ -5545,6 +5549,125 @@ impl SessionTask for NeverEndingTask {
     }
 }
 
+#[derive(Clone, Copy)]
+struct GuardianDeniedApprovalTask;
+
+impl SessionTask for GuardianDeniedApprovalTask {
+    fn kind(&self) -> TaskKind {
+        TaskKind::Regular
+    }
+
+    fn span_name(&self) -> &'static str {
+        "session_task.guardian_denied_approval"
+    }
+
+    async fn run(
+        self: Arc<Self>,
+        session: Arc<SessionTaskContext>,
+        ctx: Arc<TurnContext>,
+        _input: Vec<UserInput>,
+        cancellation_token: CancellationToken,
+    ) -> Option<String> {
+        let session = session.clone_session();
+        for _ in 0..3 {
+            crate::guardian::record_guardian_denial_for_test(&session, &ctx, &ctx.sub_id).await;
+        }
+
+        cancellation_token.cancelled().await;
+        None
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn guardian_auto_review_interrupts_after_three_consecutive_denials() {
+    let (sess, tc, rx) = make_session_and_context_with_rx().await;
+    let input = vec![UserInput::Text {
+        text: "trigger guardian denials".to_string(),
+        text_elements: Vec::new(),
+    }];
+    sess.spawn_task(Arc::clone(&tc), input, GuardianDeniedApprovalTask)
+        .await;
+
+    let mut observed = Vec::new();
+    let aborted = tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        loop {
+            let event = rx.recv().await.expect("event");
+            if let EventMsg::TurnAborted(event) = &event.msg {
+                let event = event.clone();
+                observed.push(EventMsg::TurnAborted(event.clone()));
+                break event;
+            }
+            observed.push(event.msg);
+        }
+    })
+    .await
+    .unwrap_or_else(|_| {
+        panic!(
+            "guardian denial circuit breaker should interrupt the turn; observed events: {observed:?}"
+        )
+    });
+    assert_eq!(aborted.reason, TurnAbortReason::Interrupted);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn guardian_helper_review_interrupts_after_three_consecutive_denials() {
+    let (sess, tc, rx) = make_session_and_context_with_rx().await;
+    let input = vec![UserInput::Text {
+        text: "keep turn active for helper reviews".to_string(),
+        text_elements: Vec::new(),
+    }];
+    sess.spawn_task(
+        Arc::clone(&tc),
+        input,
+        NeverEndingTask {
+            kind: TaskKind::Regular,
+            listen_to_cancellation_token: true,
+        },
+    )
+    .await;
+
+    let session_for_review = Arc::clone(&sess);
+    let turn_for_review = Arc::clone(&tc);
+    let turn_id = tc.sub_id.clone();
+    let review_thread = std::thread::spawn(move || {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("helper review runtime");
+        runtime.block_on(async move {
+            for _ in 0..3 {
+                crate::guardian::record_guardian_denial_for_test(
+                    &session_for_review,
+                    &turn_for_review,
+                    &turn_id,
+                )
+                .await;
+            }
+        });
+    });
+    review_thread.join().expect("helper review thread");
+
+    let mut observed = Vec::new();
+    let aborted = timeout(StdDuration::from_secs(5), async {
+        loop {
+            let event = rx.recv().await.expect("event");
+            if let EventMsg::TurnAborted(event) = &event.msg {
+                let event = event.clone();
+                observed.push(EventMsg::TurnAborted(event.clone()));
+                break event;
+            }
+            observed.push(event.msg);
+        }
+    })
+    .await
+    .unwrap_or_else(|_| {
+        panic!(
+            "helper review circuit breaker should interrupt the turn; observed events: {observed:?}"
+        )
+    });
+    assert_eq!(aborted.reason, TurnAbortReason::Interrupted);
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 #[test_log::test]
 async fn abort_regular_task_emits_turn_aborted_only() {

codex-rs/core/src/session/turn.rs

@@ -1484,6 +1484,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<String> {
         },
         EventMsg::Error(_)
         | EventMsg::Warning(_)
+        | EventMsg::GuardianWarning(_)
         | EventMsg::RealtimeConversationStarted(_)
         | EventMsg::RealtimeConversationSdp(_)
         | EventMsg::RealtimeConversationRealtime(_)

codex-rs/core/src/state/service.rs

@@ -8,6 +8,7 @@ use crate::client::ModelClient;
 use crate::config::StartedNetworkProxy;
 use crate::exec_policy::ExecPolicyManager;
 use crate::guardian::GuardianRejection;
+use crate::guardian::GuardianRejectionCircuitBreaker;
 use crate::mcp::McpManager;
 use crate::plugins::PluginsManager;
 use crate::skills_watcher::SkillsWatcher;
@@ -26,6 +27,7 @@ use codex_rollout::state_db::StateDbHandle;
 use codex_rollout_trace::RolloutTraceRecorder;
 use codex_thread_store::LocalThreadStore;
 use std::path::PathBuf;
+use tokio::runtime::Handle;
 use tokio::sync::Mutex;
 use tokio::sync::RwLock;
 use tokio::sync::watch;
@@ -52,6 +54,8 @@ pub(crate) struct SessionServices {
     pub(crate) session_telemetry: SessionTelemetry,
     pub(crate) tool_approvals: Mutex<ApprovalStore>,
     pub(crate) guardian_rejections: Mutex<HashMap<String, GuardianRejection>>,
+    pub(crate) guardian_rejection_circuit_breaker: Mutex<GuardianRejectionCircuitBreaker>,
+    pub(crate) runtime_handle: Handle,
     pub(crate) skills_manager: Arc<SkillsManager>,
     pub(crate) plugins_manager: Arc<PluginsManager>,
     pub(crate) mcp_manager: Arc<McpManager>,

codex-rs/core/src/tasks/mod.rs

@@ -269,6 +269,12 @@ impl Session {
         let cancellation_token = CancellationToken::new();
         let done = Arc::new(Notify::new());
 
+        self.services
+            .guardian_rejection_circuit_breaker
+            .lock()
+            .await
+            .clear_turn(&turn_context.sub_id);
+
         let queued_response_items = self.take_queued_response_items_for_next_turn().await;
         let mailbox_items = self.get_pending_input().await;
         let turn_state = {
@@ -410,6 +416,40 @@ impl Session {
         }
     }
 
+    pub(crate) async fn abort_turn_if_active(
+        self: &Arc<Self>,
+        turn_id: &str,
+        reason: TurnAbortReason,
+    ) -> bool {
+        let active_turn = {
+            let mut active = self.active_turn.lock().await;
+            if active
+                .as_ref()
+                .is_some_and(|active_turn| active_turn.tasks.contains_key(turn_id))
+            {
+                active.take()
+            } else {
+                None
+            }
+        };
+        let Some(mut active_turn) = active_turn else {
+            return false;
+        };
+
+        for task in active_turn.drain_tasks() {
+            self.handle_task_abort(task, reason.clone()).await;
+        }
+        // Let interrupted tasks observe cancellation before dropping pending approvals, or an
+        // in-flight approval wait can surface as a model-visible rejection before TurnAborted.
+        active_turn.clear_pending().await;
+
+        if reason == TurnAbortReason::Interrupted {
+            self.maybe_start_turn_for_pending_work().await;
+        }
+
+        true
+    }
+
     pub async fn on_task_finished(
         self: &Arc<Self>,
         turn_context: Arc<TurnContext>,
@@ -568,6 +608,11 @@ impl Session {
             time_to_first_token_ms,
         });
         self.send_event(turn_context.as_ref(), event).await;
+        self.services
+            .guardian_rejection_circuit_breaker
+            .lock()
+            .await
+            .clear_turn(&turn_context.sub_id);
 
         if should_clear_active_turn {
             let session = Arc::clone(self);
@@ -654,6 +699,11 @@ impl Session {
             duration_ms,
         });
         self.send_event(task.turn_context.as_ref(), event).await;
+        self.services
+            .guardian_rejection_circuit_breaker
+            .lock()
+            .await
+            .clear_turn(&task.turn_context.sub_id);
     }
 }