from __future__ import annotations import base64 import errno import fcntl import json import os import re import shutil import sys import tempfile import uuid from contextlib import contextmanager from dataclasses import dataclass from datetime import UTC, datetime from enum import Enum from pathlib import Path from typing import Any, TypedDict from account_lock import AccountLockedError, AccountLockStore, account_key from runtime_paths import runtime_dir _LAZY_SYMBOLS = { "account_uuid_of": ("claude_credentials", "account_uuid_of"), "pinned_account_uuid": ("claude_credentials", "pinned_account_uuid"), "ROTATION_BACKUP_DIR": ("claude_health_client", "ROTATION_BACKUP_DIR"), "ClaudeTokenIdentity": ("claude_identity", "ClaudeTokenIdentity"), "PiAuthSync": ("pi_auth_sync", "PiAuthSync"), "SharedClaudeState": ("shared_claude_state", "SharedClaudeState"), "SharedCodexState": ("shared_codex_state", "SharedCodexState"), } def _lazy_symbol(name: str) -> Any: if name in globals(): return globals()[name] module_name, symbol_name = _LAZY_SYMBOLS[name] value = getattr(__import__(module_name, fromlist=[symbol_name]), symbol_name) globals()[name] = value return value def __getattr__(name: str) -> Any: if name not in _LAZY_SYMBOLS: raise AttributeError(name) return _lazy_symbol(name) AUTH_NAMESPACE = "https://api.openai.com/auth" class RegistryState(TypedDict): accounts: list[dict[str, object]] deleted_legacy_slugs: list[str] class AuthorityMode(str, Enum): NATIVE = "native" SUBROUTER_DARK = "subrouter-dark" SUBROUTER = "subrouter" @dataclass(frozen=True) class AuthorityBinding: mode: AuthorityMode authority_name: str route_id: str provider: str proxy_grant_ref: Path quiesced: bool = False def __post_init__(self) -> None: if not isinstance(self.mode, AuthorityMode): raise ValueError("authority binding mode is invalid") if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}", self.authority_name): raise ValueError("authority binding name is invalid") if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}", self.route_id): raise ValueError("authority route identifier is invalid") if self.provider not in {"claude", "codex"}: raise ValueError("authority binding provider is invalid") if not isinstance(self.proxy_grant_ref, Path) or not self.proxy_grant_ref.is_absolute(): raise ValueError("authority proxy grant reference must be absolute") if not isinstance(self.quiesced, bool): raise ValueError("authority binding quiesced state is invalid") @classmethod def from_dict(cls, value: object, *, provider: str) -> "AuthorityBinding | None": if value is None: return None if not isinstance(value, dict): raise ValueError("authority binding must be an object") required = { "mode", "authority_name", "route_id", "provider", "proxy_grant_ref", } allowed = required | {"quiesced"} if not required <= set(value) or not set(value) <= allowed: raise ValueError("authority binding has unknown or missing fields") try: mode = AuthorityMode(value["mode"]) except (TypeError, ValueError) as exc: raise ValueError("authority binding mode is invalid") from exc if mode == AuthorityMode.NATIVE: raise ValueError("native accounts must omit authority_binding") fields = { key: value[key] for key in ("authority_name", "route_id", "provider", "proxy_grant_ref") } if any(not isinstance(item, str) or not item.strip() for item in fields.values()): raise ValueError("authority binding fields must be non-empty strings") if fields["provider"] != provider: raise ValueError("authority binding provider does not match account provider") if any( marker in key.lower() for key in value for marker in ("token", "secret", "password", "key_hash") ): raise ValueError("authority binding cannot contain credential fields") quiesced = value.get("quiesced", False) if not isinstance(quiesced, bool): raise ValueError("authority binding quiesced state is invalid") grant_ref = Path(fields["proxy_grant_ref"]) if not grant_ref.is_absolute(): raise ValueError("authority proxy grant reference must be absolute") return cls( mode=mode, authority_name=fields["authority_name"], route_id=fields["route_id"], provider=fields["provider"], proxy_grant_ref=grant_ref, quiesced=quiesced, ) def as_dict(self) -> dict[str, str | bool]: return { "mode": self.mode.value, "authority_name": self.authority_name, "route_id": self.route_id, "provider": self.provider, "proxy_grant_ref": str(self.proxy_grant_ref), "quiesced": self.quiesced, } @dataclass(frozen=True) class AccountRef: tool: str slug: str @property def key(self) -> str: return f"{self.tool}:{self.slug}" @dataclass class Account: ref: AccountRef alias: str account_home: Path email: str | None plan: str | None account_id: str | None authority_binding: AuthorityBinding | None def __init__( self, *args: object, **kwargs: object, ) -> None: if args: if len(args) == 6 and isinstance(args[0], AccountRef): kwargs = { **kwargs, "ref": args[0], "alias": args[1], "account_home": args[2], "email": args[3], "plan": args[4], "account_id": args[5], } elif len(args) == 6 and isinstance(args[0], str): kwargs = { **kwargs, "slug": args[0], "alias": args[1], "codex_home": args[2], "email": args[3], "plan": args[4], "account_id": args[5], } else: raise TypeError("Account expects either AccountRef or slug-based arguments") ref = kwargs.pop("ref", None) slug = kwargs.pop("slug", None) tool = kwargs.pop("tool", None) alias = kwargs.pop("alias", None) account_home = kwargs.pop("account_home", None) codex_home = kwargs.pop("codex_home", None) email = kwargs.pop("email", None) plan = kwargs.pop("plan", None) account_id = kwargs.pop("account_id", None) authority_binding = kwargs.pop("authority_binding", None) if kwargs: unexpected = ", ".join(sorted(kwargs)) raise TypeError(f"Account got unexpected arguments: {unexpected}") if ref is None: if not isinstance(slug, str) or not slug: raise TypeError("Account requires ref or slug") ref = AccountRef(tool if isinstance(tool, str) and tool else "codex", slug) if not isinstance(ref, AccountRef): raise TypeError("Account ref must be an AccountRef") if account_home is None: account_home = codex_home if account_home is None: raise TypeError("Account requires account_home or codex_home") if not isinstance(alias, str) or not alias: raise TypeError("Account requires alias") if not isinstance(account_home, Path): account_home = Path(account_home) self.ref = ref self.alias = alias self.account_home = account_home self.email = email if isinstance(email, str) or email is None else str(email) self.plan = plan if isinstance(plan, str) or plan is None else str(plan) self.account_id = ( account_id if isinstance(account_id, str) or account_id is None else str(account_id) ) if authority_binding is not None: if not isinstance(authority_binding, AuthorityBinding): raise TypeError("Account authority_binding must be an AuthorityBinding") if authority_binding.mode == AuthorityMode.NATIVE: raise ValueError("native accounts must omit authority_binding") if authority_binding.provider != ref.tool: raise ValueError("authority binding provider does not match account provider") self.authority_binding = authority_binding @property def slug(self) -> str: return self.ref.slug @property def tool(self) -> str: return self.ref.tool @property def codex_home(self) -> Path: return self.account_home @property def tray_key(self) -> str: return self.ref.key class AccountRegistryKind(Enum): CODEX = "codex" CLAUDE = "claude" GROK = "grok" PROTECTED_GROK_SLUG = "roy-grok" RESERVED_SLUGS = {"dynamic"} ROUTING_RULES_FILENAMES: dict[AccountRegistryKind, str] = { AccountRegistryKind.CODEX: "routing_rules.json", AccountRegistryKind.CLAUDE: "claude_routing_rules.json", } HEALTH_CACHE_FILENAMES: dict[AccountRegistryKind, str] = { AccountRegistryKind.CODEX: "health_cache.json", AccountRegistryKind.CLAUDE: "claude_health_cache.json", AccountRegistryKind.GROK: "grok_health_cache.json", } @dataclass class StagedAccount: registry: "AccountRegistry" alias: str slug: str account_home: Path _staged_dir: Path _closed: bool = False def commit(self) -> Account: self._ensure_open() return self.registry._commit_staged_account(self) def rollback(self) -> None: self._ensure_open() self.registry._rollback_staged_account(self) def _ensure_open(self) -> None: if self._closed: raise RuntimeError("staged account already closed") def _mark_closed(self) -> None: self._closed = True class AccountRegistry: def __init__( self, base_dir: Path | None = None, legacy_codex_home: Path | None = None, kind: AccountRegistryKind = AccountRegistryKind.CODEX, legacy_claude_home: Path | None = None, active_claude_json: Path | None = None, legacy_grok_home: Path | None = None, pi_agent_dir: Path | None = None, ) -> None: user_home = Path.home() self.base_dir = base_dir if base_dir is not None else runtime_dir() self.legacy_codex_home = ( legacy_codex_home if legacy_codex_home is not None else user_home / ".codex" ) self.legacy_claude_home = ( legacy_claude_home if legacy_claude_home is not None else user_home / ".claude" ) self.legacy_grok_home = ( legacy_grok_home if legacy_grok_home is not None else user_home / ".grok" ) self.active_claude_json = ( active_claude_json if active_claude_json is not None else self.legacy_claude_home.parent / ".claude.json" ) self.kind = kind if kind == AccountRegistryKind.CLAUDE: shared_state_cls = _lazy_symbol("SharedClaudeState") self.accounts_dir = self.base_dir / "claude-accounts" self.registry_path = self.base_dir / "claude_accounts.json" self.default_path = self.base_dir / "claude_default_slug" self._account_home_name = "CLAUDE_HOME" self._shared_codex_state = shared_state_cls( shared_home=self.legacy_claude_home, accounts_dir=self.accounts_dir, ) elif kind == AccountRegistryKind.GROK: self.accounts_dir = self.base_dir / "grok-accounts" self.registry_path = self.base_dir / "grok_accounts.json" self.default_path = self.base_dir / "grok_default_slug" self._account_home_name = "GROK_HOME" self._shared_codex_state = None else: shared_state_cls = _lazy_symbol("SharedCodexState") self.accounts_dir = self.base_dir / "accounts" self.registry_path = self.base_dir / "accounts.json" self.default_path = self.base_dir / "default_slug" self._account_home_name = "CODEX_HOME" self._shared_codex_state = shared_state_cls( legacy_codex_home=self.legacy_codex_home, accounts_dir=self.accounts_dir, ) if pi_agent_dir is not None and kind == AccountRegistryKind.CODEX: pi_auth_sync_cls = _lazy_symbol("PiAuthSync") self._pi_auth_sync = pi_auth_sync_cls( agent_dir=pi_agent_dir, accounts_dir=self.accounts_dir, ) else: self._pi_auth_sync = None routing_rules_filename = ROUTING_RULES_FILENAMES.get(kind) self.routing_rules_path = ( None if routing_rules_filename is None else self.base_dir / routing_rules_filename ) self.health_cache_path = self.base_dir / HEALTH_CACHE_FILENAMES[kind] self._account_locks = AccountLockStore(self.base_dir) self._lock_path = self.accounts_dir / ".registry.lock" self.base_dir.mkdir(parents=True, exist_ok=True) self.accounts_dir.mkdir(parents=True, exist_ok=True) def migrate_legacy(self) -> None: if self.kind == AccountRegistryKind.CLAUDE: self._migrate_legacy_claude() return registry = self._read_registry() known_slugs = {entry["slug"] for entry in registry["accounts"]} deleted_legacy_slugs = set(registry["deleted_legacy_slugs"]) for legacy_auth in sorted(self.legacy_codex_home.glob("auth.*.json")): slug = legacy_auth.name[len("auth.") : -len(".json")] if ( slug in known_slugs or slug in deleted_legacy_slugs or (self.accounts_dir / slug).exists() ): continue account_home = self.accounts_dir / slug / "CODEX_HOME" account_home.mkdir(parents=True, exist_ok=False) shutil.copy2(legacy_auth, account_home / "auth.json") self.sync_shared_links(account_home) registry["accounts"].append({"slug": slug, "alias": slug}) known_slugs.add(slug) self._write_registry(registry) if self.default_path.exists(): return current_auth_payload = self._decode_auth_file(self.legacy_codex_home / "auth.json") current_account_id = current_auth_payload.get("account_id") if current_account_id is None: return for entry in registry["accounts"]: payload = self._decode_auth_file( self.accounts_dir / entry["slug"] / "CODEX_HOME" / "auth.json" ) if payload.get("account_id") == current_account_id: self._atomic_write_text(self.default_path, f"{entry['slug']}\n") return def _migrate_legacy_claude(self) -> None: credentials_path = self.legacy_claude_home / ".credentials.json" if not credentials_path.exists(): return registry = self._read_registry() if registry["accounts"]: if not self.default_path.exists(): first_slug = registry["accounts"][0].get("slug") if isinstance(first_slug, str) and first_slug: self._atomic_write_text(self.default_path, f"{first_slug}\n") return try: credentials_realpath = credentials_path.resolve() accounts_realpath = self.accounts_dir.resolve() if credentials_realpath == accounts_realpath or credentials_realpath.is_relative_to( accounts_realpath ): return except OSError: return metadata = self._decode_claude_file(self.active_claude_json) alias = self._legacy_claude_alias(metadata) existing_entry = next( (entry for entry in registry["accounts"] if entry.get("alias") == alias), None, ) slug = existing_entry["slug"] if existing_entry is not None else self.new_slug(alias) known_slugs = {entry["slug"] for entry in registry["accounts"]} if slug not in known_slugs and not (self.accounts_dir / slug).exists(): claude_home = self.accounts_dir / slug / "CLAUDE_HOME" claude_home.mkdir(parents=True, exist_ok=False) shutil.copy2(credentials_path, claude_home / ".credentials.json") if self.active_claude_json.exists(): shutil.copy2(self.active_claude_json, claude_home / "claude.json") registry["accounts"].append({"slug": slug, "alias": alias}) self._write_registry(registry) if not self.default_path.exists(): self._atomic_write_text(self.default_path, f"{slug}\n") def list(self) -> list[Account]: entries = self._read_registry()["accounts"] if self.kind == AccountRegistryKind.CLAUDE: entries = [entry for entry in entries if self._should_list_claude_account(entry)] return [self._account_from_entry(entry) for entry in entries] def set_authority_binding( self, slug: str, binding: AuthorityBinding | None ) -> Account: if binding is not None and binding.provider != self.kind.value: raise ValueError("authority binding provider does not match registry") if binding is not None and binding.mode == AuthorityMode.NATIVE: raise ValueError("native accounts must omit authority_binding") with self._account_root_lock(): registry = self._read_registry() for entry in registry["accounts"]: if entry.get("slug") != slug: continue if binding is None: entry.pop("authority_binding", None) else: entry["authority_binding"] = binding.as_dict() self._write_registry(registry) return self._account_from_entry(entry) raise KeyError(slug) def default_slug(self) -> str | None: if not self.default_path.exists(): return None slug = self.default_path.read_text(encoding="utf-8").strip() if not slug: return None known = {account.slug for account in self.list()} return slug if slug in known else None def is_locked(self, slug: str) -> bool: return self._account_locks.is_locked(self.kind.value, slug) def locked_slugs(self) -> frozenset[str]: return self._account_locks.locked_slugs(self.kind.value) def set_locked(self, slug: str, locked: bool) -> bool: with self._account_root_lock(): known_slugs = {entry["slug"] for entry in self._read_registry()["accounts"]} if slug not in known_slugs: raise KeyError(slug) changed = self._account_locks.set_locked(self.kind.value, slug, locked) if not changed: return False if self._read_default_pointer() == slug: if locked: # Keep the lock even when active-credential cleanup fails: failing # closed in every supported launcher is safer than silently reopening # the account. self._deactivate_locked_account(slug) else: try: self._sync_active_links(slug) except Exception as exc: relock_error: Exception | None = None cleanup_error: Exception | None = None try: self._account_locks.set_locked(self.kind.value, slug, True) except Exception as restore_exc: relock_error = restore_exc try: self._deactivate_locked_account(slug) except Exception as cleanup_exc: cleanup_error = cleanup_exc if relock_error is not None: raise RuntimeError( "failed to restore account lock after activation failed" ) from relock_error if cleanup_error is not None and hasattr(exc, "add_note"): exc.add_note( "account was re-locked, but active credential cleanup also " f"failed: {cleanup_error}" ) raise self._touch_ratelimit_wake() return True def set_default(self, account: Account) -> None: with self._account_root_lock(): if self.is_locked(account.slug): raise AccountLockedError(self.kind.value, (account.slug,)) previous = self._read_default_pointer() self._write_default_slug(account.slug) try: self._sync_active_links(account.slug) except Exception: self._write_default_slug(previous) self._sync_active_links(previous) raise def rename(self, slug: str, new_alias: str) -> str: new_alias = new_alias.strip() if not new_alias: raise ValueError("account label must not be empty") with self._account_root_lock(): registry = self._read_registry() entry = next((e for e in registry["accounts"] if e["slug"] == slug), None) if entry is None: raise KeyError(slug) self._reject_name_collision(new_alias, registry["accounts"], own_slug=slug) new_slug = ( slug if self._slug_is_pinned(slug) else self.new_slug(new_alias, own_slug=slug) ) if new_slug == slug: entry["alias"] = new_alias self._write_registry(registry) return slug self._reslug(registry, slug, new_slug, new_alias) return new_slug def _slug_is_pinned(self, slug: str) -> bool: return ( self.kind == AccountRegistryKind.GROK and slug == PROTECTED_GROK_SLUG and os.environ.get("SYSTRAY_ALLOW_ROY_GROK_MUTATION") != "1" ) def _reslug( self, registry: RegistryState, slug: str, new_slug: str, new_alias: str, ) -> None: account_dir = self.accounts_dir / slug target_dir = self.accounts_dir / new_slug if target_dir.exists() or target_dir.is_symlink(): raise FileExistsError(f"account directory already in use: {new_slug}") next_registry: RegistryState = { "accounts": [ {**e, "slug": new_slug, "alias": new_alias} if e["slug"] == slug else e for e in registry["accounts"] ], "deleted_legacy_slugs": [ deleted for deleted in registry["deleted_legacy_slugs"] if deleted != new_slug ], } previous_default = self._read_default_pointer() next_default = new_slug if previous_default == slug else previous_default sidecars = self._read_sidecar_state() lock_snapshot = self._account_locks.locked_keys() old_lock_key = account_key(self.kind.value, slug) new_lock_key = account_key(self.kind.value, new_slug) lock_states = { old_lock_key: old_lock_key in lock_snapshot, new_lock_key: new_lock_key in lock_snapshot, } moved = account_dir.exists() if moved: os.replace(account_dir, target_dir) try: self._account_locks.rekey(self.kind.value, slug, new_slug) self._write_account_state(next_registry, next_default) self._rekey_sidecar_state(slug, new_slug) except Exception: if moved: os.replace(target_dir, account_dir) self._restore_sidecar_state(sidecars) self._account_locks.restore_key_states(lock_states) self._write_account_state(registry, previous_default) raise def _sidecar_paths(self) -> tuple[Path, ...]: paths = [self.health_cache_path] if self.routing_rules_path is not None: paths.append(self.routing_rules_path) return tuple(paths) def _read_sidecar_state(self) -> dict[Path, str | None]: state: dict[Path, str | None] = {} for path in self._sidecar_paths(): try: state[path] = path.read_text(encoding="utf-8") except OSError: state[path] = None return state def _restore_sidecar_state(self, state: dict[Path, str | None]) -> None: for path, content in state.items(): if content is None: self._remove_path(path) else: self._atomic_write_text(path, content) def _rekey_sidecar_state(self, slug: str, new_slug: str) -> None: self._rekey_health_cache(slug, new_slug) self._rekey_routing_rules(slug, new_slug) def _rekey_health_cache(self, slug: str, new_slug: str) -> None: payload = self._load_json_object(self.health_cache_path) if slug not in payload: return payload[new_slug] = payload.pop(slug) self._atomic_write_text( self.health_cache_path, json.dumps(payload, sort_keys=True) ) def _rekey_routing_rules(self, slug: str, new_slug: str) -> None: if self.routing_rules_path is None: return rules = self._load_json_object(self.routing_rules_path) if not rules: return changed = False projects = rules.get("projects") if isinstance(projects, dict): for route in projects.values(): if isinstance(route, dict) and route.get("account") == slug: route["account"] = new_slug changed = True if rules.get("default") == slug: rules["default"] = new_slug changed = True chain = rules.get("fallback_chain") if isinstance(chain, list) and slug in chain: rules["fallback_chain"] = [new_slug if item == slug else item for item in chain] changed = True caps = rules.get("account_caps") if isinstance(caps, dict) and slug in caps: caps[new_slug] = caps.pop(slug) changed = True if changed: self._atomic_write_text( self.routing_rules_path, json.dumps(rules, indent=2) + "\n" ) def _drop_routing_rules_account(self, slug: str) -> None: if self.routing_rules_path is None: return rules = self._load_json_object(self.routing_rules_path) caps = rules.get("account_caps") if rules else None if not isinstance(caps, dict) or slug not in caps: return caps.pop(slug) self._atomic_write_text( self.routing_rules_path, json.dumps(rules, indent=2) + "\n" ) def _reject_name_collision( self, name: str, accounts: list[dict[str, str]], *, own_slug: str | None = None, ) -> None: cf = name.casefold() if cf in {reserved.casefold() for reserved in RESERVED_SLUGS}: raise ValueError(f"reserved account name: {name}") for entry in accounts: if entry["slug"] == own_slug: continue if entry["slug"].casefold() == cf or entry["alias"].casefold() == cf: raise FileExistsError(f"account name already in use: {name}") def resolve_profile_token(self, token: str) -> str | None: token = token.strip() if not token: return None accounts = self._read_registry()["accounts"] for entry in accounts: if entry["slug"] == token: return entry["slug"] cf = token.casefold() for entry in accounts: if entry["alias"].casefold() == cf: return entry["slug"] return None def authority_binding_for(self, slug: str) -> AuthorityBinding | None: """Read only registry binding metadata; never open provider credential files.""" if not isinstance(slug, str) or not slug: raise KeyError(slug) for entry in self._read_registry()["accounts"]: if entry.get("slug") != slug: continue return AuthorityBinding.from_dict( entry.get("authority_binding"), provider=self.kind.value ) raise KeyError(slug) def remove(self, slug: str) -> None: if ( self.kind == AccountRegistryKind.GROK and slug == PROTECTED_GROK_SLUG and os.environ.get("SYSTRAY_ALLOW_ROY_GROK_MUTATION") != "1" ): raise PermissionError( "roy-grok is protected; set SYSTRAY_ALLOW_ROY_GROK_MUTATION=1 to remove" ) with self._account_root_lock(): registry = self._read_registry() current_accounts = registry["accounts"] new_accounts = [entry for entry in current_accounts if entry["slug"] != slug] if len(new_accounts) == len(current_accounts): raise KeyError(slug) account_dir = self.accounts_dir / slug quarantine_dir = self.accounts_dir / f".{slug}.quarantine-{uuid.uuid4().hex}" previous_default = self._read_default_pointer() next_default = self._select_default_slug(new_accounts, previous_default, removed_slug=slug) next_registry = { **registry, "accounts": new_accounts, "deleted_legacy_slugs": ( sorted({*registry["deleted_legacy_slugs"], slug}) if self.kind == AccountRegistryKind.CODEX else registry["deleted_legacy_slugs"] ), } sidecars = self._read_sidecar_state() lock_key = account_key(self.kind.value, slug) lock_was_set = lock_key in self._account_locks.locked_keys() os.replace(account_dir, quarantine_dir) try: self._write_account_state(next_registry, next_default) self._drop_routing_rules_account(slug) self._account_locks.drop(self.kind.value, slug) except Exception: if quarantine_dir.exists() and not account_dir.exists(): os.replace(quarantine_dir, account_dir) self._restore_sidecar_state(sidecars) self._account_locks.restore_key_states({lock_key: lock_was_set}) self._write_account_state(registry, previous_default) raise shutil.rmtree(quarantine_dir) def normalize_slug(self, raw: str) -> str: base = re.sub(r"-+", "-", re.sub(r"[^a-z0-9]+", "-", raw.strip().lower())).strip("-") if not base: raise ValueError("account slug cannot be empty") return base def new_slug(self, alias: str, own_slug: str | None = None) -> str: try: base = self.normalize_slug(alias) except ValueError: base = "account" if base in RESERVED_SLUGS: raise ValueError(f"reserved account slug: {base}") known = { entry["slug"] for entry in self._read_registry()["accounts"] if entry["slug"] != own_slug } if base not in known: return base suffix = 2 while f"{base}-{suffix}" in known: suffix += 1 return f"{base}-{suffix}" def add_dir(self, slug: str, alias: str) -> Path: if slug in RESERVED_SLUGS: raise ValueError(f"reserved account slug: {slug}") with self._account_root_lock(): registry = self._read_registry() self._reject_name_collision(alias, registry["accounts"], own_slug=slug) account_home = self.accounts_dir / slug / self._account_home_name account_home.mkdir(parents=True, exist_ok=False) if self.kind == AccountRegistryKind.CODEX: self.sync_shared_links(account_home) registry["accounts"].append({"slug": slug, "alias": alias}) registry["deleted_legacy_slugs"] = [ deleted_slug for deleted_slug in registry["deleted_legacy_slugs"] if deleted_slug != slug ] self._write_registry(registry) return account_home def stage_add(self, alias: str, slug: str | None = None) -> StagedAccount: with self._account_root_lock(): registry = self._read_registry() if slug is None: slug = self.new_slug(alias) else: slug = self.normalize_slug(slug) if slug in RESERVED_SLUGS: raise ValueError(f"reserved account slug: {slug}") known_slugs = {entry["slug"] for entry in registry["accounts"]} if slug in known_slugs or (self.accounts_dir / slug).exists(): raise FileExistsError(slug) self._reject_name_collision(alias, registry["accounts"], own_slug=slug) staged_dir = Path(tempfile.mkdtemp(prefix=f".{slug}.staged-", dir=self.accounts_dir)) account_home = staged_dir / self._account_home_name try: account_home.mkdir(parents=True, exist_ok=False) if self.kind == AccountRegistryKind.CODEX: self.sync_shared_links(account_home) except Exception: shutil.rmtree(staged_dir, ignore_errors=True) raise return StagedAccount( registry=self, alias=alias, slug=slug, account_home=account_home, _staged_dir=staged_dir, ) def sync_shared_links(self, account_home: Path) -> list[Path]: if self._shared_codex_state is None: return [] return self._shared_codex_state.sync_shared_links(account_home) def sync_all_shared_links(self) -> list[Path]: if self._shared_codex_state is None: return [] return self._shared_codex_state.sync_all_shared_links() def _read_registry(self) -> RegistryState: if not self.registry_path.exists(): return {"accounts": [], "deleted_legacy_slugs": []} data = json.loads(self.registry_path.read_text(encoding="utf-8")) accounts = data.get("accounts") deleted_legacy_slugs = data.get("deleted_legacy_slugs", []) if not isinstance(accounts, list) or not isinstance(deleted_legacy_slugs, list): return {"accounts": [], "deleted_legacy_slugs": []} return { "accounts": list(accounts), "deleted_legacy_slugs": [ slug for slug in deleted_legacy_slugs if isinstance(slug, str) ], } def _write_registry(self, registry: RegistryState) -> None: self._atomic_write_text( self.registry_path, json.dumps(registry, indent=2, sort_keys=False) + "\n", ) def _account_from_entry(self, entry: dict[str, object]) -> Account: slug = entry.get("slug") alias = entry.get("alias") if not isinstance(slug, str) or not slug or not isinstance(alias, str) or not alias: raise ValueError("account registry entry is invalid") account_home = self.accounts_dir / slug / self._account_home_name payload = self._read_account_metadata(account_home, slug) return Account( ref=AccountRef(self.kind.value, slug), alias=alias, account_home=account_home, email=payload.get("email"), plan=payload.get("plan"), account_id=payload.get("account_id"), authority_binding=AuthorityBinding.from_dict( entry.get("authority_binding"), provider=self.kind.value ), ) def _should_list_claude_account(self, entry: dict[str, object]) -> bool: slug = entry.get("slug") if not isinstance(slug, str) or not slug: return False account_home = self.accounts_dir / slug / self._account_home_name credentials_path = account_home / ".credentials.json" try: payload = json.loads(credentials_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return True if not self._contains_token_key(payload): return True if self._contains_non_empty_token(payload): return True metadata = self._read_account_metadata(account_home, slug) return any(value is not None for value in metadata.values()) @staticmethod def _contains_token_key(value: object) -> bool: if isinstance(value, dict): for key, nested in value.items(): if isinstance(key, str) and "token" in key.lower(): return True if AccountRegistry._contains_token_key(nested): return True if isinstance(value, list): return any(AccountRegistry._contains_token_key(item) for item in value) return False @staticmethod def _contains_non_empty_token(value: object) -> bool: if isinstance(value, dict): for key, nested in value.items(): if isinstance(key, str) and "token" in key.lower(): if isinstance(nested, str) and nested.strip(): return True if AccountRegistry._contains_non_empty_token(nested): return True if isinstance(value, list): return any(AccountRegistry._contains_non_empty_token(item) for item in value) return False def _read_account_metadata( self, account_home: Path, slug: str ) -> dict[str, str | None]: if self.kind == AccountRegistryKind.CLAUDE: identity_payload = self._decode_account_identity_file( account_home / "account_identity.json" ) legacy_payload = self._decode_claude_file(account_home / "claude.json") return { key: identity_payload.get(key) if identity_payload.get(key) is not None else legacy_payload.get(key) for key in ("email", "plan", "account_id") } if self.kind == AccountRegistryKind.GROK: identity_payload = self._decode_account_identity_file( account_home / "account_identity.json" ) if any(identity_payload.get(key) for key in ("email", "plan", "account_id")): return identity_payload return self._decode_grok_auth_file(account_home / "auth.json", slug=slug) return self._decode_auth_file(account_home / "auth.json", slug=slug) def _decode_grok_auth_file(self, path: Path, slug: str | None = None) -> dict[str, str | None]: try: payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError("not an object") for entry in payload.values(): if not isinstance(entry, dict): continue email = entry.get("email") account_id = entry.get("principal_id") or entry.get("user_id") return { "email": email if isinstance(email, str) else None, "plan": "grok", "account_id": account_id if isinstance(account_id, str) else None, } raise ValueError("no auth entry") except Exception as exc: if slug is not None: print(f"failed to decode grok auth for account '{slug}': {exc}", file=sys.stderr) return {"email": None, "plan": "grok", "account_id": None} def _decode_auth_file(self, path: Path, slug: str | None = None) -> dict[str, str | None]: try: payload = json.loads(path.read_text(encoding="utf-8")) token = payload["tokens"]["id_token"] parts = token.split(".") if len(parts) < 2: raise ValueError("invalid jwt") decoded = self._decode_jwt_payload(parts[1]) auth_claims = decoded.get(AUTH_NAMESPACE) if not isinstance(auth_claims, dict): auth_claims = {} plan = self._normalize_codex_plan( auth_claims.get("chatgpt_plan_type"), auth_claims.get("plan"), auth_claims.get("subscription_type"), auth_claims.get("subscriptionType"), decoded.get("plan"), ) return { "email": decoded.get("email"), "plan": plan, "account_id": auth_claims.get("chatgpt_account_id"), } except Exception as exc: if slug is not None: print(f"failed to decode auth.json for account '{slug}': {exc}", file=sys.stderr) return {"email": None, "plan": None, "account_id": None} def _decode_account_identity_file(self, path: Path) -> dict[str, str | None]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {"email": None, "plan": None, "account_id": None} if not isinstance(payload, dict): return {"email": None, "plan": None, "account_id": None} email = payload.get("email") account_id = payload.get("org_id") plan = payload.get("subscription_type") return { "email": email if isinstance(email, str) else None, "plan": plan if isinstance(plan, str) else None, "account_id": account_id if isinstance(account_id, str) else None, } def _decode_claude_file(self, path: Path) -> dict[str, str | None]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {"email": None, "plan": None, "account_id": None} if not isinstance(payload, dict): return {"email": None, "plan": None, "account_id": None} oauth_account = payload.get("oauthAccount") if not isinstance(oauth_account, dict): oauth_account = {} email = oauth_account.get("emailAddress") if not isinstance(email, str): email = payload.get("emailAddress") plan = oauth_account.get("organizationRateLimitTier") if not isinstance(plan, str): plan = oauth_account.get("seatTier") if not isinstance(plan, str): plan = payload.get("plan") account_id = oauth_account.get("accountUuid") if not isinstance(account_id, str): account_id = payload.get("userID") return { "email": email if isinstance(email, str) else None, "plan": plan if isinstance(plan, str) else None, "account_id": account_id if isinstance(account_id, str) else None, } _FIRST_RUN_SUPPRESSOR_KEYS = ( "hasCompletedOnboarding", "lastOnboardingVersion", "migrationVersion", "officialMarketplaceAutoInstallAttempted", "officialMarketplaceAutoInstalled", "opusProMigrationComplete", "sonnet1m45MigrationComplete", "hasResetAutoModeOptInForDefaultOffer", "installMethod", "autoUpdates", "autoUpdatesProtectedForNative", ) def _seed_first_run_flags(self, account_home: Path) -> bool: # A fresh Claude login writes only minimal metadata into the account's # .claude.json. Activating such an account makes Claude Code treat it as # a first run and rewrite the SHARED ~/.claude/settings.json, dropping # user-authored hooks. Pre-seed the identity-free "already set up" flags # from the live global config so the first launch is a normal launch. if self.kind != AccountRegistryKind.CLAUDE: return False source = self._load_json_object(self.active_claude_json) flags = { key: source[key] for key in self._FIRST_RUN_SUPPRESSOR_KEYS if key in source } target = self._claude_metadata_path(account_home) if target.exists(): try: loaded = json.loads(target.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return False if not isinstance(loaded, dict): return False payload = loaded else: payload = {} merged = {**flags, **payload, "hasCompletedOnboarding": True} if merged == payload: return True self._atomic_write_text(target, json.dumps(merged, indent=2) + "\n") return True def _backup_claude_settings(self) -> None: settings = self.legacy_claude_home / "settings.json" if not settings.exists(): return backups_dir = self.legacy_claude_home / "backups" backups_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%S.%fZ") shutil.copyfile(settings, backups_dir / f"settings.json.{timestamp}") backups = sorted(backups_dir.glob("settings.json.*"), reverse=True) for stale in backups[5:]: stale.unlink() @staticmethod def _load_json_object(path: Path) -> dict: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} return payload if isinstance(payload, dict) else {} def _legacy_claude_alias(self, metadata: dict[str, str | None]) -> str: email = metadata.get("email") if isinstance(email, str) and email.strip(): return email.split("@", 1)[0] return "legacy" @staticmethod def _normalize_codex_plan(*candidates: object) -> str | None: for candidate in candidates: if not isinstance(candidate, str): continue normalized = candidate.strip() if not normalized: continue if normalized.lower() == "profile": continue return normalized return None @staticmethod def _decode_jwt_payload(segment: str) -> dict: padding = "=" * (-len(segment) % 4) data = base64.urlsafe_b64decode(segment + padding) return json.loads(data.decode("utf-8")) @staticmethod def _atomic_write_text(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=path.parent, delete=False, ) as handle: handle.write(content) temp_name = handle.name os.replace(temp_name, path) @staticmethod def _atomic_symlink(source: Path, target: Path) -> None: if not source.exists(): raise FileNotFoundError(source) target.parent.mkdir(parents=True, exist_ok=True) temp_path = target.parent / f".{target.name}.tmp-{os.getpid()}" if temp_path.exists() or temp_path.is_symlink(): temp_path.unlink() temp_path.symlink_to(source) os.replace(temp_path, target) def _commit_staged_account(self, staged: StagedAccount) -> Account: self._seed_first_run_flags(staged.account_home) final_dir = self.accounts_dir / staged.slug with self._account_root_lock(): registry = self._read_registry() known_slugs = {entry["slug"] for entry in registry["accounts"]} if staged.slug in known_slugs or final_dir.exists(): raise FileExistsError(staged.slug) self._reject_name_collision( staged.alias, registry["accounts"], own_slug=staged.slug, ) os.replace(staged._staged_dir, final_dir) try: registry["accounts"].append({"slug": staged.slug, "alias": staged.alias}) registry["deleted_legacy_slugs"] = [ deleted_slug for deleted_slug in registry["deleted_legacy_slugs"] if deleted_slug != staged.slug ] self._write_registry(registry) except Exception: os.replace(final_dir, staged._staged_dir) raise staged._mark_closed() return self._account_from_entry({"slug": staged.slug, "alias": staged.alias}) def _rollback_staged_account(self, staged: StagedAccount) -> None: shutil.rmtree(staged._staged_dir, ignore_errors=True) staged._mark_closed() @contextmanager def _account_root_lock(self) -> Path: self.accounts_dir.mkdir(parents=True, exist_ok=True) with self._lock_path.open("a+", encoding="utf-8") as handle: fcntl.flock(handle.fileno(), fcntl.LOCK_EX) try: yield self._lock_path finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) def _write_account_state(self, registry: RegistryState, default_slug: str | None) -> None: default_locked = default_slug is not None and self.is_locked(default_slug) self._write_registry(registry) self._write_default_slug(default_slug) if default_locked: self._deactivate_locked_account(default_slug) else: self._sync_active_links(default_slug) def _write_default_slug(self, slug: str | None) -> None: if slug is None: self._remove_path(self.default_path) return self._atomic_write_text(self.default_path, f"{slug}\n") def _sync_active_links(self, slug: str | None) -> None: if slug is None: self._clear_active_links() return if self.is_locked(slug): self._clear_active_links() return account_home = self.accounts_dir / slug / self._account_home_name if self.kind == AccountRegistryKind.CLAUDE: self._seed_first_run_flags(account_home) self._backup_claude_settings() self._preserve_live_claude_grant() self._atomic_symlink( account_home / ".credentials.json", self.legacy_claude_home / ".credentials.json", ) claude_json = self._claude_metadata_path(account_home) if claude_json.exists(): self._atomic_symlink(claude_json, self.active_claude_json) else: self._remove_path(self.active_claude_json, symlink_only=True) self._touch_ratelimit_wake() return if self.kind == AccountRegistryKind.GROK: self._activate_grok_auth(account_home / "auth.json") self._touch_ratelimit_wake() return pi_target = None if self._pi_auth_sync is not None: # Fallible derivation runs before any active-link mutation, so a # failure leaves both codex and pi pointing at the old account. pi_target = self._pi_auth_sync.prepare(slug) codex_link = self.legacy_codex_home / "auth.json" previous_codex_target = codex_link.readlink() if codex_link.is_symlink() else None self._atomic_symlink(account_home / "auth.json", codex_link) if pi_target is not None: try: self._pi_auth_sync.activate(pi_target) except Exception: if previous_codex_target is not None: self._atomic_symlink(previous_codex_target, codex_link) raise self._touch_ratelimit_wake() def _preserve_live_claude_grant(self) -> None: """Hand a vendor-written live grant back to its account before a symlink replaces it. The Claude CLI replaces the activation symlink with a regular file whenever it writes tokens, so the grant it rotates lives only in the live home until the next activation. """ live = self.legacy_claude_home / ".credentials.json" if live.is_symlink() or not live.is_file(): return pinned = _lazy_symbol("pinned_account_uuid") token_identity_cls = _lazy_symbol("ClaudeTokenIdentity") try: content = live.read_text(encoding="utf-8") except OSError: return owner_uuid = token_identity_cls().account_uuid_for(live) if owner_uuid is None: return for account in self.list(): if pinned(account.account_home) != owner_uuid: continue target = account.account_home / ".credentials.json" self._atomic_write_text(target, content) os.chmod(target, 0o600) return # No pinned account claims this grant; rescue it — refresh tokens are # one-time-use, so letting the symlink replace the live file would # destroy the only copy. rescue_dir = _lazy_symbol("ROTATION_BACKUP_DIR") rescue = rescue_dir / f"unclaimed-{owner_uuid}.credentials.json" self._atomic_write_text(rescue, content) os.chmod(rescue, 0o600) def _activate_grok_auth(self, account_auth: Path) -> None: """Point ~/.grok/auth.json at account auth without losing a real file. If legacy auth is a real file (not a symlink), it is left in place only when content already matches the account file; otherwise we refuse to clobber unless the account file was populated from that same content. """ if not account_auth.is_file(): raise FileNotFoundError(account_auth) legacy = self.legacy_grok_home / "auth.json" self.legacy_grok_home.mkdir(parents=True, exist_ok=True) if legacy.exists() and not legacy.is_symlink(): # Real file: only replace with symlink if byte-identical to account copy. try: if legacy.read_bytes() != account_auth.read_bytes(): raise RuntimeError( "refusing to replace real ~/.grok/auth.json that differs from " f"account auth at {account_auth}; copy-import first" ) except OSError as exc: raise RuntimeError(f"cannot compare grok auth files: {exc}") from exc self._atomic_symlink(account_auth, legacy) def _touch_ratelimit_wake(self) -> None: # mega-plan-harness runs paused on a provider rate limit watch this # file; a new active account may restore capacity, so wake them all # to re-probe immediately instead of sleeping out their timers. try: (self.base_dir / "ratelimit-wake").touch() except OSError: pass def _clear_active_links(self) -> None: if self.kind == AccountRegistryKind.CLAUDE: self._remove_path(self.legacy_claude_home / ".credentials.json", symlink_only=True) self._remove_path(self.active_claude_json, symlink_only=True) return if self.kind == AccountRegistryKind.GROK: self._remove_path(self.legacy_grok_home / "auth.json", symlink_only=True) return self._remove_path(self.legacy_codex_home / "auth.json", symlink_only=True) if self._pi_auth_sync is not None: self._pi_auth_sync.clear_active() def _deactivate_locked_account(self, slug: str) -> None: """Remove live credentials while preserving vendor-rotated regular files. The vendor CLIs may replace an activation symlink with a regular credential file. Merely removing symlinks would leave that account directly usable after it was locked, so a regular live file is atomically moved back into the selected account before the live path is removed. """ account_home = self.accounts_dir / slug / self._account_home_name if self.kind == AccountRegistryKind.CLAUDE: self._deactivate_live_path( self.legacy_claude_home / ".credentials.json", account_home / ".credentials.json", ) self._deactivate_live_path( self.active_claude_json, self._claude_metadata_path(account_home), ) self._touch_ratelimit_wake() return if self.kind == AccountRegistryKind.GROK: self._deactivate_live_path( self.legacy_grok_home / "auth.json", account_home / "auth.json", ) self._touch_ratelimit_wake() return self._deactivate_live_path( self.legacy_codex_home / "auth.json", account_home / "auth.json", ) if self._pi_auth_sync is not None: self._pi_auth_sync.clear_active() self._touch_ratelimit_wake() @staticmethod def _deactivate_live_path(active: Path, account_copy: Path) -> None: if active.is_symlink(): active.unlink() return if not active.exists(): return if not active.is_file(): raise RuntimeError(f"refusing to remove non-file active credential path: {active}") account_copy.parent.mkdir(parents=True, exist_ok=True) try: os.replace(active, account_copy) except OSError as exc: if exc.errno != errno.EXDEV: raise temporary = account_copy.parent / f".{account_copy.name}.lock-move-{os.getpid()}" try: shutil.copy2(active, temporary) os.replace(temporary, account_copy) active.unlink() finally: if temporary.exists() or temporary.is_symlink(): temporary.unlink() os.chmod(account_copy, 0o600) @staticmethod def _claude_metadata_path(account_home: Path) -> Path: dot_config = account_home / ".claude.json" if AccountRegistry._has_claude_oauth_metadata(dot_config): return dot_config return account_home / "claude.json" @staticmethod def _has_claude_oauth_metadata(path: Path) -> bool: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return False if not isinstance(payload, dict): return False oauth_account = payload.get("oauthAccount") return isinstance(oauth_account, dict) def _read_default_pointer(self) -> str | None: if not self.default_path.exists(): return None slug = self.default_path.read_text(encoding="utf-8").strip() return slug or None def _select_default_slug( self, accounts: list[dict[str, str]], previous_default: str | None, removed_slug: str, ) -> str | None: remaining_slugs = {entry["slug"] for entry in accounts} if previous_default == removed_slug or previous_default not in remaining_slugs: return accounts[0]["slug"] if accounts else None return previous_default @staticmethod def _remove_path(path: Path, symlink_only: bool = False) -> None: if symlink_only and not path.is_symlink(): return if path.exists() or path.is_symlink(): path.unlink()