From 0221367c20ffb16a7f4b9b0f19adeeed7e1055c0 Mon Sep 17 00:00:00 2001 From: Jay German Date: Sun, 24 May 2026 21:24:54 -0700 Subject: [PATCH] feat: Codex rate-limit gauge row (5h + 7d windows) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a third row beneath the Claude cost rows showing Codex Pro rate-limit window consumption (flat-rate, not dollars). - CodexRateLimits dataclass: loads from ~/.codex/sessions rollouts, reverse-scans JSONL for last token_count event, overridable via YAS_CODEX_SESSIONS_DIR env var - Renderer.codex_pct_colour(): green <60%, yellow 60-85%, red >=85% - Renderer.codex_rate_row(): width-responsive — wide: icon 5h bar NN% | 7d bar NN% | plan [stale Xh] medium: icon 5h NN% · 7d NN% narrow: icon 5h NN% Stale indicator when newest event is older than 1h. No-data state renders 'no codex data' in dim grey. - Wired into build_wide() as a content row after the two Claude cost lines (session and day) - 18 new unit tests covering load, empty dirs, stale detection, color thresholds, and all three width modes --- README.md | 17 +++ claude/statusline_command.py | 182 ++++++++++++++++++++++++ test/test_codex_rate_limits.py | 245 +++++++++++++++++++++++++++++++++ 3 files changed, 444 insertions(+) create mode 100644 test/test_codex_rate_limits.py diff --git a/README.md b/README.md index 3254d15..b276dec 100644 --- a/README.md +++ b/README.md @@ -40,3 +40,20 @@ To demo/test: ```bash make demo ``` + +## Codex Rate-Limit Gauge + +A third row beneath the Claude cost rows shows your **Codex rate-limit window consumption** (rate limits, not dollars — Codex Pro is flat-rate). + +``` + 󰞬 5h ▆░░░░░░░ 12% │ 7d ▆▆░░░░░░ 28% │ pro +``` + +- Reads from `~/.codex/sessions///
/rollout-*.jsonl` +- Override the sessions root with `YAS_CODEX_SESSIONS_DIR` +- Shows the **5-hour** and **7-day** window consumption percentages +- Color thresholds: `< 60%` green · `60–85%` yellow · `> 85%` red +- If the latest rollout is older than 1 hour, a `(stale Xh)` indicator appears +- Narrows gracefully: medium width shows both percentages without bars; narrow shows only the 5h window +- If no Codex session data is found, renders `no codex data` in dim grey + diff --git a/claude/statusline_command.py b/claude/statusline_command.py index 172fcaf..b08a331 100755 --- a/claude/statusline_command.py +++ b/claude/statusline_command.py @@ -114,6 +114,7 @@ def terminal_width() -> int: # chat round-trips never lose the bytes. Render only in a Nerd-Font-capable # terminal. ICON_COST = '\uefc8' # nf-md currency-usd (cost row) +ICON_CODEX = '\U000f19e3' # nf-md-clock-time-eight-outline (Codex rate-limit row) ICON_TOK_RATE = '\U000f18a7' # nf-md gauge (t/m rate label) GLYPH_MODEL = '\U000f08b9' # nf-md-monitor-dashboard GLYPH_THINKING = '\U000f1a53' # nf-md-brain @@ -455,6 +456,100 @@ def from_dict(cls, d: dict) -> RateLimits: ) +@dataclass +class CodexRateLimits: + """Rate-limit state read from ~/.codex/sessions rollout files. + + All fields are None when no Codex rollout data is available. + ``primary`` = 5-hour window + ``secondary`` = 7-day window (10080 min) + """ + + primary_pct: float | None = None + secondary_pct: float | None = None + primary_resets_at: int | None = None + secondary_resets_at: int | None = None + plan_type: str | None = None + data_age_seconds: float | None = None + + # Stale threshold: if the newest token_count event is older than this, we + # surface a "(stale Xh)" indicator next to the plan label. + STALE_SECONDS: int = 3600 + + def is_stale(self) -> bool: + if self.data_age_seconds is None: + return False + return self.data_age_seconds > self.STALE_SECONDS + + @classmethod + def load(cls, sessions_root: Path | None = None) -> CodexRateLimits: + """Load the most recent token_count event from Codex rollout files. + + Args: + sessions_root: Override the default ``~/.codex/sessions`` root. + Falls back to the ``YAS_CODEX_SESSIONS_DIR`` env + var, then to the XDG default. + """ + if sessions_root is None: + env_dir = os.environ.get('YAS_CODEX_SESSIONS_DIR') + if env_dir: + sessions_root = Path(env_dir) + else: + sessions_root = HOME / '.codex' / 'sessions' + + rollout = cls._find_latest_rollout(sessions_root) + if rollout is None: + return cls() + + data_age = time.time() - rollout.stat().st_mtime + event = cls._last_token_count(rollout) + if event is None: + return cls() + + rl = event.get('rate_limits') or {} + primary = rl.get('primary') or {} + secondary = rl.get('secondary') or {} + return cls( + primary_pct = float(primary.get('used_percent', 0)), + secondary_pct = float(secondary.get('used_percent', 0)), + primary_resets_at = primary.get('resets_at'), + secondary_resets_at = secondary.get('resets_at'), + plan_type = rl.get('plan_type'), + data_age_seconds = data_age, + ) + + @staticmethod + def _find_latest_rollout(root: Path) -> Path | None: + """Return the most recently modified rollout-*.jsonl under root.""" + if not root.is_dir(): + return None + candidates = sorted(root.glob('*/*/[0-9][0-9]/rollout-*.jsonl'), key=lambda p: p.stat().st_mtime, reverse=True) + if not candidates: + # Also try direct rollout-*.jsonl at root (flat layout) + candidates = sorted(root.glob('rollout-*.jsonl'), key=lambda p: p.stat().st_mtime, reverse=True) + return candidates[0] if candidates else None + + @staticmethod + def _last_token_count(rollout: Path) -> dict | None: + """Reverse-scan ``rollout`` and return the last token_count payload.""" + try: + lines = rollout.read_bytes().splitlines() + except OSError: + return None + for raw in reversed(lines): + if b'token_count' not in raw: + continue + try: + d = json.loads(raw) + except (ValueError, TypeError): + continue + if d.get('type') == 'event_msg': + p = d.get('payload') or {} + if p.get('type') == 'token_count': + return p + return None + + @dataclass class SessionInfo: session_id: str = '' @@ -2413,6 +2508,88 @@ def helper(self, five_hour: RateBucket) -> str: except Exception as e: return f'{e.__class__.__name__}, {str(e)}' + # ------------------------------------------------------------------ + # Codex rate-limit gauge (3rd statusline row) + # ------------------------------------------------------------------ + + def codex_pct_colour(self, pct: float) -> str: + """Color for a Codex rate-limit percentage. + + Thresholds differ from Claude's fill_colour (70/90) because + Codex Pro is flat-rate — hitting 85%+ of a window is operationally + more significant than a dollar cost approaching a soft limit. + """ + if pct >= 85.0: + return self.alert + if pct >= 60.0: + return self.warn + return self.safe + + def _codex_pct_bar(self, pct: float, bar_w: int = 8) -> str: + """A short filled/empty bar for a Codex window percentage.""" + filled = max(0, min(bar_w, round(pct / 100 * bar_w))) + empty = bar_w - filled + clr = self.codex_pct_colour(pct) + return f'{clr}{BarChars.HEAVY * filled}{self.R}{self.BAR_EMPTY}{BarChars.EMPTY * empty}{self.R}' + + def codex_rate_row(self, rl: CodexRateLimits, width: int = 100) -> str: + """Render the Codex rate-limit gauge row. + + Width modes: + wide (>80) : icon 5h bar NN% | 7d bar NN% | plan [stale Xh] + medium (<=80): icon 5h NN% . 7d NN% + narrow (<=55): icon 5h NN% + """ + # No-data state + if rl.primary_pct is None: + return f' {self.LABEL}{ICON_CODEX} no codex data{self.R}' + + primary_pct = rl.primary_pct + secondary_pct = rl.secondary_pct if rl.secondary_pct is not None else 0.0 + plan = (rl.plan_type or '').lower() + + p_clr = self.codex_pct_colour(primary_pct) + s_clr = self.codex_pct_colour(secondary_pct) + + icon_str = f'{self.LABEL}{ICON_CODEX}{self.R}' + + if width <= NARROW_WIDTH: + # Narrow: icon + 5h pct only + return f' {icon_str} {self.LABEL}5h{self.R} {p_clr}{primary_pct:.0f}%{self.R}' + + if width <= MEDIUM_WIDTH: + # Medium: icon + 5h pct . 7d pct + return ( + f' {icon_str}' + f' {self.LABEL}5h{self.R} {p_clr}{primary_pct:.0f}%{self.R}' + f' {self.LABEL}\xb7{self.R}' + f' {self.LABEL}7d{self.R} {s_clr}{secondary_pct:.0f}%{self.R}' + ) + + # Wide: bars + plan label + optional stale indicator + p_bar = self._codex_pct_bar(primary_pct) + s_bar = self._codex_pct_bar(secondary_pct) + + stale_str = '' + if rl.is_stale() and rl.data_age_seconds is not None: + stale_h = int(rl.data_age_seconds // 3600) + stale_m = int((rl.data_age_seconds % 3600) // 60) + stale_label = f'{stale_h}h' if stale_h else f'{stale_m}m' + stale_str = f' {self.COMMIT}(stale {stale_label}){self.R}' + + sep = f' {self.LABEL}│{self.R} ' + + return ( + f' {icon_str}' + f' {self.LABEL}5h{self.R} {p_bar} {p_clr}{primary_pct:.0f}%{self.R}' + f'{sep}' + f'{self.LABEL}7d{self.R} {s_bar} {s_clr}{secondary_pct:.0f}%{self.R}' + f'{sep}' + f'{self.LABEL}{plan}{self.R}' + f'{stale_str}' + ) + + @dataclass class RowSpec: kind: str # 'top_border', 'bottom_border', 'separator', 'separator_dim', 'content' @@ -2563,6 +2740,7 @@ def build_wide(session: SessionInfo, width: int, r: Renderer) -> LayoutSpec: subagents = RunningSubagents.from_session(session.session_id, session.workspace.project_dir) tasks = TaskList.from_session(session.transcript_path) elapsed = elapsed_from_transcript(session.transcript_path) + codex_rl = CodexRateLimits.load() git = GitInfo.from_cwd(session.cwd) helper_text, right_text, right_w = r.model_right_section( @@ -2621,6 +2799,10 @@ def build_wide(session: SessionInfo, width: int, r: Renderer) -> LayoutSpec: for lt in line_tokens: rows.append(RowSpec('content', content=lt)) + # Codex rate-limit gauge row (3rd cost row — flat-rate windows, not dollars) + codex_row = r.codex_rate_row(codex_rl, width) + rows.append(RowSpec('content', content=codex_row)) + # First post-tokens separator threads `ups` back into the tokens vseps and # is drawn as the heavy "seam" marking the static→dynamic split. Only the # first one — later inter-section separators keep their normal style. When diff --git a/test/test_codex_rate_limits.py b/test/test_codex_rate_limits.py new file mode 100644 index 0000000..88279a1 --- /dev/null +++ b/test/test_codex_rate_limits.py @@ -0,0 +1,245 @@ +"""Tests for CodexRateLimits — Codex rate-limit gauge (3rd statusline row). + +Covers: +- Loading from a synthetic rollouts dir with known values +- Empty rollouts dir → all-None fields +- Stale data → data_age_seconds > threshold +- Color threshold transitions (60%, 85%) +- Width-mode render: full / medium / narrow +""" + +import json +import time +from pathlib import Path + +import pytest +import statusline_command as sl + +from helper import strip_ansi + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _write_rollout(root: Path, year: int, month: int, day: int, events: list[dict]) -> Path: + """Write a rollout-0.jsonl file under /YYYY/MM/DD/.""" + day_dir = root / f'{year:04d}' / f'{month:02d}' / f'{day:02d}' + day_dir.mkdir(parents=True, exist_ok=True) + rollout = day_dir / 'rollout-0.jsonl' + with rollout.open('w') as fh: + for evt in events: + fh.write(json.dumps(evt) + '\n') + return rollout + + +def _token_count_event(primary_pct: float, secondary_pct: float, + primary_resets: int, secondary_resets: int, + plan_type: str = 'pro', + ts: str = '2026-05-25T03:47:10.921Z') -> dict: + return { + 'timestamp': ts, + 'type': 'event_msg', + 'payload': { + 'type': 'token_count', + 'rate_limits': { + 'primary': {'used_percent': primary_pct, 'window_minutes': 300, 'resets_at': primary_resets}, + 'secondary': {'used_percent': secondary_pct, 'window_minutes': 10080, 'resets_at': secondary_resets}, + 'plan_type': plan_type, + }, + }, + } + + +def _other_event() -> dict: + return {'timestamp': '2026-05-25T03:00:00.000Z', 'type': 'event_msg', 'payload': {'type': 'other'}} + + +# --------------------------------------------------------------------------- +# CodexRateLimits.load() — data loading tests +# --------------------------------------------------------------------------- + +class TestCodexRateLimitsLoad: + def test_load_known_values(self, tmp_path: Path) -> None: + now_epoch = int(time.time()) + primary_resets = now_epoch + 3600 + secondary_resets = now_epoch + 86400 + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(12.0, 28.0, primary_resets, secondary_resets), + ]) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.primary_pct == pytest.approx(12.0) + assert rl.secondary_pct == pytest.approx(28.0) + assert rl.primary_resets_at == primary_resets + assert rl.secondary_resets_at == secondary_resets + assert rl.plan_type == 'pro' + + def test_load_picks_last_token_count(self, tmp_path: Path) -> None: + """When multiple token_count events exist, the LAST one wins.""" + now_epoch = int(time.time()) + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(5.0, 10.0, now_epoch + 100, now_epoch + 200), + _other_event(), + _token_count_event(55.0, 77.0, now_epoch + 150, now_epoch + 250), + ]) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.primary_pct == pytest.approx(55.0) + assert rl.secondary_pct == pytest.approx(77.0) + + def test_empty_dir_returns_all_none(self, tmp_path: Path) -> None: + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.primary_pct is None + assert rl.secondary_pct is None + assert rl.primary_resets_at is None + assert rl.secondary_resets_at is None + assert rl.plan_type is None + assert rl.data_age_seconds is None + + def test_no_token_count_events_returns_all_none(self, tmp_path: Path) -> None: + """A rollout file with only non-token_count events → all None.""" + _write_rollout(tmp_path, 2026, 5, 25, [_other_event(), _other_event()]) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.primary_pct is None + + def test_env_var_override(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + now_epoch = int(time.time()) + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(33.0, 44.0, now_epoch + 100, now_epoch + 200), + ]) + monkeypatch.setenv('YAS_CODEX_SESSIONS_DIR', str(tmp_path)) + # load() with no argument should pick up the env var + rl = sl.CodexRateLimits.load() + assert rl.primary_pct == pytest.approx(33.0) + monkeypatch.delenv('YAS_CODEX_SESSIONS_DIR', raising=False) + + def test_data_age_seconds_recent(self, tmp_path: Path) -> None: + """data_age_seconds should be small for a just-written rollout.""" + now_epoch = int(time.time()) + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(1.0, 2.0, now_epoch + 100, now_epoch + 200), + ]) + rl = sl.CodexRateLimits.load(tmp_path) + # The rollout was written <5s ago; age should be < 10s + assert rl.data_age_seconds is not None + assert rl.data_age_seconds < 10.0 + + def test_stale_indicator(self, tmp_path: Path) -> None: + """A rollout older than 3600s should be considered stale.""" + now_epoch = int(time.time()) + rollout = _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(1.0, 2.0, now_epoch + 100, now_epoch + 200), + ]) + # Back-date the file's mtime by 2 hours + old_mtime = time.time() - 7200 + import os + os.utime(rollout, (old_mtime, old_mtime)) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.data_age_seconds is not None + assert rl.data_age_seconds > 3600 + assert rl.is_stale() + + def test_plan_type_free(self, tmp_path: Path) -> None: + now_epoch = int(time.time()) + _write_rollout(tmp_path, 2026, 5, 25, [ + _token_count_event(0.0, 0.0, now_epoch + 100, now_epoch + 200, plan_type='free'), + ]) + rl = sl.CodexRateLimits.load(tmp_path) + assert rl.plan_type == 'free' + + +# --------------------------------------------------------------------------- +# Color threshold tests +# --------------------------------------------------------------------------- + +class TestCodexColorThresholds: + def setup_method(self) -> None: + self.r = sl.Renderer() + + def test_under_60_is_safe(self) -> None: + assert self.r.codex_pct_colour(0.0) == self.r.safe + assert self.r.codex_pct_colour(59.9) == self.r.safe + + def test_60_to_85_is_warn(self) -> None: + assert self.r.codex_pct_colour(60.0) == self.r.warn + assert self.r.codex_pct_colour(84.9) == self.r.warn + + def test_above_85_is_alert(self) -> None: + assert self.r.codex_pct_colour(85.0) == self.r.alert + assert self.r.codex_pct_colour(100.0) == self.r.alert + + +# --------------------------------------------------------------------------- +# Render tests — codex_rate_row +# --------------------------------------------------------------------------- + +class TestCodexRateRow: + def setup_method(self) -> None: + self.r = sl.Renderer() + + def _rl(self, primary: float = 12.0, secondary: float = 28.0, + plan: str = 'pro') -> sl.CodexRateLimits: + return sl.CodexRateLimits( + primary_pct=primary, + secondary_pct=secondary, + primary_resets_at=None, + secondary_resets_at=None, + plan_type=plan, + data_age_seconds=30.0, + ) + + def test_full_mode_contains_5h_and_7d(self) -> None: + row = self.r.codex_rate_row(self._rl(), width=120) + plain = strip_ansi(row) + assert '5h' in plain + assert '7d' in plain + assert '12' in plain # primary pct + assert '28' in plain # secondary pct + assert 'pro' in plain + + def test_full_mode_contains_plan(self) -> None: + row = self.r.codex_rate_row(self._rl(plan='free'), width=120) + plain = strip_ansi(row) + assert 'free' in plain + + def test_medium_mode_has_both_pcts(self) -> None: + row = self.r.codex_rate_row(self._rl(), width=79) + plain = strip_ansi(row) + assert '5h' in plain + assert '7d' in plain + assert '12' in plain + assert '28' in plain + + def test_narrow_mode_has_primary_only(self) -> None: + row = self.r.codex_rate_row(self._rl(), width=50) + plain = strip_ansi(row) + assert '5h' in plain + assert '12' in plain + # In narrow mode we may omit 7d label and secondary pct entirely + assert '7d' not in plain + + def test_no_data_state(self) -> None: + rl = sl.CodexRateLimits( + primary_pct=None, secondary_pct=None, + primary_resets_at=None, secondary_resets_at=None, + plan_type=None, data_age_seconds=None, + ) + row = self.r.codex_rate_row(rl, width=120) + plain = strip_ansi(row) + assert 'no codex data' in plain.lower() or plain.strip() == '' + + def test_stale_shows_indicator(self) -> None: + rl = sl.CodexRateLimits( + primary_pct=5.0, secondary_pct=10.0, + primary_resets_at=None, secondary_resets_at=None, + plan_type='pro', data_age_seconds=7200.0, + ) + row = self.r.codex_rate_row(rl, width=120) + plain = strip_ansi(row) + # Stale indicator: either "stale" text or a warning symbol + assert 'stale' in plain.lower() or '?' in plain or '⚠' in plain + + def test_bar_chars_present_in_full_mode(self) -> None: + """Full mode should contain filled/empty bar chars.""" + row = self.r.codex_rate_row(self._rl(primary=50.0, secondary=50.0), width=120) + # Bar characters: ▓ (heavy) or █ (filled) and ░ (empty) + assert '▓' in row or '█' in row or '░' in row