From 0735c519789d300097554425cc3d7cf3f2d718a1 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Wed, 9 Sep 2026 22:11:27 +0000 Subject: [PATCH] Block goals after three empty automatic continuation turns (#44320) ## Why Automatic goal continuations can repeatedly return empty final answers without making progress. Stop this loop by marking the goal as `blocked` after three consecutive empty turns with no other activity. ## What changed - Observe completed turn items through a new `on_item_completed` lifecycle hook. - Track empty final answers only for automatically admitted goal turns, resetting the streak on activity, user turns, or goal changes. - Preserve normal turn completion and streamed message deltas when blocking a goal. ## Testing Add accounting coverage for the three-turn threshold and streak resets, plus app-server tests for empty continuations and recovery through final-answer text, commentary, or tool activity. GitOrigin-RevId: 4b9d2cb2e306b0adc316972429cc35000115b88b --- codex-rs/app-server/tests/suite/v2/mod.rs | 1 + .../suite/v2/thread_goal_empty_responses.rs | 110 ++++++++++++++++++ codex-rs/core/src/session/mod.rs | 9 ++ .../ext/extension-api/src/contributors.rs | 10 ++ codex-rs/ext/goal/src/accounting.rs | 90 ++++++++++++++ codex-rs/ext/goal/src/extension.rs | 27 +++++ codex-rs/ext/goal/src/runtime.rs | 27 ++++- codex-rs/ext/goal/tests/accounting.rs | 59 ++++++++++ 8 files changed, 331 insertions(+), 2 deletions(-) create mode 100644 codex-rs/app-server/tests/suite/v2/thread_goal_empty_responses.rs diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 740f82d0f5..4846897d71 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -107,6 +107,7 @@ mod thread_environments; mod thread_fork; #[path = "thread_fork_multi_agent_tests.rs"] mod thread_fork_multi_agent; +mod thread_goal_empty_responses; mod thread_inject_items; mod thread_list; mod thread_loaded_list; 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 new file mode 100644 index 0000000000..0b291c9c84 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_goal_empty_responses.rs @@ -0,0 +1,110 @@ +//! Exercises the goal extension's empty-response 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::ThreadGoalSetResponse; +use codex_app_server_protocol::ThreadGoalStatus; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnCompletedNotification; +use codex_app_server_protocol::TurnStatus; +use codex_features::Feature; +use core_test_support::responses; +use pretty_assertions::assert_eq; +use serde_json::json; +use tempfile::TempDir; +use tokio::time::Duration; +use tokio::time::timeout; + +#[test_case::test_case(None; "empty")] +#[test_case::test_case(Some("final_answer"); "draft")] +#[test_case::test_case(Some("commentary"); "commentary")] +#[test_case::test_case(Some("tool"); "tool")] +#[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 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 turn == 3 + && let Some(recovery) = recovery + { + if recovery == "tool" { + events.push(responses::ev_function_call("get-goal", "get_goal", "{}")); + events.push(responses::ev_completed(&id)); + scripts.push(responses::sse(events)); + id.push_str("-after-tool"); + events = vec![responses::ev_response_created(&id)]; + } else { + let mut added = responses::ev_assistant_message("progress", ""); + added["type"] = json!("response.output_item.added"); + events.push(added); + events.push(responses::ev_output_text_delta("Useful progress")); + let mut event = responses::ev_assistant_message("progress", "Useful progress"); + event["item"]["phase"] = json!(recovery); + events.push(event); + } + } + let mut empty_final = responses::ev_assistant_message(&format!("empty-final-{turn}"), ""); + empty_final["item"]["phase"] = json!("final_answer"); + events.push(empty_final); + events.push(responses::ev_completed(&id)); + scripts.push(responses::sse(events)); + } + let server = create_mock_responses_server_sequence(scripts).await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()) + .with_model("gpt-5.4") + .enable_feature(Feature::Goals) + .write(codex_home.path())?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + .build_initialized() + .await?; + let request = mcp + .send_thread_start_request_with_auto_env(ThreadStartParams::default()) + .await?; + let ThreadStartResponse { thread, .. } = mcp.read_response(request).await?; + let request = mcp + .send_raw_request( + "thread/goal/set", + Some(json!({"threadId": thread.id, "objective": "Finish the work"})), + ) + .await?; + let _: ThreadGoalSetResponse = mcp.read_response(request).await?; + + for turn in 1..=turns { + if turn == 3 && matches!(recovery, Some("final_answer" | "commentary")) { + let delta: serde_json::Value = timeout( + Duration::from_secs(30), + mcp.read_notification("item/agentMessage/delta"), + ) + .await??; + assert_eq!(json!("Useful progress"), delta["delta"]); + } + let completed: TurnCompletedNotification = timeout( + Duration::from_secs(30), + mcp.read_notification("turn/completed"), + ) + .await??; + assert_eq!(TurnStatus::Completed, 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 + ); + server.verify().await; + Ok(()) +} diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 219b8f584e..827542947f 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -2521,6 +2521,15 @@ impl Session { item: TurnItem, ) { record_turn_ttfm_metric(turn_context, &item).await; + for contributor in self.services.extensions.turn_lifecycle_contributors() { + contributor + .on_item_completed( + &self.services.thread_extension_data, + turn_context.extension_data.as_ref(), + &item, + ) + .await; + } let completed_at_ms = now_unix_timestamp_ms(); let item_id = item.id(); let started_at_ms = turn_context diff --git a/codex-rs/ext/extension-api/src/contributors.rs b/codex-rs/ext/extension-api/src/contributors.rs index 8c75997b63..89480df2f1 100644 --- a/codex-rs/ext/extension-api/src/contributors.rs +++ b/codex-rs/ext/extension-api/src/contributors.rs @@ -194,6 +194,16 @@ pub trait TurnLifecycleContributor: Send + Sync { }) } + /// Observes a completed item without changing it or delaying streamed deltas. + fn on_item_completed<'a>( + &'a self, + _thread_store: &'a ExtensionData, + _turn_store: &'a ExtensionData, + _item: &'a TurnItem, + ) -> ExtensionFuture<'a, ()> { + Box::pin(std::future::ready(())) + } + /// Called before the host drops the completed turn runtime and turn store. fn on_turn_stop<'a>(&'a self, input: TurnStopInput<'a>) -> ExtensionFuture<'a, ()> { Box::pin(async move { diff --git a/codex-rs/ext/goal/src/accounting.rs b/codex-rs/ext/goal/src/accounting.rs index 6a4a29978d..b2a89ce2ed 100644 --- a/codex-rs/ext/goal/src/accounting.rs +++ b/codex-rs/ext/goal/src/accounting.rs @@ -1,6 +1,9 @@ use codex_extension_api::ToolCallOutcome; use codex_extension_api::ToolName; use codex_protocol::config_types::ModeKind; +use codex_protocol::items::AgentMessageContent; +use codex_protocol::items::TurnItem; +use codex_protocol::models::MessagePhase; use codex_protocol::protocol::TokenUsage; use codex_state::ThreadGoalStatus; use std::collections::HashMap; @@ -28,6 +31,8 @@ struct GoalAccountingInner { budget_limit_reported_goal_id: Option, execution_failure_goal_id: Option, consecutive_execution_failure_turns: u8, + automatic_goal_turn_id: Option, + consecutive_empty_turns: u8, last_accounted_descendant_token_usage: i64, } @@ -39,6 +44,8 @@ struct GoalTurnAccounting { account_tokens: bool, failed_execution: bool, successful_tool: bool, + empty_final: bool, + has_activity: bool, } #[derive(Debug)] @@ -106,9 +113,11 @@ impl GoalAccountingState { outcome: ToolCallOutcome, ) { let mut inner = self.inner(); + inner.consecutive_empty_turns = 0; let Some(turn) = inner.turns.get_mut(turn_id) else { return; }; + turn.has_activity = true; if turn.active_goal_id.is_none() { return; } @@ -151,6 +160,75 @@ impl GoalAccountingState { (inner.consecutive_execution_failure_turns >= 3).then_some(goal_id) } + pub(crate) fn record_item(&self, turn_id: &str, item: &TurnItem) { + let mut inner = self.inner(); + let Some(turn) = inner.turns.get_mut(turn_id) else { + return; + }; + match item { + TurnItem::AgentMessage(message) => { + let has_text = message.content.iter().any(|content| match content { + AgentMessageContent::Text { text } => !text.trim().is_empty(), + }); + turn.has_activity |= has_text || message.questions.is_some(); + turn.empty_final |= + !has_text && !matches!(message.phase, Some(MessagePhase::Commentary)); + } + TurnItem::Reasoning(reasoning) => { + turn.has_activity |= reasoning + .summary_text + .iter() + .any(|text| !text.trim().is_empty()); + } + TurnItem::UserMessage(_) + | TurnItem::FunctionCallOutput(_) + | TurnItem::HookPrompt(_) + | TurnItem::Plan(_) + | TurnItem::CommandExecution(_) + | TurnItem::DynamicToolCall(_) + | TurnItem::CollabAgentToolCall(_) + | TurnItem::SubAgentActivity(_) + | TurnItem::WebSearch(_) + | TurnItem::ImageView(_) + | TurnItem::Extension(_) + | TurnItem::ImageGeneration(_) + | TurnItem::EnteredReviewMode(_) + | TurnItem::ExitedReviewMode(_) + | TurnItem::FileChange(_) + | TurnItem::McpToolCall(_) + | TurnItem::ContextCompaction(_) => turn.has_activity = true, + } + if turn.has_activity { + inner.consecutive_empty_turns = 0; + } + } + + pub(crate) fn mark_goal_continuation(&self, turn_id: String) { + self.inner().automatic_goal_turn_id = Some(turn_id); + } + + pub(crate) fn reset_empty_responses(&self) { + let mut inner = self.inner(); + inner.automatic_goal_turn_id = None; + inner.consecutive_empty_turns = 0; + } + + /// 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 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; + turn.empty_final = false; + if !empty { + inner.consecutive_empty_turns = 0; + return None; + } + inner.consecutive_empty_turns = inner.consecutive_empty_turns.saturating_add(1); + (inner.consecutive_empty_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 @@ -215,6 +293,7 @@ impl GoalAccountingState { turn.active_goal_id = Some(goal_id.clone()); if inner.current_turn_id.as_deref() == Some(turn_id) { if inner.wall_clock.active_goal_id.as_deref() != Some(goal_id.as_str()) { + inner.consecutive_empty_turns = 0; inner.last_accounted_descendant_token_usage = self.descendant_token_usage.load(Ordering::Relaxed); } @@ -242,6 +321,8 @@ impl GoalAccountingState { turn.active_goal_id = Some(goal_id.clone()); if goal_changed { turn.reset_baseline_to_current(); + inner.automatic_goal_turn_id = None; + inner.consecutive_empty_turns = 0; inner.last_accounted_descendant_token_usage = self.descendant_token_usage.load(Ordering::Relaxed); } @@ -256,6 +337,7 @@ impl GoalAccountingState { inner.budget_limit_reported_goal_id = None; } if inner.wall_clock.active_goal_id.as_deref() != Some(goal_id.as_str()) { + inner.consecutive_empty_turns = 0; inner.last_accounted_descendant_token_usage = self.descendant_token_usage.load(Ordering::Relaxed); } @@ -272,6 +354,8 @@ impl GoalAccountingState { inner.budget_limit_reported_goal_id = None; inner.execution_failure_goal_id = None; inner.consecutive_execution_failure_turns = 0; + inner.automatic_goal_turn_id = None; + inner.consecutive_empty_turns = 0; Some(turn_id) } @@ -286,6 +370,8 @@ impl GoalAccountingState { inner.budget_limit_reported_goal_id = None; inner.execution_failure_goal_id = None; inner.consecutive_execution_failure_turns = 0; + inner.automatic_goal_turn_id = None; + inner.consecutive_empty_turns = 0; } pub(crate) fn progress_snapshot(&self, turn_id: &str) -> Option { @@ -454,6 +540,8 @@ impl Default for GoalAccountingInner { budget_limit_reported_goal_id: None, execution_failure_goal_id: None, consecutive_execution_failure_turns: 0, + automatic_goal_turn_id: None, + consecutive_empty_turns: 0, last_accounted_descendant_token_usage: 0, } } @@ -479,6 +567,8 @@ impl GoalTurnAccounting { account_tokens, failed_execution: false, successful_tool: false, + empty_final: false, + has_activity: false, } } diff --git a/codex-rs/ext/goal/src/extension.rs b/codex-rs/ext/goal/src/extension.rs index 89848996fa..e95280482e 100644 --- a/codex-rs/ext/goal/src/extension.rs +++ b/codex-rs/ext/goal/src/extension.rs @@ -27,6 +27,7 @@ use codex_extension_api::TurnStartInput; use codex_extension_api::TurnStopInput; use codex_otel::MetricsClient; use codex_protocol::ThreadId; +use codex_protocol::items::TurnItem; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; @@ -271,6 +272,23 @@ where }) } + fn on_item_completed<'a>( + &'a self, + thread_store: &'a ExtensionData, + turn_store: &'a ExtensionData, + item: &'a TurnItem, + ) -> ExtensionFuture<'a, ()> { + Box::pin(async move { + if let Some(runtime) = goal_runtime_handle(thread_store) + && runtime.is_enabled() + { + runtime + .accounting_state() + .record_item(turn_store.level_id(), item); + } + }) + } + fn on_turn_stop<'a>(&'a self, input: TurnStopInput<'a>) -> ExtensionFuture<'a, ()> { Box::pin(async move { let Some(runtime) = goal_runtime_handle(input.thread_store) else { @@ -296,6 +314,14 @@ where ); return; } + if let Err(err) = runtime + .stop_active_goal_for_turn(turn_id, ActiveGoalStopReason::EmptyResponse) + .await + { + input.thread_store.remove::(); + tracing::warn!("failed to stop goal after empty responses for {turn_id}: {err}"); + return; + } if let Err(err) = runtime .account_active_goal_progress( turn_id, @@ -334,6 +360,7 @@ where let Some(runtime) = goal_runtime_handle(input.thread_store) else { return; }; + runtime.accounting_state().reset_empty_responses(); if !runtime.is_enabled() { return; } diff --git a/codex-rs/ext/goal/src/runtime.rs b/codex-rs/ext/goal/src/runtime.rs index 84fa4204ba..477290c751 100644 --- a/codex-rs/ext/goal/src/runtime.rs +++ b/codex-rs/ext/goal/src/runtime.rs @@ -40,6 +40,7 @@ pub(crate) enum ActiveGoalStopReason { TurnError, UsageLimit, ExecutionUnavailable { expected_goal_id: String }, + EmptyResponse, } struct GoalRuntimeInner { @@ -157,6 +158,8 @@ impl GoalRuntimeHandle { if !self.is_enabled() { return Ok(()); } + // Invalidate the old turn before the persisted objective/status changes. + self.inner.accounting_state.reset_empty_responses(); if let Some(turn_id) = self.inner.accounting_state.current_turn_id() { self.account_active_goal_progress( @@ -187,6 +190,7 @@ impl GoalRuntimeHandle { return Ok(()); } + self.inner.accounting_state.reset_empty_responses(); let replaced_existing_goal = previous_goal .as_ref() .is_some_and(|previous_goal| previous_goal.goal_id != goal.goal_id); @@ -262,7 +266,7 @@ impl GoalRuntimeHandle { .await } - /// Accounts the ending turn and stops its active goal after a terminal error. + /// Accounts the ending turn and stops its active goal after an error or repeated empty output. pub(crate) async fn stop_active_goal_for_turn( &self, turn_id: &str, @@ -297,6 +301,21 @@ impl GoalRuntimeHandle { codex_state::ThreadGoalStatus::UsageLimited, None, ), + ActiveGoalStopReason::EmptyResponse => { + let Some(expected_goal_id) = + self.inner.accounting_state.empty_response_goal(turn_id) + else { + return Ok(()); + }; + if accounting_goal_id != expected_goal_id { + return Ok(()); + } + ( + "empty-response", + codex_state::ThreadGoalStatus::Blocked, + Some(expected_goal_id), + ) + } ActiveGoalStopReason::ExecutionUnavailable { expected_goal_id } => ( "execution-unavailable", codex_state::ThreadGoalStatus::Blocked, @@ -459,7 +478,11 @@ impl GoalRuntimeHandle { ) .await { - Ok(StartIfIdleSubmission::Started { .. }) => {} + Ok(StartIfIdleSubmission::Started { turn_id }) => { + // Turn-stop evaluation takes the same permit, so even a fast response + // cannot finish before this host-admitted continuation is identified. + self.inner.accounting_state.mark_goal_continuation(turn_id); + } Ok(StartIfIdleSubmission::NotSubmitted { reason }) => { tracing::debug!( ?reason, diff --git a/codex-rs/ext/goal/tests/accounting.rs b/codex-rs/ext/goal/tests/accounting.rs index 556950c0a6..49f0d9251a 100644 --- a/codex-rs/ext/goal/tests/accounting.rs +++ b/codex-rs/ext/goal/tests/accounting.rs @@ -8,6 +8,10 @@ use accounting::GoalAccountingState; use codex_extension_api::ToolCallOutcome; use codex_extension_api::ToolName; use codex_protocol::config_types::ModeKind; +use codex_protocol::items::AgentMessageContent; +use codex_protocol::items::AgentMessageItem; +use codex_protocol::items::TurnItem; +use codex_protocol::models::MessagePhase; use codex_protocol::protocol::TokenUsage; use codex_state::ThreadGoalStatus; use pretty_assertions::assert_eq; @@ -55,6 +59,61 @@ fn goal_accounting_ignores_plan_mode_turns() { assert_eq!(None, recorded); } +#[test] +fn empty_continuations_require_three_turns_without_activity_or_goal_changes() { + let empty_final = TurnItem::AgentMessage(AgentMessageItem { + id: "empty".into(), + content: vec![AgentMessageContent::Text { text: " \n".into() }], + phase: Some(MessagePhase::FinalAnswer), + memory_citation: None, + delivery: None, + questions: None, + }); + for (interruption, blocking_turn) in [ + ("none", 3), + ("user", 6), + ("tool", 6), + ("goal", 5), + ("reset", 6), + ("missing final", 6), + ] { + let state = GoalAccountingState::default(); + for turn in 1..=blocking_turn { + let id = turn.to_string(); + let goal_id = if interruption == "goal" && turn >= 3 { + "new" + } else { + "goal" + }; + state.start_turn(&id, ModeKind::Default, &TokenUsage::default()); + state.mark_turn_goal_active(&id, goal_id); + if !(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") { + state.mark_goal_continuation(id.clone()); + } + if turn == 3 && interruption == "tool" { + state.record_tool_outcome( + &id, + &ToolName::plain("shell"), + ToolCallOutcome::Completed { success: false }, + ); + } + if turn == 3 && interruption == "reset" { + state.reset_empty_responses(); + } + assert_eq!( + (turn == blocking_turn).then(|| goal_id.to_string()), + state.empty_response_goal(&id), + "interruption: {interruption}, turn: {turn}", + ); + state.finish_turn(&id); + } + } +} + #[test] fn execution_failures_do_not_transfer_to_a_replacement_goal() { let state = GoalAccountingState::default();