feat: reapply fork behavior on rust-v0.145.0

Author: Loongphy

codex-rs/Cargo.lock

@@ -489,6 +489,13 @@ pub struct GetAccountParams {
     /// themselves and call `account/login/start` with `chatgptAuthTokens`.
     #[serde(default, skip_serializing_if = "std::ops::Not::not")]
     pub refresh_token: bool,
+
+    /// When `true`, reloads the auth snapshot from storage before returning.
+    ///
+    /// This keeps long-lived clients in sync with `auth.json` updates without
+    /// requiring a full app-server restart.
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    pub reload_auth_from_storage: bool,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -497,6 +504,8 @@ pub struct GetAccountParams {
 pub struct GetAccountResponse {
     pub account: Option<Account>,
     pub requires_openai_auth: bool,
+    /// Whether this request reloaded a different auth snapshot from storage.
+    pub auth_changed: bool,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]

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

@@ -311,6 +311,7 @@ impl MessageProcessor {
         let pending_thread_unloads = Arc::new(Mutex::new(HashSet::new()));
         let thread_watch_manager =
             crate::thread_status::ThreadWatchManager::new_with_outgoing(outgoing.clone());
+        let auth_transition_lock = Arc::new(Mutex::new(()));
         let thread_list_state_permit = Arc::new(Semaphore::new(/*permits*/ 1));
         let workspace_settings_cache =
             Arc::new(workspace_settings::WorkspaceSettingsCache::default());
@@ -336,6 +337,8 @@ impl MessageProcessor {
             outgoing.clone(),
             Arc::clone(&config),
             config_manager.clone(),
+            thread_watch_manager.clone(),
+            Arc::clone(&auth_transition_lock),
         );
         let apps_processor = AppsRequestProcessor::new(
             auth_manager.clone(),
@@ -422,6 +425,7 @@ impl MessageProcessor {
             Arc::clone(&pending_thread_unloads),
             thread_state_manager.clone(),
             thread_watch_manager.clone(),
+            Arc::clone(&auth_transition_lock),
             Arc::clone(&thread_list_state_permit),
             thread_goal_processor.clone(),
             state_db.clone(),
@@ -440,6 +444,7 @@ impl MessageProcessor {
             pending_thread_unloads,
             thread_state_manager,
             thread_watch_manager,
+            auth_transition_lock,
             thread_list_state_permit,
             Arc::clone(&skills_watcher),
         );

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

@@ -1,3 +1,4 @@
+use crate::auth_mode::auth_mode_to_api;
 use crate::bespoke_event_handling::apply_bespoke_event_handling;
 use crate::command_exec::CommandExecManager;
 use crate::command_exec::StartCommandExecParams;
@@ -394,6 +395,7 @@ use codex_feedback::FeedbackUploadOptions;
 use codex_git_utils::git_diff_to_remote;
 use codex_git_utils::resolve_root_git_project_for_trust;
 use codex_login::AuthManager;
+use codex_login::AuthReloadStatus;
 use codex_login::CODEX_OPEN_APP_URL;
 use codex_login::CodexAuth;
 use codex_login::LoginSuccessPage;
@@ -517,6 +519,81 @@ use uuid::Uuid;
 #[cfg(test)]
 use codex_app_server_protocol::ServerRequest;
 
+async fn reload_auth_from_storage_if_idle(
+    auth_manager: &Arc<AuthManager>,
+    thread_manager: &Arc<ThreadManager>,
+    config_manager: &ConfigManager,
+    outgoing: &OutgoingMessageSender,
+    thread_watch_manager: &ThreadWatchManager,
+    chatgpt_base_url: &str,
+    reason: &str,
+) {
+    if *thread_watch_manager.subscribe_running_turn_count().borrow() != 0 {
+        return;
+    }
+
+    let status = auth_manager.reload_with_status().await;
+    match handle_auth_reload_status(
+        status,
+        auth_manager,
+        thread_manager,
+        config_manager,
+        outgoing,
+        chatgpt_base_url,
+        reason,
+    )
+    .await
+    {
+        AuthReloadStatus::Reloaded { .. } => {}
+        AuthReloadStatus::Failed => {
+            warn!("failed to reload auth from storage before {reason}");
+        }
+    }
+}
+
+async fn handle_auth_reload_status(
+    status: AuthReloadStatus,
+    auth_manager: &Arc<AuthManager>,
+    thread_manager: &Arc<ThreadManager>,
+    config_manager: &ConfigManager,
+    outgoing: &OutgoingMessageSender,
+    chatgpt_base_url: &str,
+    reason: &str,
+) -> AuthReloadStatus {
+    match status {
+        AuthReloadStatus::Reloaded { changed } => {
+            if changed {
+                let invalidated_thread_count =
+                    thread_manager.invalidate_model_transport_caches().await;
+                info!(
+                    "auth reloaded from storage before {reason}; invalidated model transport caches for {invalidated_thread_count} tracked thread(s)"
+                );
+                config_manager.replace_cloud_config_bundle_loader(
+                    Arc::clone(auth_manager),
+                    chatgpt_base_url.to_string(),
+                );
+                config_manager
+                    .sync_default_client_residency_requirement()
+                    .await;
+                let auth = auth_manager.auth_cached();
+                outgoing
+                    .send_server_notification(ServerNotification::AccountUpdated(
+                        AccountUpdatedNotification {
+                            auth_mode: auth
+                                .as_ref()
+                                .map(CodexAuth::api_auth_mode)
+                                .map(auth_mode_to_api),
+                            plan_type: auth.as_ref().and_then(CodexAuth::account_plan_type),
+                        },
+                    ))
+                    .await;
+            }
+            AuthReloadStatus::Reloaded { changed }
+        }
+        AuthReloadStatus::Failed => AuthReloadStatus::Failed,
+    }
+}
+
 mod account_processor;
 mod apps_processor;
 mod bedrock_auth;

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

@@ -73,6 +73,8 @@ pub(crate) struct AccountRequestProcessor {
     outgoing: Arc<OutgoingMessageSender>,
     config: Arc<Config>,
     config_manager: ConfigManager,
+    thread_watch_manager: ThreadWatchManager,
+    auth_transition_lock: Arc<Mutex<()>>,
     active_login: Arc<Mutex<Option<ActiveLogin>>>,
 }
 
@@ -83,13 +85,17 @@ impl AccountRequestProcessor {
         outgoing: Arc<OutgoingMessageSender>,
         config: Arc<Config>,
         config_manager: ConfigManager,
+        thread_watch_manager: ThreadWatchManager,
+        auth_transition_lock: Arc<Mutex<()>>,
     ) -> Self {
         Self {
             auth_manager,
             thread_manager,
             outgoing,
             config,
             config_manager,
+            thread_watch_manager,
+            auth_transition_lock,
             active_login: Arc::new(Mutex::new(None)),
         }
     }
@@ -995,6 +1001,35 @@ impl AccountRequestProcessor {
         params: GetAccountParams,
     ) -> Result<GetAccountResponse, JSONRPCErrorError> {
         let do_refresh = params.refresh_token;
+        let mut auth_changed = false;
+
+        if params.reload_auth_from_storage {
+            let _auth_transition_guard = self.auth_transition_lock.lock().await;
+            if *self
+                .thread_watch_manager
+                .subscribe_running_turn_count()
+                .borrow()
+                == 0
+            {
+                let status = self.auth_manager.reload_with_status().await;
+                match handle_auth_reload_status(
+                    status,
+                    &self.auth_manager,
+                    &self.thread_manager,
+                    &self.config_manager,
+                    &self.outgoing,
+                    &self.config.chatgpt_base_url,
+                    "account/get",
+                )
+                .await
+                {
+                    AuthReloadStatus::Reloaded { changed } => auth_changed = changed,
+                    AuthReloadStatus::Failed => {
+                        return Err(internal_error("failed to reload auth from storage"));
+                    }
+                }
+            }
+        }
 
         self.refresh_token_if_requested(do_refresh).await;
 
@@ -1010,6 +1045,7 @@ impl AccountRequestProcessor {
         Ok(GetAccountResponse {
             account,
             requires_openai_auth: account_state.requires_openai_auth,
+            auth_changed,
         })
     }
 

codex-rs/app-server/src/request_processors/account_processor.rs

@@ -388,6 +388,7 @@ pub(crate) struct ThreadRequestProcessor {
     pub(super) pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
     pub(super) thread_state_manager: ThreadStateManager,
     pub(super) thread_watch_manager: ThreadWatchManager,
+    pub(super) auth_transition_lock: Arc<Mutex<()>>,
     pub(super) thread_list_state_permit: Arc<Semaphore>,
     pub(super) thread_goal_processor: ThreadGoalRequestProcessor,
     pub(super) state_db: Option<StateDbHandle>,
@@ -421,6 +422,7 @@ impl ThreadRequestProcessor {
         pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
         thread_state_manager: ThreadStateManager,
         thread_watch_manager: ThreadWatchManager,
+        auth_transition_lock: Arc<Mutex<()>>,
         thread_list_state_permit: Arc<Semaphore>,
         thread_goal_processor: ThreadGoalRequestProcessor,
         state_db: Option<StateDbHandle>,
@@ -439,6 +441,7 @@ impl ThreadRequestProcessor {
             pending_thread_unloads,
             thread_state_manager,
             thread_watch_manager,
+            auth_transition_lock,
             thread_list_state_permit,
             thread_goal_processor,
             state_db,
@@ -947,6 +950,18 @@ impl ThreadRequestProcessor {
         supports_openai_form_elicitation: bool,
         request_context: RequestContext,
     ) -> Result<(), JSONRPCErrorError> {
+        let _auth_transition_guard = self.auth_transition_lock.lock().await;
+        reload_auth_from_storage_if_idle(
+            &self.auth_manager,
+            &self.thread_manager,
+            &self.config_manager,
+            &self.outgoing,
+            &self.thread_watch_manager,
+            &self.config.chatgpt_base_url,
+            "thread/start",
+        )
+        .await;
+
         let ThreadStartParams {
             model,
             model_provider,
@@ -3019,6 +3034,18 @@ impl ThreadRequestProcessor {
         app_server_client_version: Option<String>,
         supports_openai_form_elicitation: bool,
     ) -> Result<(), JSONRPCErrorError> {
+        let _auth_transition_guard = self.auth_transition_lock.lock().await;
+        reload_auth_from_storage_if_idle(
+            &self.auth_manager,
+            &self.thread_manager,
+            &self.config_manager,
+            &self.outgoing,
+            &self.thread_watch_manager,
+            &self.config.chatgpt_base_url,
+            "thread/resume",
+        )
+        .await;
+
         if let Ok(thread_id) = ThreadId::from_string(&params.thread_id)
             && self
                 .pending_thread_unloads

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -95,6 +95,7 @@ pub(crate) struct TurnRequestProcessor {
     pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
     thread_state_manager: ThreadStateManager,
     thread_watch_manager: ThreadWatchManager,
+    auth_transition_lock: Arc<Mutex<()>>,
     thread_list_state_permit: Arc<Semaphore>,
     skills_watcher: Arc<SkillsWatcher>,
 }
@@ -150,6 +151,7 @@ impl TurnRequestProcessor {
         pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
         thread_state_manager: ThreadStateManager,
         thread_watch_manager: ThreadWatchManager,
+        auth_transition_lock: Arc<Mutex<()>>,
         thread_list_state_permit: Arc<Semaphore>,
         skills_watcher: Arc<SkillsWatcher>,
     ) -> Self {
@@ -166,6 +168,7 @@ impl TurnRequestProcessor {
             pending_thread_unloads,
             thread_state_manager,
             thread_watch_manager,
+            auth_transition_lock,
             thread_list_state_permit,
             skills_watcher,
         }
@@ -478,6 +481,7 @@ impl TurnRequestProcessor {
         app_server_client_version: Option<String>,
         supports_openai_form_elicitation: bool,
     ) -> Result<TurnStartResponse, JSONRPCErrorError> {
+        let _auth_transition_guard = self.auth_transition_lock.lock().await;
         let (thread_id, thread) =
             self.load_thread(&params.thread_id)
                 .await
@@ -486,6 +490,16 @@ impl TurnRequestProcessor {
                 })?;
         self.ensure_direct_input_allowed(&request_id, thread.as_ref())
             .await?;
+        reload_auth_from_storage_if_idle(
+            &self.auth_manager,
+            &self.thread_manager,
+            &self.config_manager,
+            &self.outgoing,
+            &self.thread_watch_manager,
+            &self.config.chatgpt_base_url,
+            "turn/start",
+        )
+        .await;
         if let Err(error) = Self::validate_v2_input_limit(&params.input) {
             self.track_error_response(
                 &request_id,
@@ -583,6 +597,9 @@ impl TurnRequestProcessor {
                 self.track_error_response(&request_id, &error, /*error_type*/ None);
                 error
             })?;
+        self.thread_watch_manager
+            .note_turn_started(&thread_id.to_string())
+            .await;
 
         if turn_has_input {
             let config_snapshot = thread.config_snapshot().await;

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -779,6 +779,14 @@ pub struct Tui {
     #[serde(default)]
     pub keymap: TuiKeymap,
 
+    /// Optional synthetic user-turn prompt injected after a turn fails with
+    /// `UsageLimitExceeded`.
+    ///
+    /// When unset, Codext uses the built-in default recovery prompt.
+    /// When set to an empty string, Codext disables this automatic recovery turn.
+    #[serde(default)]
+    pub usage_limit_resume_prompt: Option<String>,
+
     /// Startup tooltip availability NUX state persisted by the TUI.
     #[serde(default)]
     pub model_availability_nux: ModelAvailabilityNuxConfig,

codex-rs/config/src/types.rs

@@ -3363,6 +3363,11 @@
           "description": "Syntax highlighting theme name (kebab-case).\n\nWhen set, overrides automatic light/dark theme detection. Use `/theme` in the TUI or see `$CODEX_HOME/themes` for custom themes.",
           "type": "string"
         },
+        "usage_limit_resume_prompt": {
+          "default": null,
+          "description": "Optional synthetic user-turn prompt injected after a turn fails with `UsageLimitExceeded`.\n\nWhen unset, Codext uses the built-in default recovery prompt. When set to an empty string, Codext disables this automatic recovery turn.",
+          "type": "string"
+        },
         "vim_mode_default": {
           "default": false,
           "description": "Start the composer in Vim mode (`Normal`) by default. Defaults to `false`.",

codex-rs/core/config.schema.json

@@ -506,6 +506,10 @@ impl ModelClient {
             .unwrap_or_else(std::sync::PoisonError::into_inner) = websocket_session;
     }
 
+    pub(crate) fn invalidate_cached_transport_state(&self) {
+        self.store_cached_websocket_session(WebsocketSession::default());
+    }
+
     pub(crate) fn force_http_fallback(
         &self,
         session_telemetry: &SessionTelemetry,