Retry goals after model capacity errors

codex-rs/core/src/session/inject.rs

@@ -46,9 +46,6 @@ impl Session {
         self: &Arc<Self>,
         input: Vec<ResponseItem>,
     ) -> Result<(), TryStartTurnIfIdleError> {
-        if input.is_empty() {
-            return Ok(());
-        }
         if self.input_queue.has_trigger_turn_mailbox_items().await {
             return Err(TryStartTurnIfIdleError::new(
                 TryStartTurnIfIdleRejectionReason::PendingTriggerTurn,

codex-rs/core/src/session/turn.rs

@@ -158,7 +158,7 @@ pub(crate) async fn run_turn(
             return Err(err);
         }
         let error = err.to_codex_protocol_error();
-        sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone())
+        sess.emit_compaction_turn_error_lifecycle(turn_context.as_ref(), error.clone())
             .await;
         error!("Failed to run pre-sampling compact");
         return Ok(None);
@@ -361,8 +361,11 @@ pub(crate) async fn run_turn(
                             return Err(err);
                         }
                         let error = err.to_codex_protocol_error();
-                        sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone())
-                            .await;
+                        sess.emit_compaction_turn_error_lifecycle(
+                            turn_context.as_ref(),
+                            error.clone(),
+                        )
+                        .await;
                         return Ok(None);
                     }
                     can_drain_pending_input = !model_needs_follow_up;

codex-rs/core/src/tasks/lifecycle.rs

@@ -76,12 +76,32 @@ impl Session {
         &self,
         turn_context: &TurnContext,
         error: CodexErrorInfo,
+    ) {
+        self.emit_turn_error_lifecycle_for(turn_context, error, /*is_compaction*/ false)
+            .await;
+    }
+
+    pub(crate) async fn emit_compaction_turn_error_lifecycle(
+        &self,
+        turn_context: &TurnContext,
+        error: CodexErrorInfo,
+    ) {
+        self.emit_turn_error_lifecycle_for(turn_context, error, /*is_compaction*/ true)
+            .await;
+    }
+
+    async fn emit_turn_error_lifecycle_for(
+        &self,
+        turn_context: &TurnContext,
+        error: CodexErrorInfo,
+        is_compaction: bool,
     ) {
         for contributor in self.services.extensions.turn_lifecycle_contributors() {
             contributor
                 .on_turn_error(codex_extension_api::TurnErrorInput {
                     turn_id: turn_context.sub_id.as_str(),
                     error: error.clone(),
+                    is_compaction,
                     session_store: &self.services.session_extension_data,
                     thread_store: &self.services.thread_extension_data,
                     turn_store: turn_context.extension_data.as_ref(),

codex-rs/ext/extension-api/src/contributors/turn_lifecycle.rs

@@ -49,6 +49,8 @@ pub struct TurnErrorInput<'a> {
     pub turn_id: &'a str,
     /// Error surfaced by the host for this turn.
     pub error: CodexErrorInfo,
+    /// Whether the error came from compaction rather than model sampling.
+    pub is_compaction: bool,
     /// Store scoped to the host session runtime.
     pub session_store: &'a ExtensionData,
     /// Store scoped to this thread runtime.

codex-rs/ext/goal/Cargo.toml

@@ -22,6 +22,7 @@ codex-protocol = { workspace = true }
 codex-state = { workspace = true }
 codex-tools = { workspace = true }
 codex-utils-template = { workspace = true }
+rand = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["sync"] }

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

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 use std::sync::Weak;
+use std::time::Duration;
 
 use codex_analytics::AnalyticsEventsClient;
 use codex_core::ThreadManager;
@@ -303,6 +304,18 @@ where
             };
 
             let reason = match input.error {
+                // Capacity failures do not consume user tokens. Jittering the
+                // delay prevents clients that fail together from retrying in lockstep.
+                CodexErrorInfo::ServerOverloaded
+                    if !input.is_compaction
+                        && runtime.tools_visible()
+                        && runtime
+                            .accounting_state()
+                            .turn_is_current_active_goal(input.turn_id) =>
+                {
+                    runtime.defer_retry(Duration::from_secs(rand::random_range(4 * 60..=6 * 60)));
+                    return;
+                }
                 CodexErrorInfo::UsageLimitExceeded => ActiveGoalStopReason::UsageLimit,
                 // The turn has ended because the error was non-retryable or its
                 // retries were exhausted. Block the goal to prevent automatic

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

@@ -2,6 +2,7 @@ use std::sync::Arc;
 use std::sync::Weak;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
+use std::time::Duration;
 
 use codex_core::ThreadManager;
 use codex_protocol::ThreadId;
@@ -45,6 +46,7 @@ struct GoalRuntimeInner {
     thread_manager: Weak<ThreadManager>,
     accounting_state: Arc<GoalAccountingState>,
     enabled: AtomicBool,
+    retry_pending: AtomicBool,
     tools_available_for_thread: bool,
     goal_state_lock: Semaphore,
 }
@@ -97,6 +99,7 @@ impl GoalRuntimeHandle {
                 thread_manager,
                 accounting_state,
                 enabled: AtomicBool::new(config.enabled),
+                retry_pending: AtomicBool::new(false),
                 tools_available_for_thread: config.tools_available_for_thread,
                 goal_state_lock: Semaphore::new(/*permits*/ 1),
             }),
@@ -115,6 +118,32 @@ impl GoalRuntimeHandle {
         self.is_enabled() && self.inner.tools_available_for_thread
     }
 
+    /// Suppresses ordinary idle continuation while a retry timer is pending,
+    /// then starts a new turn from the existing conversation history. The weak
+    /// reference ensures the timer neither keeps an unloaded runtime alive nor
+    /// survives a thread lifecycle.
+    pub(crate) fn defer_retry(&self, delay: Duration) {
+        if self.inner.retry_pending.swap(true, Ordering::Relaxed) {
+            return;
+        }
+
+        let runtime = Arc::downgrade(&self.inner);
+        drop(tokio::spawn(async move {
+            tokio::time::sleep(delay).await;
+            let Some(inner) = runtime.upgrade() else {
+                return;
+            };
+            inner.retry_pending.store(false, Ordering::Relaxed);
+            let runtime = GoalRuntimeHandle { inner };
+            if let Err(err) = runtime.retry_if_idle().await {
+                tracing::warn!(
+                    "failed to retry active goal for idle thread {}: {err}",
+                    runtime.thread_id()
+                );
+            }
+        }));
+    }
+
     pub(crate) fn thread_id(&self) -> ThreadId {
         self.inner.thread_id
     }
@@ -357,6 +386,21 @@ impl GoalRuntimeHandle {
     }
 
     pub(crate) async fn continue_if_idle(&self) -> Result<(), String> {
+        if self.inner.retry_pending.load(Ordering::Relaxed) {
+            return Ok(());
+        }
+        self.start_if_idle(|goal| vec![continuation_steering_item(&protocol_goal_from_state(goal))])
+            .await
+    }
+
+    async fn retry_if_idle(&self) -> Result<(), String> {
+        self.start_if_idle(|_| Vec::new()).await
+    }
+
+    async fn start_if_idle(
+        &self,
+        input_for_goal: impl FnOnce(codex_state::ThreadGoal) -> Vec<ResponseItem>,
+    ) -> Result<(), String> {
         if !self.tools_visible() {
             self.inner.accounting_state.clear_active_goal();
             return Ok(());
@@ -389,9 +433,8 @@ impl GoalRuntimeHandle {
             self.inner.accounting_state.clear_active_goal();
             return Ok(());
         }
-        let item = continuation_steering_item(&protocol_goal_from_state(goal));
 
-        if let Err(err) = thread.try_start_turn_if_idle(vec![item]).await {
+        if let Err(err) = thread.try_start_turn_if_idle(input_for_goal(goal)).await {
             let reason = err.reason();
             tracing::debug!(
                 ?reason,

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

@@ -520,7 +520,11 @@ async fn turn_error_usage_limit_accounts_progress_and_clears_accounting() -> any
         )
         .await;
     harness
-        .notify_turn_error("turn-1", CodexErrorInfo::UsageLimitExceeded)
+        .notify_turn_error(
+            "turn-1",
+            CodexErrorInfo::UsageLimitExceeded,
+            /*is_compaction*/ false,
+        )
         .await;
 
     let goal = runtime
@@ -574,7 +578,7 @@ async fn turn_error_usage_limit_accounts_progress_and_clears_accounting() -> any
 }
 
 #[tokio::test]
-async fn turn_error_blocks_goal() -> anyhow::Result<()> {
+async fn capacity_error_retries_sampling_but_not_compaction() -> anyhow::Result<()> {
     let runtime = test_runtime().await?;
     let thread_id = test_thread_id()?;
     seed_thread_metadata(runtime.as_ref(), thread_id).await?;
@@ -591,7 +595,26 @@ async fn turn_error_blocks_goal() -> anyhow::Result<()> {
         .await?;
 
     harness
-        .notify_turn_error("turn-1", CodexErrorInfo::Other)
+        .notify_turn_error(
+            "turn-1",
+            CodexErrorInfo::ServerOverloaded,
+            /*is_compaction*/ false,
+        )
+        .await;
+
+    let goal = runtime
+        .thread_goals()
+        .get_thread_goal(thread_id)
+        .await?
+        .ok_or_else(|| anyhow::anyhow!("goal should exist"))?;
+    assert_eq!(codex_state::ThreadGoalStatus::Active, goal.status);
+
+    harness
+        .notify_turn_error(
+            "turn-1",
+            CodexErrorInfo::ServerOverloaded,
+            /*is_compaction*/ true,
+        )
         .await;
 
     let goal = runtime
@@ -1305,13 +1328,14 @@ impl GoalExtensionHarness {
         }
     }
 
-    async fn notify_turn_error(&self, turn_id: &str, error: CodexErrorInfo) {
+    async fn notify_turn_error(&self, turn_id: &str, error: CodexErrorInfo, is_compaction: bool) {
         let turn_store = ExtensionData::new(turn_id);
         for contributor in self.registry.turn_lifecycle_contributors() {
             contributor
                 .on_turn_error(TurnErrorInput {
                     turn_id,
                     error: error.clone(),
+                    is_compaction,
                     session_store: &self.session_store,
                     thread_store: &self.thread_store,
                     turn_store: &turn_store,