Add SubscriptionExhausted hook for auth rotation

codex-rs/analytics/src/events.rs

@@ -679,6 +679,7 @@ fn analytics_hook_event_name(event_name: HookEventName) -> &'static str {
         HookEventName::SessionStart => "SessionStart",
         HookEventName::UserPromptSubmit => "UserPromptSubmit",
         HookEventName::Stop => "Stop",
+        HookEventName::SubscriptionExhausted => "SubscriptionExhausted",
     }
 }
 

codex-rs/app-server-protocol/src/protocol/v2.rs

@@ -445,7 +445,7 @@ v2_enum_from_core!(
 
 v2_enum_from_core!(
     pub enum HookEventName from CoreHookEventName {
-        PreToolUse, PermissionRequest, PostToolUse, SessionStart, UserPromptSubmit, Stop
+        PreToolUse, PermissionRequest, PostToolUse, SessionStart, UserPromptSubmit, Stop, SubscriptionExhausted
     }
 );
 
@@ -988,6 +988,9 @@ pub struct ManagedHooksRequirements {
     #[serde(rename = "Stop")]
     #[ts(rename = "Stop")]
     pub stop: Vec<ConfiguredHookMatcherGroup>,
+    #[serde(rename = "SubscriptionExhausted")]
+    #[ts(rename = "SubscriptionExhausted")]
+    pub subscription_exhausted: Vec<ConfiguredHookMatcherGroup>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]

codex-rs/app-server/src/config_api.rs

@@ -308,6 +308,7 @@ fn map_hooks_requirements_to_api(hooks: ManagedHooksRequirementsToml) -> Managed
         session_start,
         user_prompt_submit,
         stop,
+        subscription_exhausted,
     } = hooks;
 
     ManagedHooksRequirements {
@@ -319,6 +320,7 @@ fn map_hooks_requirements_to_api(hooks: ManagedHooksRequirementsToml) -> Managed
         session_start: map_hook_matcher_groups_to_api(session_start),
         user_prompt_submit: map_hook_matcher_groups_to_api(user_prompt_submit),
         stop: map_hook_matcher_groups_to_api(stop),
+        subscription_exhausted: map_hook_matcher_groups_to_api(subscription_exhausted),
     }
 }
 
@@ -627,6 +629,7 @@ mod tests {
                 session_start: Vec::new(),
                 user_prompt_submit: Vec::new(),
                 stop: Vec::new(),
+                subscription_exhausted: Vec::new(),
             }),
         );
         assert_eq!(

codex-rs/config/src/hook_config.rs

@@ -41,6 +41,8 @@ pub struct HookEventsToml {
     pub user_prompt_submit: Vec<MatcherGroup>,
     #[serde(rename = "Stop", default)]
     pub stop: Vec<MatcherGroup>,
+    #[serde(rename = "SubscriptionExhausted", default)]
+    pub subscription_exhausted: Vec<MatcherGroup>,
 }
 
 impl HookEventsToml {
@@ -52,13 +54,15 @@ impl HookEventsToml {
             session_start,
             user_prompt_submit,
             stop,
+            subscription_exhausted,
         } = self;
         pre_tool_use.is_empty()
             && permission_request.is_empty()
             && post_tool_use.is_empty()
             && session_start.is_empty()
             && user_prompt_submit.is_empty()
             && stop.is_empty()
+            && subscription_exhausted.is_empty()
     }
 
     pub fn handler_count(&self) -> usize {
@@ -69,6 +73,7 @@ impl HookEventsToml {
             session_start,
             user_prompt_submit,
             stop,
+            subscription_exhausted,
         } = self;
         [
             pre_tool_use,
@@ -77,21 +82,26 @@ impl HookEventsToml {
             session_start,
             user_prompt_submit,
             stop,
+            subscription_exhausted,
         ]
         .into_iter()
         .flatten()
         .map(|group| group.hooks.len())
         .sum()
     }
 
-    pub fn into_matcher_groups(self) -> [(HookEventName, Vec<MatcherGroup>); 6] {
+    pub fn into_matcher_groups(self) -> [(HookEventName, Vec<MatcherGroup>); 7] {
         [
             (HookEventName::PreToolUse, self.pre_tool_use),
             (HookEventName::PermissionRequest, self.permission_request),
             (HookEventName::PostToolUse, self.post_tool_use),
             (HookEventName::SessionStart, self.session_start),
             (HookEventName::UserPromptSubmit, self.user_prompt_submit),
             (HookEventName::Stop, self.stop),
+            (
+                HookEventName::SubscriptionExhausted,
+                self.subscription_exhausted,
+            ),
         ]
     }
 }

codex-rs/core/config.schema.json

@@ -1037,6 +1037,13 @@
           },
           "type": "array"
         },
+        "SubscriptionExhausted": {
+          "default": [],
+          "items": {
+            "$ref": "#/definitions/MatcherGroup"
+          },
+          "type": "array"
+        },
         "UserPromptSubmit": {
           "default": [],
           "items": {

codex-rs/core/src/client.rs

@@ -828,6 +828,15 @@ impl ModelClientSession {
             .set_connection_reused(/*connection_reused*/ false);
     }
 
+    /// Clears transport and sticky-routing state after an out-of-band auth or account switch.
+    ///
+    /// Reusing the previous `x-codex-turn-state` can keep a retry pinned to the same exhausted
+    /// route even after the hook has selected a different account.
+    pub(crate) fn reset_after_subscription_exhaustion_recovery(&mut self) {
+        self.reset_websocket_session();
+        self.turn_state = Arc::new(OnceLock::new());
+    }
+
     fn build_responses_request(
         &self,
         provider: &codex_api::Provider,

codex-rs/core/src/hook_runtime.rs

@@ -12,6 +12,8 @@ use codex_hooks::PostToolUseRequest;
 use codex_hooks::PreToolUseOutcome;
 use codex_hooks::PreToolUseRequest;
 use codex_hooks::SessionStartOutcome;
+use codex_hooks::SubscriptionExhaustedOutcome;
+use codex_hooks::SubscriptionExhaustedRequest;
 use codex_hooks::UserPromptSubmitOutcome;
 use codex_hooks::UserPromptSubmitRequest;
 use codex_otel::HOOK_RUN_DURATION_METRIC;
@@ -278,6 +280,24 @@ pub(crate) async fn run_user_prompt_submit_hooks(
     .await
 }
 
+pub(crate) async fn run_subscription_exhausted_hooks(
+    sess: &Arc<Session>,
+    turn_context: &Arc<TurnContext>,
+    request: SubscriptionExhaustedRequest,
+) -> bool {
+    let hooks = sess.hooks();
+    let preview_runs = hooks.preview_subscription_exhausted(&request);
+    emit_hook_started_events(sess, turn_context, preview_runs).await;
+
+    let SubscriptionExhaustedOutcome { hook_events } =
+        hooks.run_subscription_exhausted(request).await;
+    let succeeded = hook_events
+        .iter()
+        .any(|event| event.run.status == HookRunStatus::Completed);
+    emit_hook_completed_events(sess, turn_context, hook_events).await;
+    succeeded
+}
+
 pub(crate) async fn inspect_pending_input(
     sess: &Arc<Session>,
     turn_context: &Arc<TurnContext>,
@@ -470,6 +490,7 @@ fn hook_run_metric_tags(run: &HookRunSummary) -> [(&'static str, &'static str);
         HookEventName::SessionStart => "SessionStart",
         HookEventName::UserPromptSubmit => "UserPromptSubmit",
         HookEventName::Stop => "Stop",
+        HookEventName::SubscriptionExhausted => "SubscriptionExhausted",
     };
     let hook_source = match run.source {
         HookSource::System => "system",
@@ -498,7 +519,7 @@ fn hook_run_metric_tags(run: &HookRunSummary) -> [(&'static str, &'static str);
     ]
 }
 
-fn hook_permission_mode(turn_context: &TurnContext) -> String {
+pub(crate) fn hook_permission_mode(turn_context: &TurnContext) -> String {
     match turn_context.approval_policy.value() {
         AskForApproval::Never => "bypassPermissions",
         AskForApproval::UnlessTrusted

codex-rs/core/src/lib.rs

@@ -105,6 +105,7 @@ pub(crate) use skills::resolve_skill_dependencies_for_turn;
 pub(crate) use skills::skills_load_input_from_config;
 mod skills_watcher;
 mod stream_events_utils;
+mod subscription_exhaustion;
 pub mod test_support;
 mod unified_exec;
 pub mod windows_sandbox;

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

@@ -49,6 +49,7 @@ use crate::stream_events_utils::last_assistant_message_from_item;
 use crate::stream_events_utils::mark_thread_memory_mode_polluted_if_external_context;
 use crate::stream_events_utils::raw_assistant_output_text_from_item;
 use crate::stream_events_utils::record_completed_response_item;
+use crate::subscription_exhaustion;
 use crate::tools::ToolRouter;
 use crate::tools::context::SharedTurnDiffTracker;
 use crate::tools::parallel::ToolCallRuntime;
@@ -372,6 +373,7 @@ pub(crate) async fn run_turn(
     // 1. At the start of a turn, so the fresh user prompt in `input` gets sampled first.
     // 2. After auto-compact, when model/tool continuation needs to resume before any steer.
     let mut can_drain_pending_input = input.is_empty();
+    let mut subscription_exhaustion_hook_attempted = false;
 
     loop {
         if run_pending_session_start_hooks(&sess, &turn_context).await {
@@ -652,6 +654,33 @@ pub(crate) async fn run_turn(
                 break;
             }
             Err(e) => {
+                if subscription_exhaustion::should_run_hook_for_error(&e)
+                    && !subscription_exhaustion_hook_attempted
+                {
+                    subscription_exhaustion_hook_attempted = true;
+                    if let Some(recovery) = subscription_exhaustion::recover_with_hook_if_available(
+                        &sess,
+                        &turn_context,
+                        &e,
+                    )
+                    .await
+                    {
+                        client_session.reset_after_subscription_exhaustion_recovery();
+                        let message = if recovery.auth_changed {
+                            "A SubscriptionExhausted hook refreshed Codex auth. Reconnecting and retrying request."
+                        } else {
+                            "A SubscriptionExhausted hook completed. Reconnecting and retrying request."
+                        };
+                        sess.send_event(
+                            &turn_context,
+                            EventMsg::Warning(WarningEvent {
+                                message: message.to_string(),
+                            }),
+                        )
+                        .await;
+                        continue;
+                    }
+                }
                 info!("Turn error: {e:#}");
                 let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));
                 sess.send_event(&turn_context, event).await;