"""Automatic redemption of earned Codex usage-limit reset credits. The Codex CLI exposes reset credits through the same authenticated backend as usage telemetry. Each scheduled eligible usage pass reads the credit endpoint, even when the usage response's embedded credit summary is empty or out of date. A pending idempotency key is written before the mutation so a lost response can be retried without spending twice. """ from __future__ import annotations import json import os import time import urllib.error import urllib.parse import urllib.request import uuid from codex_reset_windows import reset_windows RESET_AT_USED_PERCENT = 95 REDEEM_COOLDOWN_SECONDS = 900 STATE_FILE = ".usage-reset.json" # A park that lifts on its own within this long keeps the credit (operator rule, # 2026-09-22: a PARKED account uses its reset; only a sub-hour wall is not worth one). PARK_HOLD_SECONDS = 3600 def _active_park(account_dir: str, now: int) -> int | None: """The epoch a USAGE park on this account lifts at (a >=threshold pass, or a 429 the client received), or None. Error cool-downs and expired markers are not parks.""" try: with open(os.path.join(account_dir, ".limited"), errors="replace") as handle: text = handle.read(4096) until = int(text.splitlines()[0]) except (OSError, ValueError, IndexError): return None if until <= now or not ("reason=limits" in text or "reason=client-rate-limit" in text): return None return until def auto_reset_enabled() -> bool: """Return whether automatic redemption is enabled (on by default).""" return os.environ.get("CODEX_MULTIACC_AUTO_RESET", "1").strip().lower() \ not in {"0", "false", "no", "off"} def _load(path: str) -> dict: try: value = json.load(open(path, encoding="utf-8")) return value if isinstance(value, dict) else {} except Exception: return {} def _write(path: str, value: dict) -> None: temp = f"{path}.tmp.{os.getpid()}" descriptor = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(value, handle, indent=2, sort_keys=True) handle.write("\n") os.replace(temp, path) def _reset_urls(usage_url: str) -> tuple[str, str]: explicit_list = os.environ.get("CODEX_MULTIACC_RESET_CREDITS_URL", "").strip() explicit_consume = os.environ.get("CODEX_MULTIACC_RESET_CONSUME_URL", "").strip() if explicit_list and explicit_consume: return explicit_list, explicit_consume parsed = urllib.parse.urlsplit(usage_url) path = parsed.path if path.endswith("/wham/usage"): prefix = path[: -len("/wham/usage")] + "/wham" elif path.endswith("/api/codex/usage"): prefix = path[: -len("/usage")] elif "/backend-api/" in path: prefix = path.split("/backend-api/", 1)[0] + "/backend-api/wham" else: prefix = path.rsplit("/usage", 1)[0] base = urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, prefix, "", "")) return (explicit_list or f"{base}/rate-limit-reset-credits", explicit_consume or f"{base}/rate-limit-reset-credits/consume") def _request(url: str, headers: dict, payload: dict | None = None) -> dict: data = json.dumps(payload).encode() if payload is not None else None request = urllib.request.Request(url, data=data, headers=headers, method="POST" if data is not None else "GET") response = urllib.request.urlopen(request, timeout=15) value = json.loads(response.read().decode()) if not isinstance(value, dict): raise ValueError("reset endpoint returned a non-object") return value def _credit_to_redeem(details: dict) -> tuple[int, str | None]: try: available_count = max(0, int(details.get("available_count") or 0)) except (TypeError, ValueError): available_count = 0 available = [item for item in details.get("credits", []) if isinstance(item, dict) and item.get("status") == "available"] available.sort(key=lambda item: item.get("expires_at") or "9999-12-31T23:59:59Z") credit_id = str(available[0].get("id") or "") if available else "" return available_count, credit_id or None def _pending_request(account_dir: str, now: int) -> tuple[dict, str]: state_path = os.path.join(account_dir, STATE_FILE) state = _load(state_path) if state.get("state") == "complete" and int(state.get("suppress_until") or 0) > now: return state, "cooldown" if state.get("state") == "pending" and state.get("redeem_request_id"): return state, "pending" return {}, "new" def _redeem_request_id(account_id: str, usage: dict, headers: dict, now: int) -> str: """Return one fleet-stable UUID for this account's current limit windows.""" subject = str(headers.get("chatgpt-account-id") or account_id) _, resets = reset_windows(usage, RESET_AT_USED_PERCENT) cycle = resets or [f"hour-{now // 3600}"] seed = json.dumps({"account": subject, "resets": cycle}, sort_keys=True) return str(uuid.uuid5(uuid.NAMESPACE_URL, seed)) def try_auto_redeem(account_dir: str, account_id: str, used_percent: int, usage: dict, usage_url: str, headers: dict, say, now: int | None = None, credit_details: dict | None = None, threshold: int = RESET_AT_USED_PERCENT) -> dict: """Redeem one available reset and return a non-secret outcome document. Eligible whenever the pool has PARKED the account: a window at/over the pool's own exclusion threshold (90 by default — the 95% line left accounts at 90-94% parked for days with credits unused), a finished limit, or an active usage park that lifts more than PARK_HOLD_SECONDS from now (a 429 the telemetry may not show).""" current = int(now if now is not None else time.time()) finished, _ = reset_windows(usage, RESET_AT_USED_PERCENT) park = _active_park(account_dir, current) parked = park is not None and park - current >= PARK_HOLD_SECONDS if not auto_reset_enabled() or (used_percent < min(threshold, RESET_AT_USED_PERCENT) and not finished and not parked): return {"status": "not_eligible"} state, disposition = _pending_request(account_dir, current) if disposition == "cooldown": return {"status": "cooldown", "outcome": state.get("outcome")} list_url, consume_url = _reset_urls(usage_url) if disposition == "new": # Newly granted credits can precede the usage response's embedded summary. # The scheduled usage cadence bounds these reads, including spent accounts. try: details = credit_details if credit_details is not None else _request(list_url, headers) available_count, credit_id = _credit_to_redeem(details) except Exception as error: say(f"{account_id}: usage reset availability failed ({type(error).__name__}); failing open") return {"status": "error"} if available_count <= 0: return {"status": "no_credit"} request_id = _redeem_request_id(account_id, usage, headers, current) state = {"schema": 1, "state": "pending", "redeem_request_id": request_id, "credit_id": credit_id, "started_at": current} _write(os.path.join(account_dir, STATE_FILE), state) payload = {"redeem_request_id": state["redeem_request_id"]} if state.get("credit_id"): payload["credit_id"] = state["credit_id"] try: response = _request(consume_url, headers, payload) except Exception as error: say(f"{account_id}: usage reset redeem failed ({type(error).__name__}); retry is idempotent") return {"status": "pending"} outcome = str(response.get("code") or "unknown") if outcome in {"reset", "already_redeemed"}: complete = {"schema": 1, "state": "complete", "outcome": outcome, "redeem_request_id": state["redeem_request_id"], "redeemed_at": current, "suppress_until": current + REDEEM_COOLDOWN_SECONDS, "windows_reset": int(response.get("windows_reset") or 0)} _write(os.path.join(account_dir, STATE_FILE), complete) say(f"{account_id}: usage reset redeemed automatically at {used_percent}% used" + ("" if used_percent >= threshold or finished else " (parked by a client-reported limit)")) return {"status": "redeemed", "outcome": outcome, "windows_reset": complete["windows_reset"], "redeemed_at": current} try: os.remove(os.path.join(account_dir, STATE_FILE)) except OSError: pass return {"status": outcome}