From 8ec80de0ef55d2fb393332d75c32df9751496a58 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Fri, 9 Oct 2026 16:44:00 +0000 Subject: [PATCH] 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 --- .../suite/v2/thread_goal_empty_responses.rs | 98 ++++++++++++++++--- codex-rs/ext/goal/src/accounting.rs | 48 ++++++++- codex-rs/ext/goal/src/extension.rs | 32 +++++- codex-rs/ext/goal/src/runtime.rs | 18 +++- codex-rs/ext/goal/tests/accounting.rs | 19 +++- 5 files changed, 188 insertions(+), 27 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/thread_goal_empty_responses.rs b/codex-rs/app-server/tests/suite/v2/thread_goal_empty_responses.rs index 0b291c9c84..76e4d600a1 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_goal_empty_responses.rs +++ b/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(()) } diff --git a/codex-rs/ext/goal/src/accounting.rs b/codex-rs/ext/goal/src/accounting.rs index 29b5610308..e338a563fa 100644 --- a/codex-rs/ext/goal/src/accounting.rs +++ b/codex-rs/ext/goal/src/accounting.rs @@ -20,12 +20,16 @@ use tokio::sync::SemaphorePermit; pub(crate) struct GoalAccountingState { inner: Mutex, 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, + interrupted_turn_id: Option, turns: HashMap, wall_clock: GoalWallClockAccounting, budget_limit_reported_goal_id: Option, @@ -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 { + 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 { 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, tokio::sync::AcquireError> { + self.turn_start_lock.acquire().await + } + pub(crate) fn current_active_goal_id_for_turn(&self, turn_id: &str) -> Option { 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, diff --git a/codex-rs/ext/goal/src/extension.rs b/codex-rs/ext/goal/src/extension.rs index 1ec9ee6378..bdc75350c3 100644 --- a/codex-rs/ext/goal/src/extension.rs +++ b/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::(); + 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); + } + } }) } diff --git a/codex-rs/ext/goal/src/runtime.rs b/codex-rs/ext/goal/src/runtime.rs index 3e35f98072..8744473a07 100644 --- a/codex-rs/ext/goal/src/runtime.rs +++ b/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 diff --git a/codex-rs/ext/goal/tests/accounting.rs b/codex-rs/ext/goal/tests/accounting.rs index f41d21f101..78867a526a 100644 --- a/codex-rs/ext/goal/tests/accounting.rs +++ b/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"),