feat(openai-codex): desktop alerts for low remaining quota from x-codex headers

Add configurable thresholds (default 25%% and 10%% remaining) under
managed_oauth, env overrides, and evaluation on each response with
x-codex-* headers. Reuse INotificationService with latch/re-arm semantics
per account and primary vs weekly window.

Made-with: Cursor

Author: Mateusz

config/config.example.yaml

@@ -483,6 +483,12 @@ backends:
           # Max sleep iterations when every account is rate-limited (each slice up to
           # max_rate_limit_wait_seconds).
           max_rate_limit_idle_polls: 48
+          # Desktop alerts when remaining Codex quota (100 - x-codex-*-used-percent) is low.
+          quota_remaining_alerts_enabled: true
+          # Notify when remaining quota drops strictly below each percent (defaults: 25 then 10).
+          quota_remaining_alert_thresholds_percent: [25, 10]
+          # Optional env overrides: OPENAI_CODEX_QUOTA_REMAINING_ALERTS_ENABLED,
+          # OPENAI_CODEX_QUOTA_REMAINING_THRESHOLDS (JSON array e.g. [25,10] or comma-separated).
     # For Responses API clients (Profile B), use:
     #   default_capabilities:
     #     protocol: openai-responses
@@ -522,6 +528,8 @@ backends:
           max_rate_limit_wait_seconds: 300
           rate_limit_local_cooldown_cap_seconds: 1800
           max_rate_limit_idle_polls: 48
+          quota_remaining_alerts_enabled: true
+          quota_remaining_alert_thresholds_percent: [25, 10]
 
 # Model-specific defaults
 model_defaults:

src/connectors/_openai_codex_connector.py

@@ -1682,14 +1682,38 @@ def update_quota_headers(self, headers: Mapping[str, Any]) -> None:
         record_raw = getattr(
             self._credential_manager, "record_codex_quota_headers", None
         )
-        if not callable(record_raw):
-            return
-        record = cast(Callable[..., Awaitable[None]], record_raw)
+        eval_raw = getattr(
+            self._credential_manager,
+            "evaluate_codex_remaining_quota_notifications",
+            None,
+        )
+        has_codex_quota = any(str(k).lower().startswith("x-codex-") for k in headers)
         try:
             loop = asyncio.get_running_loop()
         except RuntimeError:
             return
 
+        if has_codex_quota and callable(eval_raw):
+            evaluate = cast(Callable[..., Awaitable[None]], eval_raw)
+
+            async def _remaining_alerts() -> None:
+                try:
+                    await evaluate(headers, managed_oauth_account_id=None)
+                except Exception as exc:
+                    logger.warning(
+                        "OpenAI Codex evaluate_codex_remaining_quota_notifications failed: %s",
+                        exc,
+                        exc_info=True,
+                    )
+
+            alert_task = loop.create_task(_remaining_alerts())
+            self._codex_quota_persist_tasks.add(alert_task)
+            alert_task.add_done_callback(self._codex_quota_persist_tasks.discard)
+
+        if not callable(record_raw):
+            return
+        record = cast(Callable[..., Awaitable[None]], record_raw)
+
         async def _persist() -> None:
             try:
                 await record(headers, force=False)

src/connectors/openai_codex/codex_quota_notifications.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import math
 import time
 from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any
@@ -20,6 +21,7 @@
 logger = logging.getLogger(__name__)
 
 _CODEX_QUOTA_NOTIFICATION_TITLE = "OpenAI Codex: Quota reached"
+_CODEX_QUOTA_LOW_REMAINING_TITLE = "OpenAI Codex: Low quota remaining"
 
 # Dedupe key component when every managed account is simultaneously unavailable:
 # one desktop alert per quota window, not one per account that hits 429 last.
@@ -161,3 +163,161 @@ async def maybe_notify_codex_quota_reached(
         return
 
     dedupe_keys.add(key)
+
+
+def codex_limit_kind_display(limit_kind: str) -> str:
+    """Human-readable Codex rolling window label for remaining-quota alerts."""
+    if limit_kind == "secondary":
+        return "weekly rolling window"
+    return "5 hour rolling window"
+
+
+def parse_codex_used_percent_from_headers(
+    headers: Mapping[str, Any],
+    *,
+    header_key: str,
+) -> float | None:
+    """Return used-percent (0-100) from headers, or None if missing/invalid."""
+    target = header_key.lower()
+    for raw_k, raw_v in headers.items():
+        if str(raw_k).lower() != target:
+            continue
+        try:
+            used = float(str(raw_v).strip())
+        except (TypeError, ValueError):
+            return None
+        if not math.isfinite(used):
+            return None
+        return used
+    return None
+
+
+def codex_remaining_percent_from_used(used_percent: float) -> float:
+    """Remaining quota percent from upstream ``*-used-percent`` (clamped)."""
+    remaining = 100.0 - float(used_percent)
+    if remaining < 0.0:
+        return 0.0
+    if remaining > 100.0:
+        return 100.0
+    return remaining
+
+
+def collect_codex_remaining_pairs(
+    headers: Mapping[str, Any]
+) -> list[tuple[str, float]]:
+    """Build (limit_kind, remaining_percent) pairs from Codex quota headers."""
+    pairs: list[tuple[str, float]] = []
+    used_primary = parse_codex_used_percent_from_headers(
+        headers,
+        header_key="x-codex-primary-used-percent",
+    )
+    if used_primary is not None:
+        pairs.append(
+            ("primary", codex_remaining_percent_from_used(used_primary)),
+        )
+    used_secondary = parse_codex_used_percent_from_headers(
+        headers,
+        header_key="x-codex-secondary-used-percent",
+    )
+    if used_secondary is not None:
+        pairs.append(
+            ("secondary", codex_remaining_percent_from_used(used_secondary)),
+        )
+    return pairs
+
+
+def build_codex_low_remaining_notification_message(
+    *,
+    email: str | None,
+    managed_account_id: str | None = None,
+    chatgpt_account_id: str | None = None,
+    limit_kind: str,
+    threshold_percent: float,
+    remaining_percent: float,
+) -> str:
+    """Body text for a Codex low remaining-quota desktop notification."""
+    account = _format_account_label(
+        email,
+        managed_account_id=managed_account_id,
+        chatgpt_account_id=chatgpt_account_id,
+    )
+    window = codex_limit_kind_display(limit_kind)
+    rem_rounded = round(float(remaining_percent), 1)
+    thr = float(threshold_percent)
+    thr_display = int(thr) if thr == int(thr) else thr
+    return (
+        f"Codex quota running low. Account: {account}, limit: {window}. "
+        f"Remaining is below your {thr_display}% threshold "
+        f"(about {rem_rounded}% of this window still available)."
+    )
+
+
+def _normalize_threshold_latch_key(threshold_percent: float) -> float:
+    return round(float(threshold_percent), 6)
+
+
+async def maybe_notify_codex_quota_remaining_low(
+    notification_service: INotificationService | None,
+    latch_keys: set[tuple[str, str, float]],
+    *,
+    managed_account_id: str,
+    email: str | None,
+    chatgpt_account_id: str | None = None,
+    threshold_percents: list[float],
+    remaining_by_limit: list[tuple[str, float]],
+) -> None:
+    """Notify when remaining quota drops strictly below configured thresholds.
+
+    ``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.
+    """
+    if notification_service is None or not notification_service.is_enabled:
+        return
+    if not threshold_percents or not remaining_by_limit:
+        return
+
+    thresholds = sorted(
+        {_normalize_threshold_latch_key(t) for t in threshold_percents if t > 0},
+        reverse=True,
+    )
+    if not thresholds:
+        return
+
+    for limit_kind, remaining in remaining_by_limit:
+        if not isinstance(limit_kind, str) or not limit_kind.strip():
+            continue
+        kind = limit_kind.strip().lower()
+        if kind not in ("primary", "secondary"):
+            continue
+        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,
+            )
+            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)

src/connectors/openai_codex/credentials.py

@@ -27,7 +27,9 @@
 from watchdog.observers import Observer
 
 from src.connectors.openai_codex.codex_quota_notifications import (
+    collect_codex_remaining_pairs,
     maybe_notify_codex_quota_reached,
+    maybe_notify_codex_quota_remaining_low,
 )
 from src.connectors.openai_codex.codex_rate_limit_logging import (
     emit_openai_codex_managed_oauth_rate_limit,
@@ -385,6 +387,7 @@ def __init__(
         self._http_client = http_client
         self._notification_service = notification_service
         self._codex_quota_notification_dedupe: set[tuple[str, str, str]] = set()
+        self._codex_remaining_quota_latch: set[tuple[str, str, float]] = set()
         self._auth_path: Path | None = None
         self._auth_credentials: dict[str, Any] | None = None
         self._last_modified: float = 0.0
@@ -1113,6 +1116,75 @@ async def record_codex_quota_headers(
             ):
                 self._managed_current_account = updated
 
+    async def evaluate_codex_remaining_quota_notifications(
+        self,
+        headers: Mapping[str, Any],
+        *,
+        managed_oauth_account_id: str | None = None,
+    ) -> None:
+        """Desktop alerts when x-codex-*-used-percent implies low remaining quota."""
+        cfg = self._managed_config
+        if not cfg.quota_remaining_alerts_enabled:
+            return
+        thresholds = list(cfg.quota_remaining_alert_thresholds_percent)
+        if not thresholds:
+            return
+        remaining_pairs = collect_codex_remaining_pairs(headers)
+        if not remaining_pairs:
+            return
+
+        async with self._codex_telemetry_lock:
+            account_id = "legacy-openai-codex"
+            email: str | None = None
+            chatgpt_account_id: str | None = None
+            resolved_managed = False
+            if self._managed_enabled():
+                aid = (
+                    str(managed_oauth_account_id).strip()
+                    if isinstance(managed_oauth_account_id, str)
+                    and managed_oauth_account_id.strip()
+                    else ""
+                )
+                current = (
+                    self._managed_selector.get_account_by_id(aid)
+                    if aid
+                    else self._managed_selector.get_current_account()
+                )
+                if current is not None:
+                    account_id = current.account_id
+                    email = current.email
+                    chatgpt_account_id = current.chatgpt_account_id
+                    resolved_managed = True
+
+            if not resolved_managed and isinstance(self._auth_credentials, Mapping):
+                managed = self._auth_credentials.get("managed_oauth")
+                if isinstance(managed, Mapping):
+                    mid = managed.get("account_id")
+                    if isinstance(mid, str) and mid.strip():
+                        account_id = mid.strip()
+                    cg = managed.get("chatgpt_account_id")
+                    if isinstance(cg, str) and cg.strip():
+                        chatgpt_account_id = cg.strip()
+                if account_id == "legacy-openai-codex":
+                    top_aid = self._auth_credentials.get("account_id")
+                    if isinstance(top_aid, str) and top_aid.strip():
+                        account_id = top_aid.strip()
+                user = self._auth_credentials.get("user")
+                if isinstance(user, Mapping):
+                    raw_email = user.get("email")
+                    if isinstance(raw_email, str) and raw_email.strip():
+                        email = raw_email.strip()
+
+        await maybe_notify_codex_quota_remaining_low(
+            self._notification_service,
+            self._codex_remaining_quota_latch,
+            managed_account_id=account_id,
+            email=email,
+            chatgpt_account_id=chatgpt_account_id,
+            threshold_percents=thresholds,
+            remaining_by_limit=remaining_pairs,
+        )
+
     async def handle_rate_limit(
         self,
         retry_after_seconds: float | None,

src/connectors/openai_codex/managed_oauth_models.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import math
 import re
 import time
 from datetime import datetime, timezone
@@ -246,6 +247,28 @@ class ManagedOAuthConfig(BaseModel):
     #: Max idle polls (sleeps) while all accounts are rate-limited before giving up
     #: (only when ``get_next_account`` is invoked with ``wait_for_rate_limit_recovery=True``).
     max_rate_limit_idle_polls: int = 48
+    #: Desktop notifications when remaining Codex quota (100 - used%%) drops below thresholds.
+    quota_remaining_alerts_enabled: bool = True
+    quota_remaining_alert_thresholds_percent: list[float] = Field(
+        default_factory=lambda: [25.0, 10.0],
+        description=(
+            "Alert when remaining quota is strictly below each value (percent of limit left). "
+            "Compared against x-codex-primary-used-percent / x-codex-secondary-used-percent."
+        ),
+    )
+
+    @field_validator("quota_remaining_alert_thresholds_percent", mode="after")
+    @classmethod
+    def _normalize_quota_remaining_thresholds(cls, v: list[float]) -> list[float]:
+        cleaned: list[float] = []
+        for raw in v:
+            if not isinstance(raw, int | float) or not math.isfinite(float(raw)):
+                continue
+            x = float(raw)
+            if 0.0 < x < 100.0:
+                cleaned.append(round(x, 6))
+        uniq = sorted(set(cleaned), reverse=True)
+        return uniq if uniq else [25.0, 10.0]
 
     @classmethod
     def from_mapping(