fix(openai-codex): reduce managed OAuth false positives on refresh and 403

- Branch refresh_access_token on ManagedOAuthRefreshError: skip penalizing
  rotation for transient and needs_reauth; soft-rotate for other failures.
- Reload disk state in rotate_on_auth_failure before saving; skip counter
  when account already needs_reauth on disk.
- Add rotate_away_without_auth_penalty; use for HTTP 403 streaming handshake
  and handle_forbidden_rotation on connector/credentials.
- Add selector and credential tests; wire executor/conftest mocks.

Also includes local changes to codex quota notifications, continuation-related
tests, acp stale kill timer, tool calls regression, and codex CLI tests.

Made-with: Cursor

Author: Mateusz

src/connectors/_openai_codex_connector.py

@@ -115,6 +115,7 @@ def has_static_credentials(self) -> bool:
 
     # Supported Codex models - sourced from official Codex CLI models.json
     SUPPORTED_CODEX_MODELS: tuple[str, ...] = (
+        "gpt-5.5",
         "gpt-5.4",
         "gpt-5.4-mini",
         "gpt-5.3-codex",
@@ -140,6 +141,7 @@ def has_static_credentials(self) -> bool:
     DEFAULT_REASONING_EFFORT: str = "medium"
     # All current models support xhigh reasoning effort
     XHIGH_SUPPORTED_MODELS: tuple[str, ...] = (
+        "gpt-5.5",
         "gpt-5.4",
         "gpt-5.4-mini",
         "gpt-5.3-codex",
@@ -1809,6 +1811,39 @@ async def _handle_auth_failure_rotation(
             return True
         return False
 
+    async def _handle_forbidden_rotation(
+        self,
+        *,
+        session_id: str | None = None,
+    ) -> bool:
+        """Rotate managed OAuth accounts after HTTP 403 without auth-failure penalties."""
+        rotate_method = getattr(
+            self._credential_manager,
+            "handle_forbidden_rotation",
+            None,
+        )
+        if not callable(rotate_method):
+            return False
+
+        try:
+            result = rotate_method(session_id=session_id)
+            rotated = await result if inspect.isawaitable(result) else bool(result)
+        except Exception as exc:
+            if logger.isEnabledFor(logging.WARNING):
+                logger.warning(
+                    "Failed to rotate managed OAuth account after HTTP 403: %s",
+                    exc,
+                    exc_info=True,
+                )
+            return False
+
+        if rotated:
+            token = self._credential_manager.get_access_token()
+            if token:
+                self.api_key = token
+            return True
+        return False
+
     # -----------------------------
     # Health Tracking API (stale token handling pattern)
     # -----------------------------

src/connectors/openai_codex/codex_quota_notifications.py

@@ -271,6 +271,11 @@ async def maybe_notify_codex_quota_remaining_low(
     ``latch_keys`` uses tuples ``(managed_account_id, limit_kind, threshold)``.
     A key is cleared when remaining rises back to ``>=`` that threshold so alerts
     can re-fire after recovery.
+
+    When several thresholds are violated in one evaluation (e.g. remaining 8%
+    with alerts at 25% and 10%), at most one desktop notification is sent for
+    that limit kind, using the tightest newly crossed threshold; all violated
+    thresholds are latched so they do not each emit a separate bubble.
     """
     if notification_service is None or not notification_service.is_enabled:
         return
@@ -293,31 +298,43 @@ async def maybe_notify_codex_quota_remaining_low(
         if not math.isfinite(float(remaining)):
             continue
         rem = float(remaining)
+
         for thr in thresholds:
             key = (managed_account_id, kind, thr)
             if rem >= thr:
                 latch_keys.discard(key)
-                continue
-            if key in latch_keys:
-                continue
-            message = build_codex_low_remaining_notification_message(
-                email=email,
-                managed_account_id=managed_account_id,
-                chatgpt_account_id=chatgpt_account_id,
-                limit_kind=kind,
-                threshold_percent=thr,
-                remaining_percent=rem,
+
+        violated = [thr for thr in thresholds if rem < thr]
+        if not violated:
+            continue
+
+        newly_violated = [
+            thr for thr in violated if (managed_account_id, kind, thr) not in latch_keys
+        ]
+        if not newly_violated:
+            continue
+
+        worst_new = min(newly_violated)
+        message = build_codex_low_remaining_notification_message(
+            email=email,
+            managed_account_id=managed_account_id,
+            chatgpt_account_id=chatgpt_account_id,
+            limit_kind=kind,
+            threshold_percent=worst_new,
+            remaining_percent=rem,
+        )
+        try:
+            await notification_service.send_notification(
+                title=_CODEX_QUOTA_LOW_REMAINING_TITLE,
+                message=message,
             )
-            try:
-                await notification_service.send_notification(
-                    title=_CODEX_QUOTA_LOW_REMAINING_TITLE,
-                    message=message,
-                )
-            except Exception as exc:
-                logger.warning(
-                    "Codex low remaining quota notification failed: %s",
-                    exc,
-                    exc_info=True,
-                )
-                continue
-            latch_keys.add(key)
+        except Exception as exc:
+            logger.warning(
+                "Codex low remaining quota notification failed: %s",
+                exc,
+                exc_info=True,
+            )
+            continue
+
+        for thr in violated:
+            latch_keys.add((managed_account_id, kind, thr))

src/connectors/openai_codex/credentials.py

@@ -463,6 +463,19 @@ async def _managed_has_accounts(self) -> bool:
             return False
         return await self._managed_storage.has_configured_accounts()
 
+    async def _codex_legacy_auth_file_fallback_allowed(self) -> bool:
+        """Return True if loading Codex desktop ``auth.json`` is allowed as fallback.
+
+        When managed OAuth is enabled and at least one account file exists under the
+        managed storage path, legacy ``auth.json`` must not be used, even if no
+        managed account is currently selectable (e.g. all quotas exhausted).
+        """
+        if not self._managed_enabled():
+            return True
+        if await self._managed_has_accounts():
+            return False
+        return bool(self._managed_config.allow_legacy_fallback)
+
     def _managed_account_to_credentials(
         self,
         account: ManagedOAuthAccount,
@@ -530,6 +543,18 @@ async def _load_auth(self, force_reload: bool = False) -> bool:
         if await self._load_managed_auth(force_reload=force_reload):
             return True
 
+        if not await self._codex_legacy_auth_file_fallback_allowed():
+            if self._managed_enabled() and await self._managed_has_accounts():
+                logger.warning(
+                    "OpenAI Codex: managed OAuth account files exist but no managed "
+                    "account is currently selectable; not using legacy auth.json "
+                    "from the Codex app."
+                )
+            self._active_source = "none"
+            self._managed_current_account = None
+            self._auth_credentials = None
+            return False
+
         explicit_legacy_override = (
             self._auth_path is not None or self._oauth_dir_override is not None
         )
@@ -872,20 +897,63 @@ async def refresh_access_token(self) -> bool:
                 "Attempting to refresh OpenAI Codex access token after authentication failure."
             )
             if await self._load_managed_auth(force_reload=True):
-                if await self._refresh_managed_access_token():
+                success, err = await self._refresh_managed_access_token()
+                if success:
                     return True
-                return await self._rotate_managed_on_auth_failure()
+                if err is None:
+                    return False
+                if err.from_transient_network:
+                    return False
+                if err.needs_reauth:
+                    await self._managed_selector.reload_accounts()
+                    nxt = await self._managed_selector.get_next_account(
+                        wait_for_rate_limit_recovery=False,
+                        ignore_session_affinity=True,
+                    )
+                    if nxt is None:
+                        return False
+                    self._managed_selector.update_account(nxt)
+                    self._managed_current_account = nxt
+                    self._auth_credentials = self._managed_account_to_credentials(nxt)
+                    self._active_source = "managed"
+                    return bool(nxt.access_token)
+                rotated = await self._managed_selector.rotate_away_without_auth_penalty(
+                    session_id=None,
+                )
+                if rotated is None:
+                    return False
+                self._managed_current_account = rotated
+                self._auth_credentials = self._managed_account_to_credentials(rotated)
+                self._active_source = "managed"
+                return bool(rotated.access_token)
+            if not await self._codex_legacy_auth_file_fallback_allowed():
+                if self._managed_enabled() and await self._managed_has_accounts():
+                    logger.warning(
+                        "OpenAI Codex: managed OAuth account files exist but no "
+                        "managed account is currently selectable; not refreshing "
+                        "from legacy auth.json."
+                    )
+                return False
             return await self._refresh_legacy_access_token()
 
-    async def _refresh_managed_access_token(self) -> bool:
+    async def _refresh_managed_access_token(
+        self,
+    ) -> tuple[bool, ManagedOAuthRefreshError | None]:
+        """Force-refresh the managed OAuth access token for the current (or next) account.
+
+        Returns:
+            ``(True, None)`` on success.
+            ``(False, None)`` when no managed account is available to refresh.
+            ``(False, exc)`` when refresh fails.
+        """
         account = self._managed_selector.get_current_account()
         if account is None:
             # Refresh path should also avoid blocking on rate-limit recovery sleeps.
             account = await self._managed_selector.get_next_account(
                 wait_for_rate_limit_recovery=False,
             )
         if account is None:
-            return False
+            return False, None
 
         try:
             updated = await self._managed_refresh.force_refresh(account)
@@ -903,22 +971,13 @@ async def _refresh_managed_access_token(self) -> bool:
                     exc,
                     exc_info=True,
                 )
-            return False
+            return False, exc
 
         self._managed_selector.update_account(updated)
         self._managed_current_account = updated
         self._auth_credentials = self._managed_account_to_credentials(updated)
         self._active_source = "managed"
-        return True
-
-    async def _rotate_managed_on_auth_failure(self) -> bool:
-        rotated = await self._managed_selector.rotate_on_auth_failure()
-        if rotated is None:
-            return False
-        self._managed_current_account = rotated
-        self._auth_credentials = self._managed_account_to_credentials(rotated)
-        self._active_source = "managed"
-        return True
+        return True, None
 
     async def _refresh_legacy_access_token(self) -> bool:
         await self._load_legacy_auth(force_reload=True)
@@ -1378,6 +1437,14 @@ async def handle_auth_failure(
         """Rotate away from currently failing managed account on auth denial."""
         async with self._token_refresh_lock:
             if not await self._load_managed_auth(force_reload=True):
+                if not await self._codex_legacy_auth_file_fallback_allowed():
+                    if self._managed_enabled() and await self._managed_has_accounts():
+                        logger.warning(
+                            "OpenAI Codex: managed OAuth account files exist but no "
+                            "managed account is currently selectable; not handling "
+                            "auth failure via legacy auth.json."
+                        )
+                    return False
                 return await self._refresh_legacy_access_token()
             rotated = await self._managed_selector.rotate_on_auth_failure(
                 session_id=session_id
@@ -1389,6 +1456,25 @@ async def handle_auth_failure(
             self._active_source = "managed"
             return True
 
+    async def handle_forbidden_rotation(
+        self,
+        *,
+        session_id: str | None = None,
+    ) -> bool:
+        """Try another managed account after HTTP 403 without auth-failure penalties."""
+        async with self._token_refresh_lock:
+            if not await self._load_managed_auth(force_reload=True):
+                return False
+            rotated = await self._managed_selector.rotate_away_without_auth_penalty(
+                session_id=session_id,
+            )
+            if rotated is None:
+                return False
+            self._managed_current_account = rotated
+            self._auth_credentials = self._managed_account_to_credentials(rotated)
+            self._active_source = "managed"
+            return True
+
     async def mark_account_used(self) -> None:
         """Mark currently selected managed account as used."""
         if self._active_source != "managed":

src/connectors/openai_codex/executor.py

@@ -609,7 +609,7 @@ async def _streaming_iterator() -> AsyncIterator[ProcessedResponse]:
                     except (HTTPException, LLMProxyError) as exc:
                         status_code, detail = _codex_initiate_streaming_error_view(exc)
                         if status_code == 403 and attempts_used < max_retries:
-                            rotated = await self._handle_auth_failure_rotation(
+                            rotated = await self._handle_forbidden_rotation(
                                 session_id=context.session_id
                             )
                             if rotated:
@@ -2021,6 +2021,34 @@ async def _handle_auth_failure_rotation(self, *, session_id: str | None) -> bool
             return True
         return False
 
+    async def _handle_forbidden_rotation(self, *, session_id: str | None) -> bool:
+        rotate_method = getattr(
+            self._base_connector,
+            "_handle_forbidden_rotation",
+            None,
+        )
+        if callable(rotate_method):
+            result = rotate_method(session_id=session_id)
+            rotated = await result if inspect.isawaitable(result) else bool(result)
+            return bool(rotated)
+
+        fallback_rotate = getattr(
+            self._credential_manager,
+            "handle_forbidden_rotation",
+            None,
+        )
+        if not callable(fallback_rotate):
+            return False
+
+        result = fallback_rotate(session_id=session_id)
+        rotated = await result if inspect.isawaitable(result) else bool(result)
+        if rotated:
+            new_token = self._credential_manager.get_access_token()
+            if new_token and hasattr(self._base_connector, "api_key"):
+                self._base_connector.api_key = new_token
+            return True
+        return False
+
     async def _mark_account_used(self) -> None:
         mark_used_method = getattr(self._credential_manager, "mark_account_used", None)
         if not callable(mark_used_method):