diff --git a/cli.py b/cli.py index 8a8fdf6f5ff5..6e117057290d 100644 --- a/cli.py +++ b/cli.py @@ -31,6 +31,7 @@ import re import concurrent.futures import base64 +import hashlib import atexit import errno import tempfile @@ -4105,6 +4106,13 @@ def __init__( # Status bar visibility (toggled via /statusbar) self._status_bar_visible = True + # The Codex usage endpoint is network-backed. Status rendering reads + # this cache only; a rate-limited daemon refreshes it in the background. + self._codex_usage_snapshot = None + self._codex_usage_last_attempt = 0.0 + self._codex_usage_refreshing = False + self._codex_usage_lock = threading.Lock() + self._codex_usage_scope = None # When True, the input separator rules and the dynamic status bar are # hidden until the next user input. Set by _recover_after_resize() so a # SIGWINCH cannot stamp a freshly-drawn status bar on top of one that @@ -4527,6 +4535,154 @@ def _format_idle_since(last_finished_at: Optional[float], turn_live: bool) -> st idle = max(0.0, time.time() - last_finished_at) return f"✓ {format_duration_compact(idle)}" + @staticmethod + def _format_codex_session_limit( + snapshot: Any, + *, + now: Optional[datetime] = None, + ) -> Optional[tuple[str, int]]: + """Format the cached Codex session window as remaining quota.""" + for window in getattr(snapshot, "windows", ()) or (): + if str(getattr(window, "label", "") or "").strip().lower() != "session": + continue + try: + remaining = round(100 - float(window.used_percent)) + except (AttributeError, OverflowError, TypeError, ValueError): + return None + remaining = max(0, min(100, remaining)) + label = f"Codex {remaining}%" + reset_at = getattr(window, "reset_at", None) + if isinstance(reset_at, datetime): + current = now + if current is None: + current = datetime.now(reset_at.tzinfo) if reset_at.tzinfo else datetime.now() + try: + seconds = max(0.0, (reset_at - current).total_seconds()) + except (TypeError, ValueError): + seconds = 0.0 + label += f" reset {format_duration_compact(seconds)}" + return label, remaining + return None + + @staticmethod + def _codex_session_limit_style(remaining: Optional[int]) -> str: + """Style Codex session-limit pressure by remaining percentage.""" + if remaining is None: + return "class:status-bar-dim" + if remaining <= 5: + return "class:status-bar-critical" + if remaining <= 15: + return "class:status-bar-bad" + if remaining <= 35: + return "class:status-bar-warn" + return "class:status-bar-good" + + def _current_provider_name(self) -> str: + """Return the active provider, preferring runtime fallback state.""" + agent = getattr(self, "agent", None) + for value in ( + getattr(agent, "provider", None), + getattr(self, "provider", None), + getattr(self, "requested_provider", None), + ): + provider = str(value or "").strip().lower() + if provider: + return provider + return "" + + def _capture_codex_usage_credentials(self) -> Optional[tuple[Any, Any, tuple[str, str]]]: + """Capture a coherent Codex client credential snapshot.""" + agent = getattr(self, "agent", None) + if agent is None: + return None + + provider_before = str(getattr(agent, "provider", "") or "").strip().lower() + client = getattr(agent, "client", None) + provider_after = str(getattr(agent, "provider", "") or "").strip().lower() + if provider_before != "openai-codex" or provider_after != "openai-codex" or client is None: + return None + + # The OpenAI client is replaced as one object after credential rotation + # or provider fallback. Reading from that captured object avoids pairing + # a base URL and API key from different runtime generations. + base_url = str(getattr(client, "base_url", "") or "").strip() + api_key = getattr(client, "api_key", None) + if not base_url_host_matches(base_url, "chatgpt.com") or not api_key: + return None + + key_text = str(api_key) + key_fingerprint = hashlib.sha256(key_text.encode("utf-8")).hexdigest() + scope = (base_url, key_fingerprint) + return base_url, api_key, scope + + def _codex_usage_scope_for_active_credentials(self) -> Optional[tuple[str, str]]: + """Return a non-secret identity for the active Codex credentials.""" + captured = self._capture_codex_usage_credentials() + return captured[2] if captured is not None else None + + def _maybe_refresh_codex_usage_snapshot(self, *, min_interval: float = 300.0) -> None: + """Start one nonblocking Codex usage refresh when the cache is stale.""" + captured = self._capture_codex_usage_credentials() + lock = getattr(self, "_codex_usage_lock", None) + if lock is None: + lock = threading.Lock() + self._codex_usage_lock = lock + if captured is None: + with lock: + self._codex_usage_scope = None + self._codex_usage_snapshot = None + self._codex_usage_last_attempt = 0.0 + return + + now = time.monotonic() + base_url, api_key, refresh_scope = captured + + with lock: + if getattr(self, "_codex_usage_scope", None) != refresh_scope: + self._codex_usage_scope = refresh_scope + self._codex_usage_snapshot = None + self._codex_usage_last_attempt = 0.0 + if getattr(self, "_codex_usage_refreshing", False): + return + last_attempt = getattr(self, "_codex_usage_last_attempt", 0.0) or 0.0 + if last_attempt and (now - last_attempt) < min_interval: + return + self._codex_usage_refreshing = True + self._codex_usage_last_attempt = now + + def _refresh() -> None: + refreshed = None + try: + from agent.account_usage import fetch_account_usage + + refreshed = fetch_account_usage( + "openai-codex", + base_url=base_url, + api_key=api_key, + ) + except Exception: + logger.debug("Codex status-bar usage refresh failed", exc_info=True) + finally: + with lock: + if refreshed is not None and self._codex_usage_scope == refresh_scope: + self._codex_usage_snapshot = refreshed + self._codex_usage_refreshing = False + self._invalidate(min_interval=0.0) + + thread = threading.Thread( + target=_refresh, + name="codex-usage-status-refresh", + daemon=True, + ) + try: + thread.start() + except Exception: + with lock: + self._codex_usage_refreshing = False + if self._codex_usage_scope == refresh_scope: + self._codex_usage_last_attempt = 0.0 + logger.debug("Could not start Codex status-bar refresh thread", exc_info=True) + def _get_status_bar_snapshot(self) -> Dict[str, Any]: # Prefer the agent's model name — it updates on fallback. # self.model reflects the originally configured model and never @@ -4569,8 +4725,18 @@ def _get_status_bar_snapshot(self) -> Dict[str, Any]: "active_background_tasks": 0, "active_background_processes": 0, "active_background_subagents": 0, + "codex_session_limit": None, + "codex_session_remaining_percent": None, } + if self._current_provider_name() == "openai-codex": + self._maybe_refresh_codex_usage_snapshot() + formatted_limit = self._format_codex_session_limit( + getattr(self, "_codex_usage_snapshot", None) + ) + if formatted_limit is not None: + snapshot["codex_session_limit"], snapshot["codex_session_remaining_percent"] = formatted_limit + # Count live /background tasks. The dict entry is removed in the # task thread's finally block, so len() reflects truly-running tasks. # len() on a CPython dict is atomic; safe to read without a lock. @@ -5085,6 +5251,9 @@ def _build_status_bar_text(self, width: Optional[int] = None) -> str: compressions = snapshot.get("compressions", 0) parts = [f"⚕ {snapshot['model_short']}", context_label, percent_label] + codex_limit = snapshot.get("codex_session_limit") + if codex_limit: + parts.append(codex_limit) if compressions: parts.append(f"🗜️ {compressions}") bg_count = snapshot.get("active_background_tasks", 0) @@ -5191,6 +5360,17 @@ def _get_status_bar_fragments(self): ("class:status-bar-dim", " "), (bar_style, percent_label), ] + codex_limit = snapshot.get("codex_session_limit") + if codex_limit: + frags.append(("class:status-bar-dim", " │ ")) + frags.append( + ( + self._codex_session_limit_style( + snapshot.get("codex_session_remaining_percent") + ), + codex_limit, + ) + ) if compressions: frags.append(("class:status-bar-dim", " │ ")) frags.append((self._compression_count_style(compressions), f"🗜️ {compressions}")) diff --git a/tests/cli/test_cli_status_bar.py b/tests/cli/test_cli_status_bar.py index 1899f0dd78e0..9ae3eed145d8 100644 --- a/tests/cli/test_cli_status_bar.py +++ b/tests/cli/test_cli_status_bar.py @@ -1,3 +1,4 @@ +import threading import time from datetime import datetime, timedelta from types import SimpleNamespace @@ -10,12 +11,54 @@ def _make_cli(model: str = "anthropic/claude-sonnet-4-20250514"): cli_obj = HermesCLI.__new__(HermesCLI) cli_obj.model = model + cli_obj.provider = "anthropic" + cli_obj.requested_provider = "anthropic" cli_obj.session_start = datetime.now() - timedelta(minutes=14, seconds=32) cli_obj.conversation_history = [{"role": "user", "content": "hi"}] cli_obj.agent = None + cli_obj._codex_usage_snapshot = None + cli_obj._codex_usage_last_attempt = 0.0 + cli_obj._codex_usage_refreshing = False + cli_obj._codex_usage_lock = threading.Lock() + cli_obj._codex_usage_scope = None return cli_obj +def _codex_usage_snapshot( + used_percent: object, + *, + reset_at: datetime | None = None, + label: str = "Session", +): + return SimpleNamespace( + windows=(SimpleNamespace(label=label, used_percent=used_percent, reset_at=reset_at),) + ) + + +def _attach_codex_runtime( + cli_obj, + *, + api_key: str = "codex-test-token", + base_url: str = "https://chatgpt.com/backend-api/codex", +) -> None: + cli_obj.provider = "openai-codex" + cli_obj.requested_provider = "openai-codex" + cli_obj.agent = SimpleNamespace( + model=cli_obj.model, + provider="openai-codex", + api_key=api_key, + base_url=base_url, + client=SimpleNamespace(api_key=api_key, base_url=base_url), + ) + + +def _seed_codex_usage(cli_obj, used_percent: object, *, reset_at: datetime | None = None) -> None: + _attach_codex_runtime(cli_obj) + cli_obj._codex_usage_snapshot = _codex_usage_snapshot(used_percent, reset_at=reset_at) + cli_obj._codex_usage_scope = cli_obj._codex_usage_scope_for_active_credentials() + cli_obj._codex_usage_last_attempt = time.monotonic() + + def _attach_agent( cli_obj, *, @@ -63,6 +106,364 @@ def test_context_style_thresholds(self): assert cli_obj._status_bar_context_style(81) == "class:status-bar-bad" assert cli_obj._status_bar_context_style(95) == "class:status-bar-critical" + def test_codex_session_limit_formats_remaining_percent(self): + result = HermesCLI._format_codex_session_limit(_codex_usage_snapshot(23.4)) + + assert result == ("Codex 77%", 77) + + def test_codex_session_limit_formats_reset_countdown_from_injected_time(self): + now = datetime(2026, 7, 15, 12, 0, 0) + reset_at = now + timedelta(hours=2, minutes=10, seconds=30) + + result = HermesCLI._format_codex_session_limit( + _codex_usage_snapshot(23.4, reset_at=reset_at), + now=now, + ) + + assert result == ("Codex 77% reset 2h 10m", 77) + + def test_codex_session_limit_rejects_missing_or_invalid_session_data(self): + assert HermesCLI._format_codex_session_limit(None) is None + assert HermesCLI._format_codex_session_limit(_codex_usage_snapshot(20, label="Weekly")) is None + assert HermesCLI._format_codex_session_limit(_codex_usage_snapshot("unknown")) is None + assert HermesCLI._format_codex_session_limit(_codex_usage_snapshot(float("nan"))) is None + assert HermesCLI._format_codex_session_limit(_codex_usage_snapshot(float("inf"))) is None + assert HermesCLI._format_codex_session_limit(_codex_usage_snapshot(float("-inf"))) is None + + def test_codex_session_limit_clamps_remaining_percent(self): + assert HermesCLI._format_codex_session_limit(_codex_usage_snapshot(-10)) == ("Codex 100%", 100) + assert HermesCLI._format_codex_session_limit(_codex_usage_snapshot(125)) == ("Codex 0%", 0) + + def test_codex_session_limit_style_thresholds(self): + cli_obj = _make_cli() + + assert cli_obj._codex_session_limit_style(None) == "class:status-bar-dim" + assert cli_obj._codex_session_limit_style(40) == "class:status-bar-good" + assert cli_obj._codex_session_limit_style(35) == "class:status-bar-warn" + assert cli_obj._codex_session_limit_style(15) == "class:status-bar-bad" + assert cli_obj._codex_session_limit_style(5) == "class:status-bar-critical" + + def test_snapshot_includes_cached_codex_limit_for_active_codex_provider(self): + cli_obj = _make_cli("openai/gpt-5.4") + _seed_codex_usage(cli_obj, 12) + + snapshot = cli_obj._get_status_bar_snapshot() + + assert snapshot["codex_session_limit"] == "Codex 88%" + assert snapshot["codex_session_remaining_percent"] == 88 + + def test_snapshot_hides_cached_codex_limit_after_provider_switch(self): + cli_obj = _make_cli() + cli_obj._codex_usage_snapshot = _codex_usage_snapshot(12) + cli_obj._codex_usage_last_attempt = time.monotonic() + + snapshot = cli_obj._get_status_bar_snapshot() + + assert snapshot["codex_session_limit"] is None + assert snapshot["codex_session_remaining_percent"] is None + + def test_active_agent_provider_overrides_stale_cli_provider(self): + cli_obj = _make_cli("openai/gpt-5.4") + _seed_codex_usage(cli_obj, 12) + cli_obj.agent = SimpleNamespace(model="anthropic/claude-opus-4.6", provider="anthropic") + + snapshot = cli_obj._get_status_bar_snapshot() + + assert snapshot["codex_session_limit"] is None + + def test_codex_limit_shows_only_in_wide_plain_text_status(self): + cli_obj = _make_cli("openai/gpt-5.4") + _seed_codex_usage(cli_obj, 66) + + assert "Codex 34%" in cli_obj._build_status_bar_text(width=120) + assert "Codex" not in cli_obj._build_status_bar_text(width=75) + assert "Codex" not in cli_obj._build_status_bar_text(width=51) + + def test_wide_fragments_style_cached_codex_limit(self): + cli_obj = _make_cli("openai/gpt-5.4") + _seed_codex_usage(cli_obj, 94) + cli_obj._status_bar_visible = True + cli_obj._get_tui_terminal_width = lambda: 120 + + fragments = cli_obj._get_status_bar_fragments() + + assert ("class:status-bar-bad", "Codex 6%") in fragments + + def test_codex_refresh_is_skipped_for_other_providers(self): + cli_obj = _make_cli() + + with patch("cli.threading.Thread") as thread_cls: + cli_obj._maybe_refresh_codex_usage_snapshot() + + thread_cls.assert_not_called() + + def test_codex_refresh_rejects_mixed_provider_or_non_codex_client(self): + cli_obj = _make_cli("openai/gpt-5.4") + cli_obj.provider = "openai-codex" + + class SwitchingProviderAgent: + model = "openai/gpt-5.4" + client = SimpleNamespace( + api_key="anthropic-token", + base_url="https://api.anthropic.com", + ) + + def __init__(self): + self._reads = 0 + + @property + def provider(self): + self._reads += 1 + return "openai-codex" if self._reads == 1 else "anthropic" + + cli_obj.agent = SwitchingProviderAgent() + cli_obj._codex_usage_snapshot = _codex_usage_snapshot(20) + + with patch("cli.threading.Thread") as thread_cls: + cli_obj._maybe_refresh_codex_usage_snapshot() + + thread_cls.assert_not_called() + assert cli_obj._codex_usage_snapshot is None + assert cli_obj._codex_usage_scope is None + + def test_codex_refresh_is_skipped_while_cache_interval_is_fresh(self): + cli_obj = _make_cli("openai/gpt-5.4") + _attach_codex_runtime(cli_obj) + cli_obj._codex_usage_scope = cli_obj._codex_usage_scope_for_active_credentials() + cli_obj._codex_usage_last_attempt = 900.0 + + with patch("cli.time.monotonic", return_value=1000.0), patch("cli.threading.Thread") as thread_cls: + cli_obj._maybe_refresh_codex_usage_snapshot(min_interval=300.0) + + thread_cls.assert_not_called() + + def test_codex_refresh_is_single_flight(self): + cli_obj = _make_cli("openai/gpt-5.4") + _attach_codex_runtime(cli_obj) + deferred_targets = [] + thread_options = [] + + class DeferredThread: + def __init__(self, *, target, **kwargs): + deferred_targets.append(target) + thread_options.append(kwargs) + + def start(self): + return None + + with patch("cli.time.monotonic", return_value=1000.0), patch( + "cli.threading.Thread", DeferredThread + ): + cli_obj._maybe_refresh_codex_usage_snapshot() + cli_obj._maybe_refresh_codex_usage_snapshot() + + assert len(deferred_targets) == 1 + assert thread_options == [{"name": "codex-usage-status-refresh", "daemon": True}] + assert cli_obj._codex_usage_refreshing is True + + def test_codex_refresh_discards_in_flight_result_after_credential_failover(self): + cli_obj = _make_cli("openai/gpt-5.4") + cli_obj.provider = "openai-codex" + cli_obj.agent = SimpleNamespace( + model="openai/gpt-5.4", + provider="openai-codex", + api_key="pooled-token-a", + base_url="https://chatgpt.com/backend-api/codex", + client=SimpleNamespace( + api_key="pooled-token-a", + base_url="https://chatgpt.com/backend-api/codex", + ), + ) + cli_obj._invalidate = MagicMock() + deferred_targets = [] + account_a_snapshot = _codex_usage_snapshot(80) + + class DeferredThread: + def __init__(self, *, target, **_kwargs): + deferred_targets.append(target) + + def start(self): + return None + + with patch("cli.time.monotonic", return_value=1000.0), patch( + "cli.threading.Thread", DeferredThread + ), patch("agent.account_usage.fetch_account_usage", return_value=account_a_snapshot): + cli_obj._maybe_refresh_codex_usage_snapshot() + cli_obj.agent.api_key = "pooled-token-b" + cli_obj.agent.client = SimpleNamespace( + api_key="pooled-token-b", + base_url="https://chatgpt.com/backend-api/codex", + ) + cli_obj._maybe_refresh_codex_usage_snapshot() + deferred_targets[0]() + + assert cli_obj._codex_usage_snapshot is None + assert cli_obj._codex_usage_refreshing is False + + def test_codex_credential_failover_clears_cache_and_bypasses_rate_limit(self): + cli_obj = _make_cli("openai/gpt-5.4") + cli_obj.provider = "openai-codex" + cli_obj.agent = SimpleNamespace( + model="openai/gpt-5.4", + provider="openai-codex", + api_key="pooled-token-a", + base_url="https://chatgpt.com/backend-api/codex", + client=SimpleNamespace( + api_key="pooled-token-a", + base_url="https://chatgpt.com/backend-api/codex", + ), + ) + cli_obj._codex_usage_snapshot = _codex_usage_snapshot(80) + cli_obj._codex_usage_scope = cli_obj._codex_usage_scope_for_active_credentials() + cli_obj._codex_usage_last_attempt = 900.0 + cli_obj._invalidate = MagicMock() + account_b_snapshot = _codex_usage_snapshot(20) + + class ImmediateThread: + def __init__(self, *, target, **_kwargs): + self.target = target + + def start(self): + self.target() + + cli_obj.agent.api_key = "pooled-token-b" + cli_obj.agent.client = SimpleNamespace( + api_key="pooled-token-b", + base_url="https://chatgpt.com/backend-api/codex", + ) + with patch("cli.time.monotonic", return_value=1000.0), patch( + "cli.threading.Thread", ImmediateThread + ), patch("agent.account_usage.fetch_account_usage", return_value=account_b_snapshot) as fetch: + cli_obj._maybe_refresh_codex_usage_snapshot() + + fetch.assert_called_once() + assert cli_obj._codex_usage_snapshot is account_b_snapshot + assert cli_obj._codex_usage_last_attempt == 1000.0 + + def test_codex_refresh_uses_same_credential_capture_for_scope_and_fetch(self): + cli_obj = _make_cli("openai/gpt-5.4") + cli_obj.provider = "openai-codex" + cli_obj._invalidate = MagicMock() + account_a_snapshot = _codex_usage_snapshot(30) + codex_url = "https://chatgpt.com/backend-api/codex" + + class RotatingAgent: + model = "openai/gpt-5.4" + provider = "openai-codex" + + def __init__(self): + self.api_key = "account-a-token" + self.client = SimpleNamespace(api_key="account-a-token", base_url=codex_url) + + @property + def base_url(self): + # Reproduce rotation between separate live-field reads. The + # already-captured client object remains an immutable A snapshot. + self.api_key = "account-b-token" + return codex_url + + cli_obj.agent = RotatingAgent() + + class ImmediateThread: + def __init__(self, *, target, **_kwargs): + self.target = target + + def start(self): + self.target() + + with patch("cli.time.monotonic", return_value=1000.0), patch( + "cli.threading.Thread", ImmediateThread + ), patch("agent.account_usage.fetch_account_usage", return_value=account_a_snapshot) as fetch: + cli_obj._maybe_refresh_codex_usage_snapshot() + + fetch.assert_called_once_with( + "openai-codex", + base_url=codex_url, + api_key="account-a-token", + ) + assert cli_obj._codex_usage_snapshot is account_a_snapshot + + def test_codex_refresh_uses_active_agent_credentials_and_repaints(self): + cli_obj = _make_cli("openai/gpt-5.4") + cli_obj.provider = "openai-codex" + cli_obj.api_key = "stale-cli-token" + cli_obj.base_url = "https://stale.example" + cli_obj.agent = SimpleNamespace( + model="openai/gpt-5.4", + provider="openai-codex", + api_key="active-agent-token", + base_url="https://chatgpt.com/backend-api/codex", + client=SimpleNamespace( + api_key="active-agent-token", + base_url="https://chatgpt.com/backend-api/codex", + ), + ) + cli_obj._invalidate = MagicMock() + refreshed = _codex_usage_snapshot(42) + + class ImmediateThread: + def __init__(self, *, target, **_kwargs): + self.target = target + + def start(self): + self.target() + + with patch("cli.time.monotonic", return_value=1000.0), patch( + "cli.threading.Thread", ImmediateThread + ), patch("agent.account_usage.fetch_account_usage", return_value=refreshed) as fetch: + cli_obj._maybe_refresh_codex_usage_snapshot() + + fetch.assert_called_once_with( + "openai-codex", + base_url="https://chatgpt.com/backend-api/codex", + api_key="active-agent-token", + ) + assert cli_obj._codex_usage_snapshot is refreshed + assert cli_obj._codex_usage_refreshing is False + cli_obj._invalidate.assert_called_once_with(min_interval=0.0) + + def test_codex_refresh_failure_keeps_stale_cache_and_is_rate_limited(self): + cli_obj = _make_cli("openai/gpt-5.4") + _attach_codex_runtime(cli_obj) + stale = _codex_usage_snapshot(73) + cli_obj._codex_usage_snapshot = stale + cli_obj._codex_usage_scope = cli_obj._codex_usage_scope_for_active_credentials() + cli_obj._invalidate = MagicMock() + starts = [] + + class ImmediateThread: + def __init__(self, *, target, **_kwargs): + self.target = target + + def start(self): + starts.append(1) + self.target() + + with patch("cli.threading.Thread", ImmediateThread), patch( + "agent.account_usage.fetch_account_usage", side_effect=RuntimeError("temporary failure") + ): + with patch("cli.time.monotonic", return_value=1000.0): + cli_obj._maybe_refresh_codex_usage_snapshot() + with patch("cli.time.monotonic", return_value=1100.0): + cli_obj._maybe_refresh_codex_usage_snapshot() + + assert len(starts) == 1 + assert cli_obj._codex_usage_snapshot is stale + assert cli_obj._codex_usage_refreshing is False + + def test_codex_refresh_thread_start_failure_allows_immediate_retry(self): + cli_obj = _make_cli("openai/gpt-5.4") + _attach_codex_runtime(cli_obj) + + with patch("cli.time.monotonic", return_value=1000.0), patch( + "cli.threading.Thread" + ) as thread_cls: + thread_cls.return_value.start.side_effect = RuntimeError("thread unavailable") + cli_obj._maybe_refresh_codex_usage_snapshot() + + assert cli_obj._codex_usage_refreshing is False + assert cli_obj._codex_usage_last_attempt == 0.0 + def test_build_status_bar_text_for_wide_terminal(self): cli_obj = _attach_agent( _make_cli(),