"""Machine-readable pool report — the ONE place that builds the JSON `list --json`, `status --json` and `limits --json` emit, for BOTH providers, so a consumer (the app-robot panel/runner daemon) parses exactly one shape no matter which verb or provider produced it. Run: python3 lib/report.py [list|status|limits] The document is a single JSON object on stdout: schema "claude-multiacc/pool.v1" — bumped only on a BREAKING change provider claude | codex kind which verb rendered it (list|status|limits) pool root, manifest path, threshold, sync target/mode/role, peers accounts[] id, email, home, home_dir, status, state, credential_class, usage, ... summary counts a dashboard can render without walking the list warnings[] duplicate emails, unreadable manifest bits — never fatal Every field is best-effort: one corrupt account file degrades that account only, never the document (a --json call that dies is worse than one that says 'unknown'). Account status vocabulary (the panel's contract): active selectable right now limited authenticates, but a >=threshold bucket is parked until limit_reset_at expired has auth material that cannot authenticate — needs an interactive login blocked authenticates, but the org/workspace disabled the CLI for it locked claude only: the OAuth login is in the macOS Keychain and the process that built this report could not open it (ssh/background session) — the Mac's own GUI session uses it fine; never reported by a launchd runner missing no auth material on this machine (and this machine should own it) remote no auth here on purpose — the manifest says another machine owns it Credential classes (what may be COPIED between machines): portable claude setup-token (server.token) — safe to export/import anywhere machine-local claude .credentials.json OR a macOS Keychain item (credentials.oauth_store says which) / codex auth.json — rotating refresh grant; copying it makes two machines fight over one grant and breaks both none nothing on disk (or in the Keychain) here """ import json import os import re import socket import sys import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import keychain # noqa: E402 from audit import creds_doc_state, token_verified # noqa: E402 from claude_reset import marker_superseded, record_from # noqa: E402 SCHEMA = 'claude-multiacc/pool.v1' # Must match the SHIM's ranking window, or this document calls data "fresh" that # selection has already stopped ranking on. The two providers differ on purpose: # api.anthropic.com hands out Retry-After: 3600, so bin/claude trusts an hour; # bin/codex still uses 900 against chatgpt.com, which has not been measured. _STALE_DEFAULT = {'claude': 3600, 'codex': 900} def _stale_after(provider): env = {'claude': 'CLAUDE_MULTIACC_STALE_AFTER', 'codex': 'CODEX_MULTIACC_STALE_AFTER'}.get(provider) default = _STALE_DEFAULT.get(provider, 900) try: v = int(os.environ.get(env) or default) if env else default except ValueError: v = default return v if v > 0 else default VALID_ID = re.compile(r'acct-\d{2}') def _iso(epoch): try: epoch = float(epoch) except (TypeError, ValueError): return None if epoch <= 0: return None try: return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(epoch)) except (ValueError, OSError, OverflowError): # absurd epoch in a corrupt file return None def _iso_ms(value): """A milliseconds field (credentials carry them) -> ISO, tolerating whatever a corrupt credential holds: a string, a list, null. One bad file must degrade its own account, never the document.""" try: return _iso(float(value or 0) / 1000.0) except (TypeError, ValueError): return None def _size(path): try: return os.path.getsize(path) except OSError: return 0 def _mtime(path): try: return os.path.getmtime(path) except OSError: return 0.0 def _load_json(path): try: with open(path) as f: doc = json.load(f) return doc if isinstance(doc, dict) else None except Exception: return None def _marker(path): """(reset_epoch, detail) of a .limited/.expired marker; (0, '') when absent.""" try: lines = open(path, errors='replace').read().splitlines() except OSError: return 0, '' reset = int(lines[0]) if lines and lines[0].strip().isdigit() else 0 detail = lines[1].strip() if len(lines) > 1 else '' return reset, detail def _sync_block(root, manifest): """Where this pool syncs to. The env values are what lib/common.sh RESOLVED (env override > manifest), passed through so the report can never disagree with what `sync` would actually do; without them the manifest is the authority.""" target = os.environ.get('MULTIACC_REPORT_SYNC_TARGET') if target is None: target = str(manifest.get('server') or '') sroot = os.environ.get('MULTIACC_REPORT_SYNC_ROOT') or str(manifest.get('server_root') or '') srepo = os.environ.get('MULTIACC_REPORT_SYNC_REPO') or str(manifest.get('server_repo') or '') mode = os.environ.get('MULTIACC_REPORT_SYNC_MODE') if not mode: mode = 'local' if target.strip().lower() in ('', 'none', 'local', 'off', 'disabled') else 'server' role = 'source' try: with open(os.path.join(root, 'sync-role'), errors='replace') as f: if any(line.strip().lower() == 'replica' for line in f): role = 'replica' except OSError: pass peers = [] raw = manifest.get('peers') if isinstance(raw, list): for p in raw: if isinstance(p, dict): peers.append({'target': str(p.get('target') or ''), 'root': str(p.get('root') or ''), 'repo': str(p.get('repo') or '')}) return {'mode': mode, 'target': None if mode == 'local' else target, 'root': sroot or None, 'repo': srepo or None, 'role': role, 'peers': peers} def _claude_credentials(d): """(credential_class, credentials-detail) for a claude account dir. The OAuth login is machine-local whether it sits in `.credentials.json` or in the macOS Keychain (`oauth_store` says which; `keychain` says whether THIS process could open it). A Keychain login this session cannot open is still a login on this Mac — reporting it as `none` is what made a whole pool of working accounts read as missing — so the class stays machine-local and the row's status says `locked`.""" cpath = os.path.join(d, '.credentials.json') tpath = os.path.join(d, 'server.token') has_oauth = _size(cpath) > 0 has_token = _size(tpath) > 0 detail = {'oauth': has_oauth, 'token': has_token, 'oauth_store': None, 'keychain': None, 'oauth_expires_at': None, 'oauth_refresh_expires_at': None, 'token_minted_at': None, 'token_age_days': None, # True once THIS exact token passed a real inference on this machine # (.server-token-verified); False = present but never proven here. 'token_verified': token_verified(d, tpath) if has_token else None} o = None if has_oauth: detail['oauth_store'] = 'file' doc = _load_json(cpath) or {} o = doc.get('claudeAiOauth') else: kc = keychain.probe(d) if kc['state'] in ('present', 'corrupt', 'locked'): has_oauth = True detail['oauth'] = True detail['oauth_store'] = 'keychain' detail['keychain'] = 'locked' if kc['state'] == 'locked' else 'readable' o = (kc['doc'] or {}).get('claudeAiOauth') if isinstance(o, dict): detail['oauth_expires_at'] = _iso_ms(o.get('expiresAt')) detail['oauth_refresh_expires_at'] = _iso_ms(o.get('refreshTokenExpiresAt')) # `oauth` says a sign-in EXISTS here; `oauth_alive` says whether it can still work — # the limits writer's own rule (audit.creds_doc_state). A dead one (refresh token # expired) keeps inference running on the portable token but can never read usage # or limit resets again, so the panel must ask for a sign-in (2026-09-22: acct-02 and # acct-07 sat dark for days behind `oauth: true`). None = could not be read here (a # Keychain this session cannot open) or no sign-in at all. detail['oauth_alive'] = None if has_oauth and isinstance(o, dict): try: state, reason = creds_doc_state({'claudeAiOauth': o}, time.time()) detail['oauth_alive'] = state == 'ok' if state != 'ok': detail['oauth_problem'] = reason except Exception: pass if has_token: mt = _mtime(tpath) detail['token_minted_at'] = _iso(mt) if mt: detail['token_age_days'] = int((time.time() - mt) / 86400) # A setup-token is the only claude credential that survives being copied. if has_token: return 'portable', detail if has_oauth: return 'machine-local', detail return 'none', detail def _codex_credentials(d, jwt_claims): """(credential_class, credentials-detail) for a codex account dir. Codex has NO portable class: auth.json carries a rotating refresh token.""" cpath = os.path.join(d, 'auth.json') has_auth = _size(cpath) > 0 detail = {'oauth': has_auth, 'token': False, 'auth_expires_at': None, 'refresh_token': False, 'plan': None, 'auth_mode': None} if not has_auth: return 'none', detail doc = _load_json(cpath) or {} detail['auth_mode'] = str(doc.get('auth_mode') or '') or None tokens = doc.get('tokens') if isinstance(tokens, dict): detail['refresh_token'] = bool(tokens.get('refresh_token')) detail['auth_expires_at'] = _iso(jwt_claims(tokens.get('access_token')).get('exp')) plan = (jwt_claims(tokens.get('id_token')).get('https://api.openai.com/auth') or {}) if isinstance(plan, dict): detail['plan'] = plan.get('chatgpt_plan_type') or None return 'machine-local', detail def _usage(d, stale_after): lim = _load_json(os.path.join(d, 'limits.json')) if not lim: return None buckets = [] for b in (lim.get('buckets') or []): if not isinstance(b, dict): continue buckets.append({'name': str(b.get('name') or ''), 'kind': str(b.get('kind') or ''), 'group': str(b.get('group') or ''), 'percent': b.get('percent'), 'resets_at': b.get('resets_at')}) fetched = lim.get('fetched_at') or 0 out = {'fetched_at': _iso(fetched), 'source': lim.get('source'), 'max_percent': lim.get('max_percent'), 'weekly_percent': lim.get('weekly_percent'), 'session_percent': lim.get('session_percent'), 'buckets': buckets} reset_count = lim.get('reset_credits_available') reset_stamp = _iso(lim.get('reset_credits_fetched_at')) if type(reset_count) is int and 0 <= reset_count <= 1_000_000 and reset_stamp: out.update(reset_credits_available=reset_count, reset_credits_fetched_at=reset_stamp) # A confirmed limit reset (claude): when it happened and which limit types it refilled. # Every park written before it for one of those windows is lifted — here, in the shim, # and (from this field) in the app-robot panel. Emitted only while the writer carries it. record = record_from(lim, time.time()) if record: out.update(reset_redeemed_at=_iso(record[0]), reset_cleared=list(record[1])) try: out['age_seconds'] = max(0, int(time.time() - float(fetched))) except (TypeError, ValueError): out['age_seconds'] = None # Past this window the shim stops ranking on these numbers entirely (bin/claude # STALE_AFTER). A panel that shows "weekly 2%" without it is reporting a number # nothing is using — which is how an 11-day telemetry outage stayed invisible. out['stale'] = out['age_seconds'] is None or out['age_seconds'] > stale_after # The horizon a stale weekly reading stays true until (bin/claude stale_weekly). # Absent on documents written before it existed, which is itself the signal that # the reading cannot be used for degraded ranking. if lim.get('weekly_resets_epoch'): out['weekly_resets_epoch'] = lim['weekly_resets_epoch'] # The writer found no INFORMATIVE bucket (every percentage 0 with no reset window) # and refused to invent numbers from it, so the ranking fields above are all None # here and the account ranks as unknown rather than as empty — the 2026-09-04 # incident, where fake-zero telemetry made two exhausted accounts the band leaders. # Emitted only when set, and always as a plain true, so a consumer that has never # seen the flag reads exactly the shape it always did. if lim.get('no_data'): out['no_data'] = True if lim.get('last_error'): out['last_error'] = lim['last_error'] if lim.get('last_error_at'): out['last_error_at'] = _iso(lim['last_error_at']) if lim.get('plan'): out['plan'] = lim['plan'] if lim.get('retry_after'): out['retry_after'] = _iso(lim['retry_after']) return out def _last_pick(root, aid): path = os.path.join(root, 'selection.log') last = None try: with open(path, errors='replace') as f: for line in f: parts = line.split() if len(parts) >= 2 and parts[1] == aid: last = parts[0] except OSError: pass return last def _rankable(u): """True when this reading can still rank its account in the shim: inside the ranking window AND carrying BOTH of the fields selection reads (bin/claude rank_weekly_of / rank_session_of, gated by pick_best's "known" rule and by telem_blind, which uses exactly this rule). A no_data document — all-zero buckets with no reset window, 2026-09-04 — is current and well-formed but carries neither, so a pool full of them is BLIND, not fresh; reporting it as fresh is how an outage hides behind a recent timestamp. BOTH, not either: the same writer change omits fields per signal, and a one-signal document (weekly-only or session-only) is unknown to pick_best just as completely — an either-test called such a pool fresh while the shim was tying every account and picking at random. bool is an int in Python but not a number to the shim's digit parser, so `weekly_percent: true` must read as unusable here too.""" if not u or u.get('stale'): return False return all(isinstance(u.get(k), (int, float)) and not isinstance(u.get(k), bool) for k in ('weekly_percent', 'session_percent')) def telemetry_state(accounts, now=None): """Three-state verdict on how the shim is CURRENTLY ranking this pool, over the same candidate set bin/claude uses: the eligible accounts, or — when every one of them is limit-marked — the valid ones it falls back to. 'fresh' at least one candidate has telemetry inside the ranking window that the shim can actually rank on — a COMPLETE reading, weekly and session both (see _rankable: a no_data document, or a one-signal one, is in-window and ranks nothing, so it does not make a pool fresh) 'degraded' none do, but EVERY candidate still has a weekly reading whose bucket has not reset yet, so the shim ranks on those (all-or-nothing: one unusable reading and the comparison is not apples-to-apples) 'blind' nothing usable anywhere — every account scores the same and selection is effectively a coin flip 'none' no candidates at all; nothing to say Kept here rather than beside either caller because `status` and `--json` disagreeing with the shim is exactly how an eleven-day outage stayed invisible. """ now = time.time() if now is None else now usable = [a for a in accounts if a.get('status') in ('active', 'limited')] cands = [a for a in usable if a.get('status') == 'active'] or usable if not cands: return 'none' if any(_rankable(a.get('usage')) for a in cands): return 'fresh' for a in cands: u = a.get('usage') or {} # DEGRADED means "everything is old, rank on old truths". A reading INSIDE the # window that is merely unrankable (no_data, or one-signal) blocks degraded in # the shim too (bin/claude assess_telemetry, codex review 2026-09-04): fresh # emptiness is not an outage of age, and re-fetching will not improve it. if not u.get('stale'): return 'blind' horizon = u.get('weekly_resets_epoch') or 0 pct = u.get('weekly_percent') # bool is an int in Python but not a number to the shim's digit parser, so a # document carrying `weekly_percent: true` must read as unusable HERE too — # otherwise this says 'degraded' while the shim ranks blind. if isinstance(horizon, bool) or isinstance(pct, bool) \ or not isinstance(horizon, (int, float)) or horizon <= now \ or not isinstance(pct, (int, float)): return 'blind' return 'degraded' def _account_row(root, a, aid, d, provider, jwt_claims, audit_account, now, kind, machine): """One account's row. Everything it reads is already tolerant; build() catches anything unexpected on top of that and degrades this account alone.""" st = audit_account(root, a, now=now, machine=machine) if provider == 'codex': cclass, cdetail = _codex_credentials(d, jwt_claims) else: cclass, cdetail = _claude_credentials(d) reset_epoch, marker_detail = _marker(os.path.join(d, '.limited')) limited = reset_epoch > now or (reset_epoch == 0 and os.path.isfile(os.path.join(d, '.limited'))) if limited and provider == 'claude': # The shim's reset_supersedes_marker: a park from before a confirmed limit reset, # for a window it refilled, no longer binds — even as a stale copy on a peer. try: with open(os.path.join(d, '.limited'), errors='replace') as handle: text = handle.read(4096) if marker_superseded(text, record_from(_load_json(os.path.join(d, 'limits.json')), time.time())): limited = False except OSError: pass status = st['state'] if status == 'ok': status = 'limited' if limited else 'active' row = { 'id': aid, 'email': a.get('email', ''), 'home': a.get('home', ''), 'added_at': a.get('added_at'), 'home_dir': d, 'exists': os.path.isdir(d), 'adopted': os.path.islink(d), 'status': status, 'state': st['state'], 'label': st['label'], 'reason': st['reason'], 'fix': st['fix'], 'selectable': st['state'] == 'ok' and not limited, # 'locked' is deliberately NOT a login worklist item: the login exists and works # from the Mac's own session; only the reporting process could not open it. 'needs_login': st['state'] in ('expired', 'blocked', 'missing'), 'credential_class': cclass, 'portable': cclass == 'portable', 'credentials': cdetail, # Flat copy for heartbeat consumers: a portable token this Mac has proven # (True), holds unproven (False), or does not hold (None). 'token_verified': cdetail.get('token_verified') if provider == 'claude' else None, 'limited': bool(limited), 'limit_reset_at': _iso(reset_epoch) if limited and reset_epoch else None, 'limit_reset_epoch': reset_epoch if limited and reset_epoch else None, 'limit_detail': marker_detail if limited else None, 'usage': _usage(d, _stale_after(provider)), } if kind == 'status': row['last_picked'] = _last_pick(root, aid) expired_reset, expired_detail = _marker(os.path.join(d, '.expired')) row['expired_marker'] = expired_detail or None return row def build(root, provider, machine, kind='list'): if provider == 'codex': from codex_audit import audit_account, jwt_claims else: from audit import audit_account jwt_claims = None now = time.time() warnings = [] manifest_path = os.path.join(root, 'accounts.json') manifest = _load_json(manifest_path) if manifest is None: manifest = {} warnings.append(f'manifest unreadable or missing: {manifest_path}') raw_accounts = manifest.get('accounts') if not isinstance(raw_accounts, list): raw_accounts = [] if manifest: warnings.append("manifest has no well-formed 'accounts' array") accounts = [] seen_emails = {} for a in raw_accounts: if not isinstance(a, dict) or not VALID_ID.fullmatch(str(a.get('id', ''))): warnings.append('skipped a manifest entry with a malformed id') continue aid = a['id'] d = os.path.join(root, aid) try: row = _account_row(root, a, aid, d, provider, jwt_claims, audit_account, now, kind, machine) except Exception as e: # one unreadable account must not empty the report warnings.append(f'{aid}: could not be read ({str(e)[:120]})') row = {'id': aid, 'email': a.get('email', ''), 'home': a.get('home', ''), 'home_dir': d, 'status': 'unknown', 'state': 'unknown', 'label': 'UNKNOWN', 'reason': f'unreadable: {str(e)[:120]}', 'fix': '', 'selectable': False, 'needs_login': False, 'credential_class': 'unknown', 'portable': False, 'credentials': {}, 'limited': False, 'limit_reset_at': None, 'usage': None} accounts.append(row) seen_emails.setdefault(str(a.get('email', '')).lower(), []).append(aid) for email, ids in seen_emails.items(): if len(ids) > 1: warnings.append(f"{email} is registered {len(ids)}x ({', '.join(ids)}) " f"— run '{provider}-accounts dedupe'") threshold = manifest.get('threshold', 90) try: threshold = int(threshold) except (TypeError, ValueError): threshold = 90 return { 'schema': SCHEMA, 'provider': provider, 'kind': kind, 'generated_at': _iso(now), 'machine': machine, 'host': socket.gethostname(), 'pool': { 'root': root, 'manifest': manifest_path, 'threshold': threshold, 'sync': _sync_block(root, manifest), }, 'accounts': accounts, 'summary': { 'total': len(accounts), 'active': sum(1 for a in accounts if a['status'] == 'active'), 'limited': sum(1 for a in accounts if a['status'] == 'limited'), 'needs_login': sum(1 for a in accounts if a['needs_login']), 'portable': sum(1 for a in accounts if a['portable']), 'selectable': sum(1 for a in accounts if a['selectable']), # The pool-wide verdict, so a dashboard can show the outage rather than a # grid of reassuring percentages nothing is ranking on. # fresh ranking on current data # degraded ranking on old-but-still-true weekly readings # blind every account scores the same; selection is a coin flip 'telemetry': telemetry_state(accounts, now), # Kept as its own boolean because a dashboard alerting on one field should # not have to know the vocabulary. True ONLY for the coin-flip case. 'ranking_blind': telemetry_state(accounts, now) == 'blind', }, 'warnings': warnings, } if __name__ == '__main__': root = sys.argv[1] provider = sys.argv[2] if len(sys.argv) > 2 else 'claude' machine = sys.argv[3] if len(sys.argv) > 3 else 'mac' kind = sys.argv[4] if len(sys.argv) > 4 else 'list' sys.path = [os.path.dirname(os.path.abspath(__file__))] + \ [p for p in sys.path if p not in ('', '.')] print(json.dumps(build(root, provider, machine, kind), indent=2, sort_keys=False))