Add limit helper and bump version to 0.6.1

Author: yieldthought

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "codexapi"
-version = "0.6.0"
+version = "0.6.1"
 description = "Minimal Python API for running the Codex CLI."
 readme = "README.md"
 requires-python = ">=3.8"

src/codexapi/__init__.py

@@ -3,6 +3,7 @@
 from .agent import Agent, agent
 from .foreach import ForeachResult, foreach
 from .pushover import Pushover
+from .rate_limits import quota_line, rate_limits
 from .ralph import Ralph
 from .science import Science
 from .task import Task, TaskFailed, TaskResult, task, task_result
@@ -11,6 +12,8 @@
     "Agent",
     "ForeachResult",
     "Pushover",
+    "quota_line",
+    "rate_limits",
     "Ralph",
     "Science",
     "Task",
@@ -21,4 +24,4 @@
     "task",
     "task_result",
 ]
-__version__ = "0.6.0"
+__version__ = "0.6.1"

src/codexapi/cli.py

@@ -18,6 +18,7 @@
 from .science import Science
 from .task import DEFAULT_MAX_ITERATIONS, TaskFailed, task
 from .taskfile import TaskFile, load_task_file, task_def_uses_item
+from .rate_limits import quota_line
 
 _SESSION_ID_RE = re.compile(
     r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
@@ -1298,6 +1299,10 @@ def main(argv=None):
         "top",
         help="Show running Codex sessions.",
     )
+    subparsers.add_parser(
+        "limit",
+        help="Show Codex rate limits.",
+    )
 
     args = parser.parse_args(argv)
     if args.command is None:
@@ -1318,6 +1323,9 @@ def main(argv=None):
     if args.command == "top":
         _run_top([])
         return
+    if args.command == "limit":
+        print(quota_line())
+        return
 
     if args.command == "foreach":
         if args.n is not None and args.n < 1:

src/codexapi/pushover.py

@@ -8,6 +8,8 @@
 import urllib.parse
 import urllib.request
 
+from .rate_limits import quota_line
+
 _PUSHOVER_PATH = "~/.pushover"
 _PUSHOVER_URL = "https://api.pushover.net/1/messages.json"
 _MAX_MESSAGE = 1024
@@ -58,6 +60,7 @@ def send(self, title, message):
         message_text = (message or "").strip()
         if not message_text:
             return False
+        message_text = _append_quota_line(message_text)
         message_text = _truncate(message_text, _MAX_MESSAGE)
         payload = urllib.parse.urlencode(
             {
@@ -177,3 +180,10 @@ def _truncate(text, limit):
 
 def _warn(message):
     print(message, file=sys.stderr)
+
+
+def _append_quota_line(message):
+    line = quota_line()
+    if not line:
+        return message
+    return f"{message}\n{line}"

src/codexapi/rate_limits.py

@@ -0,0 +1,97 @@
+"""Helpers for reading Codex rate limits from session logs."""
+
+import json
+import os
+import time
+from pathlib import Path
+
+_QUOTA_PREFIX = "Limits:"
+
+
+def rate_limits():
+    """Return the latest rate_limits dict from Codex session logs."""
+    root = Path(os.environ.get("CODEX_HOME", "~/.codex")).expanduser()
+    sessions = root / "sessions"
+    if not sessions.exists():
+        return None
+    candidates = []
+    for dirpath, _dirnames, filenames in os.walk(sessions):
+        for name in filenames:
+            if not name.endswith(".jsonl"):
+                continue
+            path = os.path.join(dirpath, name)
+            try:
+                mtime = os.path.getmtime(path)
+            except OSError:
+                continue
+            candidates.append((mtime, path))
+    if not candidates:
+        return None
+    for _mtime, path in sorted(candidates, reverse=True):
+        found = _extract_rate_limits(path)
+        if found is not None:
+            return found
+    return None
+
+
+def _extract_rate_limits(path):
+    last = None
+    try:
+        with open(path, "r", encoding="utf-8", errors="replace") as handle:
+            for line in handle:
+                if '"rate_limits"' not in line:
+                    continue
+                try:
+                    event = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                payload = event.get("payload") or {}
+                rate_data = payload.get("rate_limits")
+                if isinstance(rate_data, dict):
+                    last = rate_data
+    except OSError:
+        return None
+    return last
+
+
+def quota_line():
+    """Return a human-readable quota line."""
+    data = rate_limits()
+    if not data:
+        return f"{_QUOTA_PREFIX} unavailable"
+    primary = data.get("primary") or {}
+    secondary = data.get("secondary") or {}
+    primary_left = _percent_left(primary.get("used_percent"))
+    secondary_left = _percent_left(secondary.get("used_percent"))
+    primary_reset = _format_reset(primary.get("resets_at"))
+    secondary_reset = _format_reset(secondary.get("resets_at"))
+    if primary_left is None or secondary_left is None:
+        return f"{_QUOTA_PREFIX} unavailable"
+    return (
+        f"{_QUOTA_PREFIX} {primary_left}% / {secondary_left}% left "
+        f"(reset in {primary_reset} / {secondary_reset})"
+    )
+
+
+def _percent_left(used_percent):
+    if not isinstance(used_percent, (int, float)):
+        return None
+    left = 100.0 - float(used_percent)
+    if left < 0:
+        left = 0.0
+    if left > 100:
+        left = 100.0
+    return int(round(left))
+
+
+def _format_reset(resets_at):
+    if not isinstance(resets_at, (int, float)):
+        return "unknown"
+    remaining = float(resets_at) - time.time()
+    if remaining < 0:
+        remaining = 0
+    hours = remaining / 3600.0
+    if hours > 24:
+        days = hours / 24.0
+        return f"{int(round(days))}d"
+    return f"{int(round(hours))}h"