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

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(())
+}

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<String, GoalTurnAccounting>,
     wall_clock: GoalWallClockAccounting,
     budget_limit_reported_goal_id: Option<String>,
+    execution_failure_goal_id: Option<String>,
+    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<String>,
     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<String> {
+        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<String> {
         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 {
+        let turn = inner.turns.get(turn_id)?;
-            return false;
+        if !turn.account_tokens {
-        };
+            return None;
-        turn.account_tokens && turn.active_goal_id.is_some()
+        }
+        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<GoalProgressSnapshot> {
@@ -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,
         }
     }
 

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()
+            if !runtime.is_enabled() {
-                && tool_attempt_counts_for_goal_progress(input.outcome)
+                return;
-                && !(input.tool_name.is_default_namespace()
+            }
-                    && input.tool_name.name == UPDATE_GOAL_TOOL_NAME);
+            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;
             }

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)
+                ("turn-error", codex_state::ThreadGoalStatus::Blocked, None)
-            }
-            ActiveGoalStopReason::UsageLimit => {
-                ("usage-limit", codex_state::ThreadGoalStatus::UsageLimited)
             }
+            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(())
     }

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();

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;
         }