Count interrupted goal continuations toward the no-activity limit (#52640)

## Why

Interrupted automatic goal continuations reset the empty-response counter, allowing repeated interruptions without activity to keep a goal running indefinitely.

## What changed

Count interrupted automatic continuations without activity toward the existing three-turn limit that marks a goal as `blocked`. Defer evaluation until the thread is idle and serialize it with turn-start accounting. Activity and user turns still reset the counter.

## Testing

Extend accounting tests to cover interrupted and mixed empty continuations, including resets after tool activity or user turns. Add an app-server regression case where a `PreCompact` hook repeatedly interrupts automatic continuations and verify the `thread/goal/updated` notification reports `blocked`.

GitOrigin-RevId: 4c34e3ed1ae612e97cf067c2aa76836b10c50ef5

codex-rs/app-server/tests/suite/v2/thread_goal_empty_responses.rs

@@ -1,12 +1,18 @@
-//! Exercises the goal extension's empty-response breaker through the public API.
+//! Exercises the goal extension's no-activity breaker through the public API.
 
 use anyhow::Result;
 use app_test_support::MockResponsesConfig;
 use app_test_support::TestAppServer;
 use app_test_support::create_mock_responses_server_sequence;
-use codex_app_server_protocol::ThreadGoalGetResponse;
+use codex_app_server_protocol::ConfigBatchWriteParams;
+use codex_app_server_protocol::ConfigEdit;
+use codex_app_server_protocol::ConfigWriteResponse;
+use codex_app_server_protocol::HooksListParams;
+use codex_app_server_protocol::HooksListResponse;
+use codex_app_server_protocol::MergeStrategy;
 use codex_app_server_protocol::ThreadGoalSetResponse;
 use codex_app_server_protocol::ThreadGoalStatus;
+use codex_app_server_protocol::ThreadGoalUpdatedNotification;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::TurnCompletedNotification;
@@ -23,15 +29,31 @@ use tokio::time::timeout;
 #[test_case::test_case(Some("final_answer"); "draft")]
 #[test_case::test_case(Some("commentary"); "commentary")]
 #[test_case::test_case(Some("tool"); "tool")]
+#[test_case::test_case(Some("interrupt"); "interrupt")]
 #[tokio::test]
 async fn empty_goal_continuations_block_after_three_without_activity(
     recovery: Option<&str>,
 ) -> Result<()> {
-    let turns = if recovery.is_some() { 6 } else { 3 };
+    let turns = match recovery {
+        Some("interrupt") => 4,
+        Some(_) => 6,
+        None => 3,
+    };
     let mut scripts = Vec::new();
     for turn in 1..=turns {
         let mut id = format!("response-{turn}");
         let mut events = vec![responses::ev_response_created(&id)];
+        if recovery == Some("interrupt") {
+            events.push(responses::ev_assistant_message(
+                "progress",
+                "Ready for the next step",
+            ));
+            events.push(responses::ev_completed_with_tokens(
+                &id, /*total_tokens*/ 330_000,
+            ));
+            scripts.push(responses::sse(events));
+            break;
+        }
         if turn == 3
             && let Some(recovery) = recovery
         {
@@ -59,15 +81,54 @@ async fn empty_goal_continuations_block_after_three_without_activity(
     }
     let server = create_mock_responses_server_sequence(scripts).await;
     let codex_home = TempDir::new()?;
-    MockResponsesConfig::new(&server.uri())
+    let mut config = MockResponsesConfig::new(&server.uri())
         .with_model("gpt-5.4")
-        .enable_feature(Feature::Goals)
-        .write(codex_home.path())?;
+        .enable_feature(Feature::Goals);
+    if recovery == Some("interrupt") {
+        config = config
+            .enable_feature(Feature::CodexHooks)
+            .with_root_config(
+                "model_auto_compact_token_limit = 200000\nmodel_post_turn_compact_threshold_percent = 0",
+            )
+            .with_extra_config(
+                r#"
+[[hooks.PreCompact]]
+matcher = "auto"
+[[hooks.PreCompact.hooks]]
+type = "command"
+command = """printf '%s' '{"continue":false}'"""
+command_windows = """Write-Output '{"continue":false}'"""
+"#,
+            );
+    }
+    config.write(codex_home.path())?;
     let mut mcp = TestAppServer::builder()
         .with_codex_home(codex_home.path())
         .without_managed_config()
         .build_initialized()
         .await?;
+    if recovery == Some("interrupt") {
+        let list = mcp
+            .send_hooks_list_request(HooksListParams {
+                cwds: vec![codex_home.path().to_path_buf()],
+            })
+            .await?;
+        let HooksListResponse { data } = mcp.read_response(list).await?;
+        let hook = &data[0].hooks[0];
+        let trust = mcp
+            .send_config_batch_write_request(ConfigBatchWriteParams {
+                edits: vec![ConfigEdit {
+                    key_path: "hooks.state".to_string(),
+                    value: json!({hook.key.clone(): {"trusted_hash": hook.current_hash}}),
+                    merge_strategy: MergeStrategy::Upsert,
+                }],
+                file_path: None,
+                expected_version: None,
+                reload_user_config: true,
+            })
+            .await?;
+        let _: ConfigWriteResponse = mcp.read_response(trust).await?;
+    }
     let request = mcp
         .send_thread_start_request_with_auto_env(ThreadStartParams::default())
         .await?;
@@ -94,17 +155,24 @@ async fn empty_goal_continuations_block_after_three_without_activity(
             mcp.read_notification("turn/completed"),
         )
         .await??;
-        assert_eq!(TurnStatus::Completed, completed.turn.status);
+        let expected = if recovery == Some("interrupt") && turn > 1 {
+            TurnStatus::Interrupted
+        } else {
+            TurnStatus::Completed
+        };
+        assert_eq!(expected, completed.turn.status);
         assert_eq!(None, completed.turn.error);
     }
-    let request = mcp
-        .send_raw_request("thread/goal/get", Some(json!({"threadId": thread.id})))
-        .await?;
-    let result: ThreadGoalGetResponse = mcp.read_response(request).await?;
-    assert_eq!(
-        ThreadGoalStatus::Blocked,
-        result.goal.expect("goal exists").status
-    );
+    timeout(Duration::from_secs(30), async {
+        loop {
+            let notification: ThreadGoalUpdatedNotification =
+                mcp.read_notification("thread/goal/updated").await?;
+            if notification.goal.status == ThreadGoalStatus::Blocked {
+                return Ok::<_, anyhow::Error>(());
+            }
+        }
+    })
+    .await??;
     server.verify().await;
     Ok(())
 }

codex-rs/ext/goal/src/accounting.rs

@@ -20,12 +20,16 @@ use tokio::sync::SemaphorePermit;
 pub(crate) struct GoalAccountingState {
     inner: Mutex<GoalAccountingInner>,
     progress_accounting_lock: Semaphore,
+    // Serialize turn-start accounting with stopping a deferred interrupted turn.
+    // Never hold this lock while submitting a turn or awaiting the goal-state permit.
+    turn_start_lock: Semaphore,
     descendant_token_usage: AtomicI64,
 }
 
 #[derive(Debug)]
 struct GoalAccountingInner {
     current_turn_id: Option<String>,
+    interrupted_turn_id: Option<String>,
     turns: HashMap<String, GoalTurnAccounting>,
     wall_clock: GoalWallClockAccounting,
     budget_limit_reported_goal_id: Option<String>,
@@ -92,6 +96,10 @@ impl GoalAccountingState {
     ) {
         let turn_id = turn_id.into();
         let mut inner = self.inner();
+        if let Some(interrupted) = inner.interrupted_turn_id.take() {
+            inner.turns.remove(&interrupted);
+            inner.consecutive_empty_turns = 0;
+        }
         inner.current_turn_id = Some(turn_id.clone());
         inner.turns.insert(
             turn_id,
@@ -106,6 +114,10 @@ impl GoalAccountingState {
         self.inner().current_turn_id.clone()
     }
 
+    pub(crate) fn interrupted_turn_id(&self) -> Option<String> {
+        self.inner().interrupted_turn_id.clone()
+    }
+
     pub(crate) fn record_tool_outcome(
         &self,
         turn_id: &str,
@@ -216,10 +228,14 @@ impl GoalAccountingState {
     /// Evaluated under the goal-state permit after automatic admission records its turn ID.
     pub(crate) fn empty_response_goal(&self, turn_id: &str) -> Option<String> {
         let mut inner = self.inner();
+        let interrupted = inner.interrupted_turn_id.as_deref() == Some(turn_id);
+        if interrupted {
+            inner.interrupted_turn_id = None;
+        }
         let automatic = inner.automatic_goal_turn_id.as_deref() == Some(turn_id);
         let turn = inner.turns.get_mut(turn_id)?;
         let goal_id = turn.active_goal_id.clone()?;
-        let empty = automatic && turn.empty_final && !turn.has_activity;
+        let empty = automatic && !turn.has_activity && (turn.empty_final || interrupted);
         turn.empty_final = false;
         if !empty {
             inner.consecutive_empty_turns = 0;
@@ -240,9 +256,20 @@ impl GoalAccountingState {
         self.progress_accounting_lock.acquire().await
     }
 
+    pub(crate) async fn turn_start_permit(
+        &self,
+    ) -> Result<SemaphorePermit<'_>, tokio::sync::AcquireError> {
+        self.turn_start_lock.acquire().await
+    }
+
     pub(crate) fn current_active_goal_id_for_turn(&self, turn_id: &str) -> Option<String> {
         let inner = self.inner();
-        if inner.current_turn_id.as_deref() != Some(turn_id) {
+        if inner
+            .current_turn_id
+            .as_deref()
+            .or(inner.interrupted_turn_id.as_deref())
+            != Some(turn_id)
+        {
             return None;
         }
         let turn = inner.turns.get(turn_id)?;
@@ -448,12 +475,27 @@ impl GoalAccountingState {
         }
     }
 
+    pub(crate) fn interrupt_turn(&self, turn_id: &str) {
+        let mut inner = self.inner();
+        if inner.current_turn_id.as_deref() == Some(turn_id) {
+            // Abort can run on the submission loop: defer evaluation until idle so
+            // it never waits for the goal-state permit held by automatic admission.
+            inner.current_turn_id = None;
+            inner.interrupted_turn_id = Some(turn_id.to_string());
+        } else {
+            inner.turns.remove(turn_id);
+        }
+    }
+
     pub(crate) fn finish_turn(&self, turn_id: &str) {
         let mut inner = self.inner();
         inner.turns.remove(turn_id);
         if inner.current_turn_id.as_deref() == Some(turn_id) {
             inner.current_turn_id = None;
         }
+        if inner.interrupted_turn_id.as_deref() == Some(turn_id) {
+            inner.interrupted_turn_id = None;
+        }
     }
 
     pub(crate) fn mark_idle_progress_accounted_for_status(
@@ -500,6 +542,7 @@ impl Default for GoalAccountingState {
         Self {
             inner: Mutex::new(GoalAccountingInner::default()),
             progress_accounting_lock: Semaphore::new(/*permits*/ 1),
+            turn_start_lock: Semaphore::new(/*permits*/ 1),
             descendant_token_usage: AtomicI64::new(0),
         }
     }
@@ -535,6 +578,7 @@ impl Default for GoalAccountingInner {
     fn default() -> Self {
         Self {
             current_turn_id: None,
+            interrupted_turn_id: None,
             turns: HashMap::new(),
             wall_clock: GoalWallClockAccounting::new(),
             budget_limit_reported_goal_id: None,

codex-rs/ext/goal/src/extension.rs

@@ -183,6 +183,16 @@ where
                 return;
             };
 
+            let accounting = runtime.accounting_state();
+            if let Some(turn_id) = accounting.interrupted_turn_id() {
+                if let Err(err) = runtime
+                    .stop_active_goal_for_turn(&turn_id, ActiveGoalStopReason::EmptyResponse)
+                    .await
+                {
+                    tracing::warn!("failed to stop goal after interrupted empty turns: {err}");
+                }
+                accounting.finish_turn(&turn_id);
+            }
             if let Err(err) = runtime.continue_if_idle().await {
                 tracing::warn!(
                     "failed to continue active goal for idle thread {}: {err}",
@@ -239,6 +249,10 @@ where
                 return;
             };
 
+            let accounting = runtime.accounting_state();
+            let Ok(_turn_start_permit) = accounting.turn_start_permit().await else {
+                return;
+            };
             if let Err(err) = self
                 .state_dbs
                 .thread_goals()
@@ -248,7 +262,6 @@ where
                 tracing::warn!("failed to clear deferred goal continuation: {err}");
             }
 
-            let accounting = runtime.accounting_state();
             accounting.start_turn(
                 input.turn_id,
                 input.collaboration_mode.mode,
@@ -369,13 +382,16 @@ where
             let Some(runtime) = goal_runtime_handle(input.thread_store) else {
                 return;
             };
-            runtime.accounting_state().reset_empty_responses();
             if !runtime.is_enabled() {
+                runtime.accounting_state().reset_empty_responses();
                 return;
             }
 
             let turn_id = input.turn_store.level_id();
             input.thread_store.remove::<TurnStartOptions>();
+            if input.reason != codex_protocol::protocol::TurnAbortReason::Interrupted {
+                runtime.accounting_state().reset_empty_responses();
+            }
             if let Err(err) = runtime
                 .account_active_goal_progress(
                     turn_id,
@@ -390,7 +406,17 @@ where
                 );
                 return;
             }
-            runtime.accounting_state().finish_turn(turn_id);
+            let accounting = runtime.accounting_state();
+            match input.reason {
+                codex_protocol::protocol::TurnAbortReason::Interrupted => {
+                    accounting.interrupt_turn(turn_id);
+                }
+                codex_protocol::protocol::TurnAbortReason::Replaced
+                | codex_protocol::protocol::TurnAbortReason::ReviewEnded
+                | codex_protocol::protocol::TurnAbortReason::BudgetLimited => {
+                    accounting.finish_turn(turn_id);
+                }
+            }
         })
     }
 

codex-rs/ext/goal/src/runtime.rs

@@ -286,6 +286,19 @@ impl GoalRuntimeHandle {
         // Hold this through accounting and the status update so external goal
         // mutations and idle continuation cannot interleave between them.
         let _goal_state_permit = self.goal_state_permit().await?;
+        let accounting = &self.inner.accounting_state;
+        let _turn_start_permit = if matches!(reason, ActiveGoalStopReason::EmptyResponse)
+            && accounting.interrupted_turn_id().as_deref() == Some(turn_id)
+        {
+            Some(
+                accounting
+                    .turn_start_permit()
+                    .await
+                    .map_err(|err| err.to_string())?,
+            )
+        } else {
+            None
+        };
         let Some(accounting_goal_id) = self
             .inner
             .accounting_state
@@ -309,9 +322,7 @@ impl GoalRuntimeHandle {
                 None,
             ),
             ActiveGoalStopReason::EmptyResponse => {
-                let Some(expected_goal_id) =
-                    self.inner.accounting_state.empty_response_goal(turn_id)
-                else {
+                let Some(expected_goal_id) = accounting.empty_response_goal(turn_id) else {
                     return Ok(());
                 };
                 if accounting_goal_id != expected_goal_id {
@@ -508,6 +519,7 @@ impl GoalRuntimeHandle {
             .inner
             .accounting_state
             .current_turn_id()
+            .or_else(|| self.inner.accounting_state.interrupted_turn_id())
             .is_some_and(|turn_id| {
                 self.inner
                     .accounting_state

codex-rs/ext/goal/tests/accounting.rs

@@ -77,6 +77,10 @@ fn empty_continuations_require_three_turns_without_activity_or_goal_changes() {
         ("reset", 6),
         ("missing final", 6),
         ("partial answer", 6),
+        ("abort", 3),
+        ("mixed", 3),
+        ("tool after abort", 6),
+        ("user after abort", 6),
     ] {
         let state = GoalAccountingState::default();
         for turn in 1..=blocking_turn {
@@ -88,20 +92,27 @@ fn empty_continuations_require_three_turns_without_activity_or_goal_changes() {
             };
             state.start_turn(&id, ModeKind::Default, &TokenUsage::default());
             state.mark_turn_goal_active(&id, goal_id);
+            let aborted = matches!(
+                interruption,
+                "abort" | "tool after abort" | "user after abort"
+            ) || (interruption == "mixed" && turn == 2);
             if turn == 3 && interruption == "partial answer" {
                 let mut partial = empty_final.clone();
                 if let TurnItem::AgentMessage(message) = &mut partial {
                     message.phase = Some(MessagePhase::PartialAnswer);
                 }
                 state.record_item(&id, &partial);
-            } else if !(turn == 3 && interruption == "missing final") {
+            } else if !(aborted || turn == 3 && interruption == "missing final") {
                 state.record_item(&id, &empty_final);
             }
-            // Admission can finish after output arrives, but before turn-stop evaluation.
-            if !(turn == 3 && interruption == "user") {
+            if aborted {
+                state.interrupt_turn(&id);
+            }
+            // Admission can finish even after abort, but before idle evaluation.
+            if !(turn == 3 && matches!(interruption, "user" | "user after abort")) {
                 state.mark_goal_continuation(id.clone());
             }
-            if turn == 3 && interruption == "tool" {
+            if turn == 3 && matches!(interruption, "tool" | "tool after abort") {
                 state.record_tool_outcome(
                     &id,
                     &ToolName::plain("shell"),