Report Codex usage and account status on GET /v1/status

README.md

@@ -111,6 +111,7 @@ account. The current catalog commonly includes:
 - OpenAI-compatible `/v1/responses` (HTTP + WebSocket)
 - Ollama-compatible endpoints
 - Reasoning effort exposed as separate models (optional)
+- Codex usage and account status on `GET /v1/status` (see below)
 
 <br>
 
@@ -158,6 +159,38 @@ All flags go after `chatmock serve`. These can also be set as environment variab
 
 <br>
 
+## Usage and account status
+
+`GET /v1/status` reports the signed-in account and the plan usage Codex reported
+on the most recent proxied request — the same snapshot the `chatmock info`
+command prints. Like the Codex CLI, it is not a live probe: the backend only
+reports usage on replies it sends, so the numbers update after each request
+through the proxy and cost nothing to read.
+
+```json
+{
+  "account": {
+    "name": "Ada Lovelace",
+    "email": "ada@example.com",
+    "plan_type": "plus",
+    "account_id": "1f6f92a2-..."
+  },
+  "rate_limits": {
+    "captured_at": "2026-08-26T08:03:58+00:00",
+    "primary": { "used_percent": 12.5, "window_minutes": 10080, "resets_in_seconds": 345600 },
+    "secondary": { "used_percent": 3.0, "window_minutes": 300, "resets_in_seconds": 1799 }
+  }
+}
+```
+
+The account values are display metadata derived from the local token claims —
+no token is ever included. A field whose claim is missing is omitted; `account`
+is `null` when signed out, and `rate_limits` is `null` until a first request
+has been proxied. Do not assume `primary` is the shorter window; classify by
+`window_minutes`.
+
+<br>
+
 ## Important notice
 
 Use responsibly and at your own risk. This project is not affiliated with OpenAI.

chatmock/codex_status.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import os
+from typing import Any, Dict
+
+from .limits import StoredRateLimitSnapshot, load_rate_limit_snapshot
+from .utils import auth_file_candidates, load_chatgpt_tokens, parse_jwt_claims
+
+_OPENAI_AUTH_CLAIM = "https://api.openai.com/auth"
+
+# The account claims only change when the auth file does, so the derived
+# info is cached against its mtime/size: a request pays one os.stat per
+# candidate path, not a read + JSON parse + JWT decode.
+_CACHE_UNSET: Any = object()
+_account_cache_key: Any = _CACHE_UNSET
+_account_cache: Dict[str, str] = {}
+
+
+def _auth_files_state() -> tuple:
+    state = []
+    for path in auth_file_candidates():
+        try:
+            st = os.stat(path)
+        except OSError:
+            continue
+        state.append((path, st.st_mtime_ns, st.st_size))
+    return tuple(state)
+
+
+def reset_account_info_cache() -> None:
+    global _account_cache_key, _account_cache
+    _account_cache_key = _CACHE_UNSET
+    _account_cache = {}
+
+
+def account_info() -> Dict[str, str]:
+    """Display metadata about the signed-in account, derived from token claims.
+
+    Never carries a token, and never a claim set wholesale: only the display
+    values, each omitted when its claim is unavailable.
+    """
+    global _account_cache_key, _account_cache
+    key = _auth_files_state()
+    if key == _account_cache_key:
+        return dict(_account_cache)
+    info = _compute_account_info()
+    _account_cache_key = key
+    _account_cache = info
+    return dict(info)
+
+
+def _display_str(value: Any) -> str | None:
+    if not isinstance(value, str):
+        return None
+    cleaned = value.strip()
+    return cleaned or None
+
+
+def _compute_account_info() -> Dict[str, str]:
+    try:
+        access_token, account_id, id_token = load_chatgpt_tokens(ensure_fresh=False)
+    except Exception:
+        return {}
+    id_claims = parse_jwt_claims(id_token) if isinstance(id_token, str) else None
+    access_claims = parse_jwt_claims(access_token) if isinstance(access_token, str) else None
+    id_claims = id_claims if isinstance(id_claims, dict) else {}
+    access_claims = access_claims if isinstance(access_claims, dict) else {}
+
+    info: Dict[str, str] = {}
+
+    email = _display_str(id_claims.get("email"))
+    name = None
+    for candidate in (id_claims.get("name"), id_claims.get("email"), id_claims.get("preferred_username"), account_id):
+        name = _display_str(candidate)
+        if name:
+            break
+    if name:
+        info["name"] = name
+    if email:
+        info["email"] = email
+
+    auth_claims = access_claims.get(_OPENAI_AUTH_CLAIM)
+    if isinstance(auth_claims, dict):
+        plan = _display_str(auth_claims.get("chatgpt_plan_type"))
+        if plan:
+            info["plan_type"] = plan
+
+    safe_account_id = _display_str(account_id)
+    if safe_account_id:
+        info["account_id"] = safe_account_id
+    return info
+
+
+def _window_payload(window: Any) -> Dict[str, Any] | None:
+    if window is None:
+        return None
+    return {
+        "used_percent": window.used_percent,
+        "window_minutes": window.window_minutes,
+        "resets_in_seconds": window.resets_in_seconds,
+    }
+
+
+def build_status_payload() -> Dict[str, Any]:
+    """The signed-in account and the last usage snapshot Codex reported.
+
+    The usage half mirrors the CLI's `info` command: it is the snapshot
+    recorded from the most recent proxied request, not a live probe — the
+    Codex backend only reports usage on replies it sends.
+    """
+    account = account_info() or None
+
+    stored: StoredRateLimitSnapshot | None = load_rate_limit_snapshot()
+    rate_limits: Dict[str, Any] | None = None
+    if stored is not None:
+        rate_limits = {
+            "captured_at": stored.captured_at.isoformat(),
+            "primary": _window_payload(stored.snapshot.primary),
+            "secondary": _window_payload(stored.snapshot.secondary),
+        }
+
+    return {"account": account, "rate_limits": rate_limits}

chatmock/reasoning.py

@@ -18,14 +18,20 @@ def build_reasoning_param(
     valid_efforts = allowed_efforts or DEFAULT_REASONING_EFFORTS
     valid_summaries = {"auto", "concise", "detailed", "none"}
 
+    # A caller's explicit effort is forwarded when upstream knows it at all;
+    # upstream validates per model. Filtering against the catalog here silently
+    # rewrote efforts the catalog omits but upstream honours (e.g. `none`).
+    explicit_effort: str | None = None
     if isinstance(overrides, dict):
         o_eff = str(overrides.get("effort", "")).strip().lower()
         o_sum = str(overrides.get("summary", "")).strip().lower()
-        if o_eff in valid_efforts and o_eff:
-            effort = o_eff
+        if o_eff in DEFAULT_REASONING_EFFORTS:
+            explicit_effort = o_eff
         if o_sum in valid_summaries and o_sum:
             summary = o_sum
-    if effort not in valid_efforts:
+    if explicit_effort is not None:
+        effort = explicit_effort
+    elif effort not in valid_efforts:
         effort = "medium"
     if summary not in valid_summaries:
         summary = "auto"

chatmock/responses_api.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import copy
 import json
 from dataclasses import dataclass
 from typing import Any, Dict, Iterable, Iterator, List
@@ -171,6 +172,8 @@ def aggregate_response_from_sse(
 ) -> tuple[Dict[str, Any] | None, Dict[str, Any] | None]:
     response_obj: Dict[str, Any] | None = None
     error_obj: Dict[str, Any] | None = None
+    completed_output_items: Dict[int, Dict[str, Any]] = {}
+    unindexed_output_items = 0
     try:
         for evt in iter_sse_event_payloads(upstream):
             if callable(on_event):
@@ -182,13 +185,34 @@ def aggregate_response_from_sse(
             if isinstance(response, dict):
                 response_obj = response
             kind = evt.get("type")
+            if kind == "response.output_item.done":
+                item = evt.get("item")
+                if isinstance(item, dict):
+                    output_index = evt.get("output_index")
+                    if not isinstance(output_index, int):
+                        # Indexed items are the protocol norm. Keep malformed or
+                        # older unindexed events deterministically after them,
+                        # preserving their arrival order.
+                        output_index = 1_000_000 + unindexed_output_items
+                        unindexed_output_items += 1
+                    completed_output_items[output_index] = copy.deepcopy(item)
             if kind == "response.failed":
                 if isinstance(response, dict) and isinstance(response.get("error"), dict):
                     error_obj = {"error": response.get("error")}
                 else:
                     error_obj = {"error": {"message": "response.failed"}}
                 break
             if kind == "response.completed":
+                if (
+                    isinstance(response_obj, dict)
+                    and completed_output_items
+                    and not response_obj.get("output")
+                ):
+                    response_obj = dict(response_obj)
+                    response_obj["output"] = [
+                        completed_output_items[index]
+                        for index in sorted(completed_output_items)
+                    ]
                 break
     finally:
         upstream.close()

chatmock/routes_openai.py

@@ -6,6 +6,7 @@
 
 from flask import Blueprint, Response, current_app, jsonify, make_response, request
 
+from .codex_status import build_status_payload
 from .fast_mode import resolve_service_tier
 from .limits import record_rate_limits_from_response
 from .http import build_cors_headers
@@ -717,6 +718,14 @@ def responses_create() -> Response:
     return resp
 
 
+@openai_bp.route("/v1/status", methods=["GET"])
+def codex_status() -> Response:
+    resp = make_response(jsonify(build_status_payload()), 200)
+    for k, v in build_cors_headers().items():
+        resp.headers.setdefault(k, v)
+    return resp
+
+
 @openai_bp.route("/v1/models", methods=["GET"])
 def list_models() -> Response:
     expose_variants = bool(current_app.config.get("EXPOSE_REASONING_MODELS"))

chatmock/utils.py

@@ -53,16 +53,19 @@ def resolve_installation_id() -> str:
         return str(uuid.uuid4())
 
 
-def read_auth_file() -> Dict[str, Any] | None:
-    for base in [
+def auth_file_candidates() -> List[str]:
+    """The auth.json paths considered, in read priority order."""
+    bases = (
         os.getenv("CHATGPT_LOCAL_HOME"),
         os.getenv("CODEX_HOME"),
         os.path.expanduser("~/.chatgpt-local"),
         os.path.expanduser("~/.codex"),
-    ]:
-        if not base:
-            continue
-        path = os.path.join(base, "auth.json")
+    )
+    return [os.path.join(base, "auth.json") for base in bases if base]
+
+
+def read_auth_file() -> Dict[str, Any] | None:
+    for path in auth_file_candidates():
         try:
             with open(path, "r", encoding="utf-8") as f:
                 return json.load(f)