From 62b458c93151595cdf2b5ef5e37aa3d8b5613aeb Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Sat, 29 Aug 2026 00:15:58 +0000 Subject: [PATCH] Block goals after repeated execution host failures (#41454) ## What changed - Track failed `exec` attempts whose handler ran for each active goal. - Mark the goal as blocked after three qualifying failure turns. - Reset the failure streak when any tool succeeds, and keep failures from carrying over to a replacement goal. ## Testing - Cover failure classification, streak resets, goal replacement, and end-to-end goal blocking through the app server. GitOrigin-RevId: 14dac7f4eb35507f12df3b2a82ab97bdafe7b897 --- .../tests/suite/v2/code_mode_host.rs | 74 ++++++++++++++ codex-rs/ext/goal/src/accounting.rs | 83 ++++++++++++++-- codex-rs/ext/goal/src/extension.rs | 30 +++++- codex-rs/ext/goal/src/runtime.rs | 40 ++++++-- codex-rs/ext/goal/tests/accounting.rs | 97 +++++++++++++++++++ .../ext/goal/tests/goal_extension_backend.rs | 79 ++++++++++++++- 6 files changed, 383 insertions(+), 20 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/code_mode_host.rs b/codex-rs/app-server/tests/suite/v2/code_mode_host.rs index d5e65817c1..a5ec397df7 100644 --- a/codex-rs/app-server/tests/suite/v2/code_mode_host.rs +++ b/codex-rs/app-server/tests/suite/v2/code_mode_host.rs @@ -7,6 +7,9 @@ use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; use codex_app_server_protocol::ThreadDecrementElicitationParams; use codex_app_server_protocol::ThreadDecrementElicitationResponse; +use codex_app_server_protocol::ThreadGoalSetResponse; +use codex_app_server_protocol::ThreadGoalStatus; +use codex_app_server_protocol::ThreadGoalUpdatedNotification; use codex_app_server_protocol::ThreadIncrementElicitationParams; use codex_app_server_protocol::ThreadIncrementElicitationResponse; use codex_app_server_protocol::ThreadStartParams; @@ -317,3 +320,74 @@ async fn app_server_prewarms_flag_selected_grpc_code_mode_host_before_first_turn assert!(status.success(), "app-server did not exit successfully"); Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn app_server_blocks_goal_after_repeated_code_mode_host_failures() -> Result<()> { + let model_server = responses::start_mock_server().await; + let mut model_responses = Vec::new(); + for turn in 1..=3 { + model_responses.push(responses::sse(vec![ + responses::ev_response_created(&format!("resp-{turn}-exec")), + responses::ev_custom_tool_call( + &format!("call-exec-{turn}"), + "exec", + "text('unreachable')", + ), + responses::ev_completed(&format!("resp-{turn}-exec")), + ])); + model_responses.push(responses::sse(vec![ + responses::ev_assistant_message( + &format!("msg-{turn}"), + "The execution host is unavailable.", + ), + responses::ev_completed(&format!("resp-{turn}-done")), + ])); + } + let response_mock = responses::mount_sse_sequence(&model_server, model_responses).await; + + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&model_server.uri()) + .enable_feature(Feature::CodeModeOnly) + .enable_feature(Feature::Goals) + .enable_feature(Feature::Sqlite) + .write(codex_home.path())?; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let host_url = format!("http://{}", listener.local_addr()?); + drop(listener); + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .with_args(&["--code-mode-host", &host_url]) + .build_initialized_with_timeout(DEFAULT_READ_TIMEOUT) + .await?; + let thread = app_server + .start_thread(ThreadStartParams::default()) + .await?; + let goal_request = app_server + .send_raw_request( + "thread/goal/set", + Some(json!({ + "threadId": thread.thread.id, + "objective": "finish the task", + "status": "active", + })), + ) + .await?; + let _: ThreadGoalSetResponse = + timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(goal_request)).await??; + + let goal = timeout(DEFAULT_READ_TIMEOUT, async { + loop { + let notification: ThreadGoalUpdatedNotification = + app_server.read_notification("thread/goal/updated").await?; + if notification.goal.status == ThreadGoalStatus::Blocked { + return Ok::<_, anyhow::Error>(notification.goal); + } + } + }) + .await??; + + assert_eq!(goal.status, ThreadGoalStatus::Blocked); + assert_eq!(response_mock.requests().len(), 6); + Ok(()) +} diff --git a/codex-rs/ext/goal/src/accounting.rs b/codex-rs/ext/goal/src/accounting.rs index 4012a9587b..6a4a29978d 100644 --- a/codex-rs/ext/goal/src/accounting.rs +++ b/codex-rs/ext/goal/src/accounting.rs @@ -1,3 +1,5 @@ +use codex_extension_api::ToolCallOutcome; +use codex_extension_api::ToolName; use codex_protocol::config_types::ModeKind; use codex_protocol::protocol::TokenUsage; use codex_state::ThreadGoalStatus; @@ -24,6 +26,8 @@ struct GoalAccountingInner { turns: HashMap, wall_clock: GoalWallClockAccounting, budget_limit_reported_goal_id: Option, + execution_failure_goal_id: Option, + consecutive_execution_failure_turns: u8, last_accounted_descendant_token_usage: i64, } @@ -33,6 +37,8 @@ struct GoalTurnAccounting { last_accounted_token_usage: TokenUsage, active_goal_id: Option, account_tokens: bool, + failed_execution: bool, + successful_tool: bool, } #[derive(Debug)] @@ -93,6 +99,58 @@ impl GoalAccountingState { self.inner().current_turn_id.clone() } + pub(crate) fn record_tool_outcome( + &self, + turn_id: &str, + tool_name: &ToolName, + outcome: ToolCallOutcome, + ) { + let mut inner = self.inner(); + let Some(turn) = inner.turns.get_mut(turn_id) else { + return; + }; + if turn.active_goal_id.is_none() { + return; + } + + match outcome { + ToolCallOutcome::Completed { success: true } => { + turn.successful_tool = true; + inner.execution_failure_goal_id = None; + inner.consecutive_execution_failure_turns = 0; + } + ToolCallOutcome::Failed { + handler_executed: true, + } if tool_name.is_default_namespace() && tool_name.name == "exec" => { + turn.failed_execution = true; + } + ToolCallOutcome::Completed { success: false } + | ToolCallOutcome::Failed { .. } + | ToolCallOutcome::Blocked + | ToolCallOutcome::Aborted => {} + } + } + + pub(crate) fn execution_failure_goal(&self, turn_id: &str) -> Option { + let mut inner = self.inner(); + let turn = inner.turns.get(turn_id)?; + let goal_id = turn.active_goal_id.clone()?; + if turn.successful_tool { + return None; + } + if !turn.failed_execution { + return None; + } + + if inner.execution_failure_goal_id.as_deref() != Some(goal_id.as_str()) { + inner.execution_failure_goal_id = Some(goal_id.clone()); + inner.consecutive_execution_failure_turns = 0; + } + inner.consecutive_execution_failure_turns = + inner.consecutive_execution_failure_turns.saturating_add(1); + (inner.consecutive_execution_failure_turns >= 3).then_some(goal_id) + } + /// Acquires the per-thread progress-accounting permit. /// /// Hold the returned permit from before taking a progress snapshot until after the persistent @@ -104,15 +162,16 @@ impl GoalAccountingState { self.progress_accounting_lock.acquire().await } - pub(crate) fn turn_is_current_active_goal(&self, turn_id: &str) -> bool { + 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) { - return false; + return None; } - let Some(turn) = inner.turns.get(turn_id) else { - return false; - }; - turn.account_tokens && turn.active_goal_id.is_some() + let turn = inner.turns.get(turn_id)?; + if !turn.account_tokens { + return None; + } + turn.active_goal_id.clone() } pub(crate) fn record_token_usage( @@ -176,6 +235,10 @@ impl GoalAccountingState { } let goal_changed = inner.wall_clock.active_goal_id.as_deref() != Some(goal_id.as_str()); let turn = inner.turns.get_mut(turn_id.as_str())?; + if turn.active_goal_id.as_deref() != Some(goal_id.as_str()) { + turn.failed_execution = false; + turn.successful_tool = false; + } turn.active_goal_id = Some(goal_id.clone()); if goal_changed { turn.reset_baseline_to_current(); @@ -207,6 +270,8 @@ impl GoalAccountingState { } inner.wall_clock.clear_active_goal(); inner.budget_limit_reported_goal_id = None; + inner.execution_failure_goal_id = None; + inner.consecutive_execution_failure_turns = 0; Some(turn_id) } @@ -219,6 +284,8 @@ impl GoalAccountingState { } inner.wall_clock.clear_active_goal(); inner.budget_limit_reported_goal_id = None; + inner.execution_failure_goal_id = None; + inner.consecutive_execution_failure_turns = 0; } pub(crate) fn progress_snapshot(&self, turn_id: &str) -> Option { @@ -385,6 +452,8 @@ impl Default for GoalAccountingInner { turns: HashMap::new(), wall_clock: GoalWallClockAccounting::new(), budget_limit_reported_goal_id: None, + execution_failure_goal_id: None, + consecutive_execution_failure_turns: 0, last_accounted_descendant_token_usage: 0, } } @@ -408,6 +477,8 @@ impl GoalTurnAccounting { current_token_usage, active_goal_id: None, account_tokens, + failed_execution: false, + successful_tool: false, } } diff --git a/codex-rs/ext/goal/src/extension.rs b/codex-rs/ext/goal/src/extension.rs index 349652da4f..0e8d9e7522 100644 --- a/codex-rs/ext/goal/src/extension.rs +++ b/codex-rs/ext/goal/src/extension.rs @@ -279,6 +279,20 @@ where } let turn_id = input.turn_store.level_id(); + if let Some(expected_goal_id) = + runtime.accounting_state().execution_failure_goal(turn_id) + && let Err(err) = runtime + .stop_active_goal_for_turn( + turn_id, + ActiveGoalStopReason::ExecutionUnavailable { expected_goal_id }, + ) + .await + { + tracing::warn!( + "failed to stop active goal after repeated execution failures for {turn_id}: {err}" + ); + return; + } if let Err(err) = runtime .account_active_goal_progress( turn_id, @@ -390,10 +404,18 @@ where let Some(runtime) = goal_runtime_handle(input.thread_store) else { return; }; - let should_count_for_goal_progress = runtime.is_enabled() - && tool_attempt_counts_for_goal_progress(input.outcome) - && !(input.tool_name.is_default_namespace() - && input.tool_name.name == UPDATE_GOAL_TOOL_NAME); + if !runtime.is_enabled() { + return; + } + runtime.accounting_state().record_tool_outcome( + input.turn_id, + input.tool_name, + input.outcome, + ); + let should_count_for_goal_progress = + tool_attempt_counts_for_goal_progress(input.outcome) + && !(input.tool_name.is_default_namespace() + && input.tool_name.name == UPDATE_GOAL_TOOL_NAME); if !should_count_for_goal_progress { return; } diff --git a/codex-rs/ext/goal/src/runtime.rs b/codex-rs/ext/goal/src/runtime.rs index d9f0f0df24..027378f837 100644 --- a/codex-rs/ext/goal/src/runtime.rs +++ b/codex-rs/ext/goal/src/runtime.rs @@ -39,6 +39,7 @@ pub(crate) struct GoalRuntimeConfig { pub(crate) enum ActiveGoalStopReason { TurnError, UsageLimit, + ExecutionUnavailable { expected_goal_id: String }, } struct GoalRuntimeInner { @@ -264,21 +265,33 @@ 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?; - if !self + let Some(accounting_goal_id) = self .inner .accounting_state - .turn_is_current_active_goal(turn_id) + .current_active_goal_id_for_turn(turn_id) + else { + return Ok(()); + }; + if let ActiveGoalStopReason::ExecutionUnavailable { expected_goal_id } = &reason + && accounting_goal_id != *expected_goal_id { return Ok(()); } - let (event_name, status) = match reason { + let (event_name, status, expected_goal_id) = match reason { ActiveGoalStopReason::TurnError => { - ("turn-error", codex_state::ThreadGoalStatus::Blocked) - } - ActiveGoalStopReason::UsageLimit => { - ("usage-limit", codex_state::ThreadGoalStatus::UsageLimited) + ("turn-error", codex_state::ThreadGoalStatus::Blocked, None) } + ActiveGoalStopReason::UsageLimit => ( + "usage-limit", + codex_state::ThreadGoalStatus::UsageLimited, + None, + ), + ActiveGoalStopReason::ExecutionUnavailable { expected_goal_id } => ( + "execution-unavailable", + codex_state::ThreadGoalStatus::Blocked, + Some(expected_goal_id), + ), }; self.account_active_goal_progress( turn_id, @@ -299,6 +312,12 @@ impl GoalRuntimeHandle { self.inner.accounting_state.clear_active_goal(); return Ok(()); }; + if expected_goal_id + .as_ref() + .is_some_and(|expected_goal_id| active_goal.goal_id != *expected_goal_id) + { + return Ok(()); + } let can_stop = active_goal.status == codex_state::ThreadGoalStatus::Active || (active_goal.status == codex_state::ThreadGoalStatus::BudgetLimited && status == codex_state::ThreadGoalStatus::UsageLimited); @@ -444,10 +463,13 @@ impl GoalRuntimeHandle { .is_some_and(|turn_id| { self.inner .accounting_state - .turn_is_current_active_goal(turn_id.as_str()) + .current_active_goal_id_for_turn(turn_id.as_str()) + .is_some() }); if !current_turn_is_goal_active { - self.inner.accounting_state.clear_active_goal(); + self.inner + .accounting_state + .reset_idle_progress_baseline_and_clear_active_goal(); } Ok(()) } diff --git a/codex-rs/ext/goal/tests/accounting.rs b/codex-rs/ext/goal/tests/accounting.rs index 9bd1186f08..556950c0a6 100644 --- a/codex-rs/ext/goal/tests/accounting.rs +++ b/codex-rs/ext/goal/tests/accounting.rs @@ -5,6 +5,8 @@ mod accounting; use accounting::BudgetLimitedGoalDisposition; use accounting::GoalAccountingState; +use codex_extension_api::ToolCallOutcome; +use codex_extension_api::ToolName; use codex_protocol::config_types::ModeKind; use codex_protocol::protocol::TokenUsage; use codex_state::ThreadGoalStatus; @@ -53,6 +55,101 @@ fn goal_accounting_ignores_plan_mode_turns() { assert_eq!(None, recorded); } +#[test] +fn execution_failures_do_not_transfer_to_a_replacement_goal() { + let state = GoalAccountingState::default(); + + for (turn, goal_id, replacement_goal_id) in [ + (1, "first-goal", Some("second-goal")), + (2, "second-goal", None), + (3, "second-goal", None), + (4, "second-goal", None), + ] { + let turn_id = format!("turn-{turn}"); + state.start_turn(&turn_id, ModeKind::Default, &TokenUsage::default()); + state.mark_turn_goal_active(&turn_id, goal_id); + state.record_tool_outcome( + &turn_id, + &ToolName::plain("exec"), + ToolCallOutcome::Failed { + handler_executed: true, + }, + ); + if let Some(replacement_goal_id) = replacement_goal_id { + state.mark_current_turn_goal_active(replacement_goal_id); + } + + assert_eq!( + (turn == 4).then(|| "second-goal".to_string()), + state.execution_failure_goal(&turn_id) + ); + state.finish_turn(&turn_id); + if turn == 3 { + state.reset_idle_progress_baseline_and_clear_active_goal(); + } + } +} + +#[test] +fn script_errors_and_failures_before_execution_do_not_block_goals() { + for outcome in [ + ToolCallOutcome::Completed { success: false }, + ToolCallOutcome::Failed { + handler_executed: false, + }, + ] { + let state = GoalAccountingState::default(); + for turn in 1..=3 { + let turn_id = format!("turn-{turn}"); + state.start_turn(&turn_id, ModeKind::Default, &TokenUsage::default()); + state.mark_turn_goal_active(&turn_id, "goal"); + state.record_tool_outcome(&turn_id, &ToolName::plain("exec"), outcome); + + assert_eq!(None, state.execution_failure_goal(&turn_id)); + state.finish_turn(&turn_id); + } + } +} + +#[test] +fn successful_tool_resets_failures_before_an_interrupted_turn_ends() { + let state = GoalAccountingState::default(); + + for (turn, tool_name, outcome) in [ + ( + 1, + "exec", + ToolCallOutcome::Failed { + handler_executed: true, + }, + ), + ( + 2, + "exec", + ToolCallOutcome::Failed { + handler_executed: true, + }, + ), + (3, "shell", ToolCallOutcome::Completed { success: true }), + ( + 4, + "exec", + ToolCallOutcome::Failed { + handler_executed: true, + }, + ), + ] { + let turn_id = format!("turn-{turn}"); + state.start_turn(&turn_id, ModeKind::Default, &TokenUsage::default()); + state.mark_turn_goal_active(&turn_id, "goal"); + state.record_tool_outcome(&turn_id, &ToolName::plain(tool_name), outcome); + if turn != 3 { + assert_eq!(None, state.execution_failure_goal(&turn_id)); + } + state.finish_turn(&turn_id); + } +} + #[test] fn goal_accounting_preserves_concurrent_descendant_usage_across_checkpoints() { let state = GoalAccountingState::default(); diff --git a/codex-rs/ext/goal/tests/goal_extension_backend.rs b/codex-rs/ext/goal/tests/goal_extension_backend.rs index e029f121fe..2612414262 100644 --- a/codex-rs/ext/goal/tests/goal_extension_backend.rs +++ b/codex-rs/ext/goal/tests/goal_extension_backend.rs @@ -835,6 +835,67 @@ async fn turn_error_blocks_goal() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +async fn failed_execution_turns_block_goal_unless_a_tool_succeeds() -> anyhow::Result<()> { + for (recovery_turn, status_only_turn, blocking_turn) in + [(None, None, 3), (Some(2), None, 5), (None, Some(2), 4)] + { + let runtime = test_runtime().await?; + let thread_id = test_thread_id()?; + seed_thread_metadata(runtime.as_ref(), thread_id).await?; + let harness = GoalExtensionHarness::new(runtime.clone(), thread_id).await?; + harness.start_turn("turn-1", &TokenUsage::default()).await; + + let tools = harness.tools(); + tool_by_name(&tools, "create_goal") + .handle(tool_call( + "create_goal", + "call-create-goal", + json!({ "objective": "ship goal extension backend" }), + )) + .await?; + + for turn in 1..=blocking_turn { + let turn_id = format!("turn-{turn}"); + if turn > 1 { + harness.start_turn(&turn_id, &TokenUsage::default()).await; + } + if status_only_turn != Some(turn) { + harness + .notify_tool_finish_with_outcome( + &turn_id, + &format!("call-exec-{turn}"), + "exec", + ToolCallOutcome::Failed { + handler_executed: true, + }, + ) + .await; + } + if recovery_turn == Some(turn) { + harness + .notify_tool_finish(&turn_id, "call-recovery", "shell") + .await; + } + harness.stop_turn(&turn_id).await; + + let goal = runtime + .thread_goals() + .get_thread_goal(thread_id) + .await? + .ok_or_else(|| anyhow::anyhow!("goal should exist"))?; + let expected = if turn == blocking_turn { + codex_state::ThreadGoalStatus::Blocked + } else { + codex_state::ThreadGoalStatus::Active + }; + assert_eq!(expected, goal.status); + } + } + + Ok(()) +} + #[tokio::test] async fn usage_limit_budget_limited_goal_accounts_remaining_progress() -> anyhow::Result<()> { let runtime = test_runtime().await?; @@ -1684,6 +1745,22 @@ impl GoalExtensionHarness { } async fn notify_tool_finish(&self, turn_id: &str, call_id: &str, tool_name: &str) { + self.notify_tool_finish_with_outcome( + turn_id, + call_id, + tool_name, + ToolCallOutcome::Completed { success: true }, + ) + .await; + } + + async fn notify_tool_finish_with_outcome( + &self, + turn_id: &str, + call_id: &str, + tool_name: &str, + outcome: ToolCallOutcome, + ) { let turn_store = ExtensionData::new(turn_id); let tool_name = codex_extension_api::ToolName::plain(tool_name); for contributor in self.registry.tool_lifecycle_contributors() { @@ -1696,7 +1773,7 @@ impl GoalExtensionHarness { call_id, tool_name: &tool_name, source: ToolCallSource::Direct, - outcome: ToolCallOutcome::Completed { success: true }, + outcome, }) .await; }