#!/usr/bin/env python3 from __future__ import annotations import base64 import ctypes import errno import json import os import re import shutil import signal import stat import socket import subprocess import sys import tempfile import time import urllib.error import urllib.request from collections.abc import Iterator from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any, Literal from account_lock import ( AccountLockError, AccountUseAuthorization, HumanOverrideError, authorize_account_use, revalidate_account_use, ) from account_registry import AccountRegistry, AuthorityBinding, AuthorityMode from authority_client import AuthorityConfigurationError, resolve_authority_data_plane from command_router import ( ClaudeAdapter, CodexAdapter, CommandRouter, account_spend_cap_block, ) from model_catalog import MODEL_ALIASES from routing_resolver import AllAccountsCappedError from runtime_paths import runtime_dir PROXY_MODEL = "gpt-5.6-sol" SMALL_MODEL = "gpt-5.6-luna" PROXY_SCOPE_MARKER = "CLAUDEX_PROXY_SCOPE" PROXY_SLICE = "claudex-proxy.slice" DANGEROUS_PERMISSIONS_FLAG = "--dangerously-skip-permissions" NATIVE_MODEL_ALIASES = { "fable": "claude-fable-5", "opus": "claude-opus-5", "sonnet": "claude-sonnet-5", } SEAT_MODEL_ALIASES = dict(MODEL_ALIASES) SEAT_UNSUPPORTED_BOUNDARY = ( "unsupported-security-boundary: remote seats accept only GPT-backed models " "(sol, terra, luna, gpt-*)" ) SEAT_GPT_MODEL_RE = re.compile(r"^gpt-[A-Za-z0-9][A-Za-z0-9._-]*$") SEAT_NATIVE_MODEL_RE = re.compile(r"^claude-") SEAT_BINDING_REL = Path(".local/state/overdeck/seat-bindings") SEAT_BINDING_MAX_BYTES = 8192 SEAT_BINDING_REQUIRED_FIELDS = ( "account", "seatId", "host", "model", "credentialFingerprint", ) SEAT_CONTROL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]") SEAT_FORWARDED_EXACT_DENY = frozenset({ "-m", "--model", "--fallback-model", "--settings", "--setting-sources", "--disallowedTools", "--allowedTools", "--disallowed-tools", "--allowed-tools", "--tools", "--agents", "--strict-mcp-config", "--mcp-config", "--plugin", "--plugins", "--plugin-dir", "--plugin-url", "--marketplace", "--enable-workflows", "--enable-workflow", "--agent", "--workflow", "--sandbox", "--no-sandbox", "Agent", "Workflow", }) SEAT_FORWARDED_PREFIX_DENY = ( "-m=", "--model=", "--fallback-model=", "--settings=", "--setting-sources=", "--disallowedTools=", "--allowedTools=", "--disallowed-tools=", "--allowed-tools=", "--tools=", "--agents=", "--strict-mcp-config=", "--mcp-config=", "--plugin=", "--plugins=", "--plugin-dir=", "--plugin-url=", "--marketplace=", "--enable-workflows=", "--enable-workflow=", "--agent=", "--workflow=", "--sandbox=", "--no-sandbox=", ) AVAILABLE_MODELS = tuple(MODEL_ALIASES.values()) NATIVE_PICKER_MODELS = tuple(NATIVE_MODEL_ALIASES.values()) HYBRID_CODEX_MODELS = AVAILABLE_MODELS HYBRID_PICKER_MODELS = (*NATIVE_MODEL_ALIASES, *HYBRID_CODEX_MODELS) HYBRID_MODEL = PROXY_MODEL HYBRID_SMALL_MODEL = SMALL_MODEL PROXY_PICKER_SLOTS = ( ("OPUS", "sol", "GPT-5.6 Sol"), ("SONNET", "terra", "GPT-5.6 Terra"), ("HAIKU", "luna", "GPT-5.6 Luna"), ) CODEX_MODEL_LABELS = { MODEL_ALIASES[alias]: label for _, alias, label in PROXY_PICKER_SLOTS } PROXY_SLOT_VARIABLES = tuple( f"ANTHROPIC_DEFAULT_{slot}_MODEL{suffix}" for slot, _, _ in PROXY_PICKER_SLOTS for suffix in ("", "_NAME", "_DESCRIPTION", "_SUPPORTED_CAPABILITIES") ) class ClaudexError(RuntimeError): exit_code = 1 class MissingBinaryError(ClaudexError): exit_code = 127 class CredentialConflictError(ClaudexError): pass class ClaudexCappedError(ClaudexError): exit_code = 75 def __init__(self, resume_at: str | None, tool: str = "Codex") -> None: super().__init__(f"claudex: selected {tool} account is spend-capped") self.resume_at = resume_at def refusal_json(self) -> str: payload: dict[str, object] = { "ok": False, "detail": "rate-limited", "reason": "all-accounts-capped", } if self.resume_at is not None: payload["resume_at"] = self.resume_at return json.dumps(payload, separators=(",", ":")) @dataclass(frozen=True, slots=True) class SelectedAccount: slug: str native_auth_path: Path proxy_config_dir: Path base_dir: Path | None = None authorization: AccountUseAuthorization | None = None authority_binding: AuthorityBinding | None = None def _codex_tray_dirs() -> tuple[Path, Path]: return runtime_dir(), Path.home() / ".codex-tray" def _codex_account_candidates(token: str) -> tuple[str, ...]: try: resolved = AccountRegistry().resolve_profile_token(token) except (OSError, ValueError, KeyError): resolved = None if resolved is None or resolved == token: return (token,) return (token, resolved) def _selected_candidate(base_dir: Path, slug: str) -> SelectedAccount | None: auth_path = base_dir / "accounts" / slug / "CODEX_HOME" / "auth.json" binding: AuthorityBinding | None = None try: binding = AccountRegistry(base_dir=base_dir).authority_binding_for(slug) except (KeyError, OSError, ValueError, json.JSONDecodeError): binding = None if binding is not None and binding.mode == AuthorityMode.SUBROUTER: return SelectedAccount( slug, auth_path, auth_path.parent.parent / "CLAUDEX_HOME", base_dir, authority_binding=binding, ) if auth_path.is_file(): return SelectedAccount( slug, auth_path, auth_path.parent.parent / "CLAUDEX_HOME", base_dir, authority_binding=binding, ) return None def _authority_selected(selected: SelectedAccount) -> bool: binding = selected.authority_binding return binding is not None and binding.mode == AuthorityMode.SUBROUTER def _requested_codex_account( slug: str, tray_dirs: tuple[Path, ...] | None = None ) -> SelectedAccount: candidate_dirs = _codex_tray_dirs() if tray_dirs is None else tray_dirs for candidate in _codex_account_candidates(slug): for base_dir in candidate_dirs: selected = _selected_candidate(base_dir, candidate) if selected is not None: return selected raise ClaudexError(f"claudex: no Systray AI Codex account named {slug!r}") def _require_codex_spend_headroom(slug: str) -> None: blocked = account_spend_cap_block(CodexAdapter(), slug) if blocked is not None: raise ClaudexCappedError(blocked.resume_at) def _validate_human_override_selector(requested_slug: str | None) -> None: if requested_slug is None: raise HumanOverrideError( "--human-override-lock requires an explicit --account" ) parts = tuple(part.strip() for part in requested_slug.split(",") if part.strip()) if len(parts) != 1: raise HumanOverrideError( "--human-override-lock requires exactly one explicit account per provider" ) def resolve_selected_account( requested_slug: str | None = None, human_override_lock: bool = False, ) -> SelectedAccount: if human_override_lock: _validate_human_override_selector(requested_slug) tray_dirs = _codex_tray_dirs() lock_policy_dir = tray_dirs[0] if requested_slug is not None: requested = _requested_codex_account(requested_slug, tray_dirs) _require_codex_spend_headroom(requested.slug) if requested.base_dir is None: return requested authorization = authorize_account_use( lock_policy_dir, "codex", requested.slug, allow_locked=human_override_lock, explicit_account=True, ) return SelectedAccount( requested.slug, requested.native_auth_path, requested.proxy_config_dir, lock_policy_dir, authorization, requested.authority_binding, ) candidates: list[tuple[float, int, SelectedAccount]] = [] for index, base_dir in enumerate(tray_dirs): default_path = base_dir / "default_slug" try: slug = default_path.read_text(encoding="utf-8").strip() mtime = default_path.stat().st_mtime except OSError: continue selected = _selected_candidate(base_dir, slug) if slug else None if selected is not None: candidates.append((mtime, -index, selected)) if not candidates: raise ClaudexError("claudex: no valid Systray AI default Codex account") _, _, selected = max(candidates, key=lambda item: (item[0], item[1])) slug = selected.slug auth_path = selected.native_auth_path base_dir = selected.base_dir assert base_dir is not None _require_codex_spend_headroom(slug) authorization = authorize_account_use( lock_policy_dir, "codex", slug, allow_locked=human_override_lock, explicit_account=False, ) account_dir = auth_path.parent.parent return SelectedAccount( slug, auth_path, account_dir / "CLAUDEX_HOME", lock_policy_dir, authorization, selected.authority_binding, ) def _jwt_exp_ms(token: str) -> int: try: segment = token.split(".")[1] padding = "=" * (-len(segment) % 4) payload = json.loads(base64.urlsafe_b64decode(segment + padding)) exp = payload["exp"] except (IndexError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: raise ClaudexError("claudex: Codex access token has no valid expiry") from exc if not isinstance(exp, int) or exp <= 0: raise ClaudexError("claudex: Codex access token has no valid expiry") return exp * 1000 def _required_string(value: object, field: str) -> str: if not isinstance(value, str) or not value.strip(): raise ClaudexError(f"claudex: missing Codex credential field '{field}'") return value def native_to_proxy_auth(native: dict[str, object]) -> dict[str, object]: tokens = native.get("tokens") if not isinstance(tokens, dict): raise ClaudexError("claudex: invalid native Codex auth format") access = _required_string(tokens.get("access_token"), "tokens.access_token") refresh = _required_string(tokens.get("refresh_token"), "tokens.refresh_token") account_id = _required_string(tokens.get("account_id"), "tokens.account_id") return { "access": access, "refresh": refresh, "expires": _jwt_exp_ms(access), "accountId": account_id, } def _read_json_object(path: Path) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ClaudexError(f"claudex: invalid credential file: {path}") from exc if not isinstance(payload, dict): raise ClaudexError(f"claudex: invalid credential file: {path}") return payload def _atomic_write_json(path: Path, payload: dict[str, object]) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(path.parent, 0o700) with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=path.parent, delete=False ) as handle: json.dump(payload, handle, separators=(",", ":")) handle.write("\n") temp_path = Path(handle.name) os.chmod(temp_path, 0o600) os.replace(temp_path, path) def _export_proxy_to_native( native_path: Path, native: dict[str, Any], proxy: dict[str, Any], ) -> None: tokens = native.get("tokens") if not isinstance(tokens, dict): raise ClaudexError("claudex: invalid native Codex auth format") tokens["access_token"] = _required_string(proxy.get("access"), "access") tokens["refresh_token"] = _required_string(proxy.get("refresh"), "refresh") tokens["account_id"] = _required_string(proxy.get("accountId"), "accountId") native["last_refresh"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) _atomic_write_json(native_path, native) def reconcile_credentials(native_path: Path, proxy_path: Path) -> None: native = _read_json_object(native_path) native_proxy = native_to_proxy_auth(native) if not proxy_path.exists(): _atomic_write_json(proxy_path, native_proxy) return proxy = _read_json_object(proxy_path) proxy_access = _required_string(proxy.get("access"), "access") proxy_refresh = _required_string(proxy.get("refresh"), "refresh") proxy_account = _required_string(proxy.get("accountId"), "accountId") if proxy_account != native_proxy["accountId"]: raise CredentialConflictError("claudex: proxy credential belongs to another account") native_exp_value = native_proxy["expires"] if not isinstance(native_exp_value, int): raise ClaudexError("claudex: invalid native Codex token expiry") native_exp = native_exp_value proxy_exp = _jwt_exp_ms(proxy_access) native_refresh = str(native_proxy["refresh"]) if native_refresh != proxy_refresh and native_exp == proxy_exp: raise CredentialConflictError( "claudex: divergent Codex refresh tokens; re-authenticate selected Systray AI account" ) if proxy_exp > native_exp: _export_proxy_to_native(native_path, native, proxy) elif native_exp > proxy_exp or proxy_access != native_proxy["access"]: _atomic_write_json(proxy_path, native_proxy) def find_available_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server: server.bind(("127.0.0.1", 0)) return int(server.getsockname()[1]) @dataclass(frozen=True, slots=True) class AccountOverrides: codex: str | None claude: str | None human_override_lock: bool = False SESSION_ID_PATTERN = re.compile(r"[0-9a-fA-F-]{8,}") ACCOUNT_FLAGS = { "--account": "both", "--profile": "both", "--codex-account": "codex", "--codex-profile": "codex", "--claude-account": "claude", "--claude-profile": "claude", } def parse_account_overrides(argv: list[str]) -> tuple[AccountOverrides, list[str]]: chosen: dict[str, str] = {} forwarded: list[str] = [] index = 0 human_override_lock = False while index < len(argv): argument = argv[index] if argument == "--human-override-lock": human_override_lock = True index += 1 continue if argument.startswith("--human-override-lock="): raise ClaudexError("claudex: --human-override-lock does not take a value") flag, _, inline = argument.partition("=") provider = ACCOUNT_FLAGS.get(flag) if provider is None: forwarded.append(argument) index += 1 continue if inline: chosen[provider] = inline index += 1 continue if argument.endswith("=") or index + 1 >= len(argv): raise ClaudexError(f"claudex: {flag} requires an account") chosen[provider] = argv[index + 1] index += 2 both = chosen.get("both") overrides = AccountOverrides( chosen.get("codex", both), chosen.get("claude", both), human_override_lock, ) return overrides, forwarded def parse_seat_account_overrides(argv: list[str]) -> tuple[AccountOverrides, list[str]]: if "--" not in argv: return parse_account_overrides(argv) separator_index = argv.index("--") accounts, wrapper_argv = parse_account_overrides(argv[:separator_index]) return accounts, [*wrapper_argv, *argv[separator_index:]] def resolve_claudex_model_token(token: str) -> str: aliases = {**MODEL_ALIASES, **NATIVE_MODEL_ALIASES} resolved = aliases.get(token) if resolved is not None: return resolved if token and not any(separator in token for separator in ("-", ".", "/")): raise ClaudexError( f"claudex: unknown model alias '{token}' " f"(valid aliases: {', '.join(aliases)})" ) return token def parse_model(argv: list[str]) -> tuple[str | None, list[str]]: model: str | None = None forwarded: list[str] = [] index = 0 known_models = {*AVAILABLE_MODELS, *NATIVE_PICKER_MODELS} while index < len(argv): argument = argv[index] if argument in {"-m", "--model"} and index + 1 < len(argv): width, token = 2, argv[index + 1] elif argument.startswith("-m=") or argument.startswith("--model="): width, token = 1, argument.split("=", 1)[1] else: forwarded.append(argument) index += 1 continue resolved = resolve_claudex_model_token(token) if resolved in known_models: model = resolved elif argument.startswith("-m"): forwarded.extend(["--model", resolved]) else: forwarded.extend(argv[index : index + width]) index += width return model, forwarded def proxy_picker_env() -> dict[str, str]: picker: dict[str, str] = {} for slot, alias, label in PROXY_PICKER_SLOTS: model = MODEL_ALIASES[alias] picker[f"ANTHROPIC_DEFAULT_{slot}_MODEL"] = model picker[f"ANTHROPIC_DEFAULT_{slot}_MODEL_NAME"] = label picker[f"ANTHROPIC_DEFAULT_{slot}_MODEL_DESCRIPTION"] = ( f"{label} through claude-code-proxy" ) return picker def default_claude_args(argv: list[str]) -> list[str]: if DANGEROUS_PERMISSIONS_FLAG in argv: return argv return [DANGEROUS_PERMISSIONS_FLAG, *argv] def model_settings_args(argv: list[str], settings_path: Path) -> list[str]: return [argv[0], "--settings", str(settings_path), *argv[1:]] def enter_proxy_slice() -> None: if os.environ.get(PROXY_SCOPE_MARKER) == "1": return systemd_run = shutil.which("systemd-run") if systemd_run is None: raise MissingBinaryError("claudex: required binary not found: systemd-run") command = [ systemd_run, "--user", "--scope", "--quiet", "--collect", "--same-dir", f"--slice={PROXY_SLICE}", f"--unit=claudex-proxy-{os.getpid()}.scope", f"--setenv={PROXY_SCOPE_MARKER}=1", sys.argv[0], *sys.argv[1:], ] os.execv(systemd_run, command) def _set_parent_death_signal(expected_parent_pid: int) -> None: libc = ctypes.CDLL(None, use_errno=True) if libc.prctl(1, signal.SIGTERM) != 0: error = ctypes.get_errno() raise OSError(error, os.strerror(error)) if os.getppid() != expected_parent_pid: os._exit(1) def wait_for_proxy(process: subprocess.Popen[str], port: int) -> None: deadline = time.monotonic() + 10 url = f"http://127.0.0.1:{port}/healthz" while time.monotonic() < deadline: if process.poll() is not None: raise ClaudexError("claudex: proxy exited before readiness") try: with urllib.request.urlopen(url, timeout=0.25) as response: if response.status == 200: return except (OSError, urllib.error.URLError): time.sleep(0.05) raise ClaudexError("claudex: proxy readiness timeout") def _require_binary(name: str) -> str: # The shim is what carries the launch into confine.sh, so resolving past it starts # the agent uncapped. An inherited PATH that omits the shim dir resolves past it, # so the installed shim is preferred over the PATH entry rather than assumed to be it. shim = Path.home() / ".claude" / "bin" / name if shim.is_file() and os.access(shim, os.X_OK): if shim.resolve().name == "_tmpjail-shim.sh": return str(shim) path = shutil.which(name) if path is None: raise MissingBinaryError(f"claudex: required binary not found: {name}") return path def _terminate(process: subprocess.Popen[str]) -> None: if process.poll() is not None: return process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) @dataclass(frozen=True, slots=True) class ClaudeAccount: slug: str home: Path env: dict[str, str] base_dir: Path | None = None authorization: AccountUseAuthorization | None = None @dataclass(frozen=True, slots=True) class SeatWrapperModel: model: str argv: list[str] def claude_account_exists(slug: str) -> bool: router = CommandRouter(ClaudeAdapter()) for candidate_dir in router.adapter.candidate_tray_dirs(): resolved_slug = router.adapter.resolve_profile_token(candidate_dir, slug) if resolved_slug is None: continue if router.adapter.account_home(candidate_dir, resolved_slug).exists(): return True return False def pinned_account_slug(accounts: AccountOverrides) -> str | None: if accounts.codex is None or accounts.claude is None: return None if accounts.codex != accounts.claude: return None return accounts.codex def should_run_codex_only_for_pinned_account(accounts: AccountOverrides) -> bool: slug = pinned_account_slug(accounts) if slug is None: return False return not claude_account_exists(slug) def resolve_credentials(account_home: Path) -> Any: from claude_credentials import resolve_credentials as resolve return resolve(account_home) def resolve_claude_account( requested_slug: str | None = None, human_override_lock: bool = False, ) -> ClaudeAccount: if human_override_lock: _validate_human_override_selector(requested_slug) router = CommandRouter(ClaudeAdapter()) if human_override_lock: resolved = router._resolve_account( requested_slug, notify_fallback=requested_slug is None, ignore_locks=True, ) else: resolved = router._resolve_account( requested_slug, notify_fallback=requested_slug is None ) if resolved is None: if requested_slug is not None: raise ClaudexError(f"claudex: no Systray AI Claude account named {requested_slug!r}") raise ClaudexError("claudex: no valid Systray AI default Claude account") if isinstance(resolved, AllAccountsCappedError): raise ClaudexCappedError(resolved.resume_at, "Claude") base_dir, slug, account_home = resolved lock_policy_dir = router.adapter.lock_policy_dir() authorization = authorize_account_use( lock_policy_dir, "claude", slug, allow_locked=human_override_lock, explicit_account=requested_slug is not None, ) adapter = router.adapter try: conflicts = adapter.sync_before_exec(base_dir) or [] except (OSError, RuntimeError) as exc: conflicts = [] print(f"claudex: shared state sync skipped: {exc}", file=sys.stderr) for conflict in conflicts: print(f"claudex: kept per-account copy of {conflict}", file=sys.stderr) revalidate_account_use(lock_policy_dir, authorization) env = dict(adapter.build_env(account_home)) env["CLAUDE_SECURESTORAGE_CONFIG_DIR"] = str(resolve_credentials(account_home).path.parent) return ClaudeAccount(slug, account_home, env, lock_policy_dir, authorization) def resolve_claude_account_slug(requested_slug: str | None = None) -> str: return resolve_claude_account(requested_slug).slug def existing_claude_account_slug(requested_slug: str | None = None) -> str | None: router = CommandRouter(ClaudeAdapter()) adapter = router.adapter if requested_slug is not None: for candidate_dir in adapter.candidate_tray_dirs(): resolved_slug = adapter.resolve_profile_token(candidate_dir, requested_slug) if resolved_slug is None: continue if adapter.account_home(candidate_dir, resolved_slug).exists(): return resolved_slug return None selected_default = adapter.selected_default(adapter.candidate_tray_dirs()) if selected_default is None: return None if adapter.account_home(selected_default.base_dir, selected_default.slug).exists(): return selected_default.slug return None def seat_binding_path(account_slug: str, home: Path | None = None) -> Path: root = Path.home() if home is None else home return root / SEAT_BINDING_REL / f"{account_slug}.json" def _seat_binding_open_flags() -> int | None: flags = os.O_RDONLY if hasattr(os, "O_CLOEXEC"): flags |= os.O_CLOEXEC if hasattr(os, "O_NONBLOCK"): flags |= os.O_NONBLOCK if not hasattr(os, "O_NOFOLLOW"): return None flags |= os.O_NOFOLLOW return flags def load_seat_binding( account_slug: str, home: Path | None = None ) -> dict[str, object] | Literal["invalid"] | None: path = seat_binding_path(account_slug, home) flags = _seat_binding_open_flags() if flags is None: return "invalid" fd = -1 raw = b"" try: fd = os.open(str(path), flags) try: st = os.fstat(fd) except OSError: return "invalid" if stat.S_ISLNK(st.st_mode): return "invalid" if not stat.S_ISREG(st.st_mode): return "invalid" if st.st_size > SEAT_BINDING_MAX_BYTES: return "invalid" try: raw = os.read(fd, SEAT_BINDING_MAX_BYTES + 1) except OSError: return "invalid" if len(raw) > SEAT_BINDING_MAX_BYTES: return "invalid" except FileNotFoundError: return None except OSError as exc: if exc.errno == errno.ENOENT: return None return "invalid" finally: if fd >= 0: try: os.close(fd) except OSError: pass try: text = raw.decode("utf-8", errors="strict") except UnicodeDecodeError: return "invalid" try: payload = json.loads(text) except json.JSONDecodeError: return "invalid" if not isinstance(payload, dict): return "invalid" for field in SEAT_BINDING_REQUIRED_FIELDS: value = payload.get(field) if not isinstance(value, str) or not value.strip(): return "invalid" return payload def enforce_local_seat_binding_gate( account_slug: str | None, home: Path | None = None ) -> None: if account_slug is None: return binding = load_seat_binding(account_slug, home) if binding is None: return if binding == "invalid": raise ClaudexError( "claudex: invalid seat binding record; use claudex --seat or repair binding" ) raise ClaudexError( f"claudex: account {account_slug!r} is seat-bound; use claudex --seat" ) def parse_seat_flag(argv: list[str]) -> tuple[bool, list[str]]: if not argv or argv[0] != "--seat": return False, argv if len(argv) > 1 and argv[1].startswith("--seat"): raise ClaudexError("claudex: unsupported seat flag form") return True, argv[1:] def resolve_seat_model_token(token: str) -> str: raw = token.strip() if not raw: raise ClaudexError("seat-remote: --model is required") if raw in NATIVE_MODEL_ALIASES or SEAT_NATIVE_MODEL_RE.match(raw): raise ClaudexError(SEAT_UNSUPPORTED_BOUNDARY) if raw in SEAT_MODEL_ALIASES: return SEAT_MODEL_ALIASES[raw] if SEAT_GPT_MODEL_RE.fullmatch(raw): return raw if not any(separator in raw for separator in ("-", ".", "/")): resolve_claudex_model_token(raw) raise ClaudexError(f"seat-remote: unknown seat model {raw}") def validate_seat_forwarded_argv(argv: list[str]) -> None: for index, argument in enumerate(argv): if not isinstance(argument, str): raise ClaudexError(f"claudex: forwarded argv not string at index {index}") if SEAT_CONTROL_RE.search(argument): raise ClaudexError(f"claudex: forwarded control character at index {index}") if argument in SEAT_FORWARDED_EXACT_DENY: raise ClaudexError(f"claudex: forwarded security arg denied: {argument}") if any(argument.startswith(prefix) for prefix in SEAT_FORWARDED_PREFIX_DENY): raise ClaudexError(f"claudex: forwarded security arg denied: {argument}") def parse_seat_wrapper_model(argv: list[str]) -> SeatWrapperModel: explicit_forwarded: list[str] = [] if "--" in argv: separator_index = argv.index("--") wrapper_argv = argv[:separator_index] explicit_forwarded = argv[separator_index + 1 :] else: wrapper_argv = list(argv) model: str | None = None duplicate = False forwarded: list[str] = [] index = 0 while index < len(wrapper_argv): argument = wrapper_argv[index] if argument in {"-m", "--model"}: if index + 1 >= len(wrapper_argv): raise ClaudexError("seat-remote: --model is required") if model is not None: duplicate = True model = wrapper_argv[index + 1] index += 2 continue if argument.startswith("-m=") or argument.startswith("--model="): if model is not None: duplicate = True model = argument.split("=", 1)[1] index += 1 continue if argument == "--fallback-model" or argument.startswith("--fallback-model="): raise ClaudexError("claudex: --fallback-model is forbidden for --seat") forwarded.append(argument) index += 1 forwarded.extend(explicit_forwarded) if duplicate: raise ClaudexError("claudex: duplicate wrapper --model for --seat") if model is None: raise ClaudexError("seat-remote: --model is required") resolved = resolve_seat_model_token(model) validate_seat_forwarded_argv(forwarded) return SeatWrapperModel(resolved, forwarded) def validate_seat_resume_argv(argv: list[str]) -> None: message = "claudex: --seat requires an explicit session id (resume or --resume )" if "--continue" in argv or "-c" in argv: raise ClaudexError(message) index = 0 while index < len(argv): argument = argv[index] if argument == "resume": if index + 1 >= len(argv) or argv[index + 1].startswith("-"): raise ClaudexError(message) index += 2 continue if argument == "--resume": if index + 1 >= len(argv): raise ClaudexError(message) index += 2 continue index += 1 def resolve_seat_remote_binary() -> str: path = shutil.which("seat-remote") if path is not None: return path canonical = Path.home() / ".claude" / "bin" / "seat-remote" if canonical.is_file() and os.access(canonical, os.X_OK): return str(canonical) raise MissingBinaryError("claudex: required binary not found: seat-remote") def _resolve_selected_for_overrides(accounts: AccountOverrides) -> SelectedAccount: override = accounts.human_override_lock and accounts.codex is not None if override: return resolve_selected_account(accounts.codex, True) return resolve_selected_account(accounts.codex) def _resolve_claude_for_overrides(accounts: AccountOverrides) -> ClaudeAccount: override = accounts.human_override_lock and accounts.claude is not None if override: return resolve_claude_account(accounts.claude, True) return resolve_claude_account(accounts.claude) def _revalidate_selected_account(account: SelectedAccount) -> None: if account.base_dir is not None and account.authorization is not None: revalidate_account_use(account.base_dir, account.authorization) def _revalidate_claude_account(account: ClaudeAccount) -> None: if account.base_dir is not None and account.authorization is not None: revalidate_account_use(account.base_dir, account.authorization) def run_seat(argv: list[str], accounts: AccountOverrides) -> int: wrapper = parse_seat_wrapper_model(argv) validate_seat_resume_argv(wrapper.argv) account = _resolve_claude_for_overrides(accounts) _revalidate_claude_account(account) project = Path.cwd() binary = resolve_seat_remote_binary() command = [ binary, "launch", "--host", "auto", "--account", account.slug, "--project", str(project), "--model", wrapper.model, "--", *wrapper.argv, ] _revalidate_claude_account(account) completed = subprocess.run(command, check=False) return completed.returncode def run_native(model: str, argv: list[str], accounts: AccountOverrides) -> int: account = _resolve_claude_for_overrides(accounts) _revalidate_claude_account(account) account_home = account.home native_env = dict(os.environ) for variable in ( "ANTHROPIC_API_KEY", "ANTHROPIC_BASE_URL", "ANTHROPIC_AUTH_TOKEN", *PROXY_SLOT_VARIABLES, "ANTHROPIC_DEFAULT_FABLE_MODEL", "CLAUDE_CODE_SUBAGENT_MODEL", "CCP_CONFIG_DIR", "CCP_BIND_ADDRESS", "CCP_CODEX_SERVICE_TIER", ): native_env.pop(variable, None) native_env.update(account.env) settings_path = account_home / "claudex-anthropic-settings.json" _atomic_write_json(settings_path, {"availableModels": list(NATIVE_PICKER_MODELS)}) native_argv = model_settings_args(default_claude_args(argv), settings_path) _revalidate_claude_account(account) result = subprocess.run( [_require_binary("claude"), "--model", model, *native_argv], env=native_env, check=False, ) return result.returncode @contextmanager def codex_proxy(selected: SelectedAccount) -> Iterator[int]: _revalidate_selected_account(selected) enter_proxy_slice() proxy_binary = _require_binary("claude-code-proxy") port = find_available_port() parent_pid = os.getpid() if _authority_selected(selected): binding = selected.authority_binding if binding is None or selected.base_dir is None: raise ClaudexError("claudex: Gateway: migration required") if binding.quiesced: raise ClaudexError("claudex: Gateway: migration paused") try: data_plane = resolve_authority_data_plane(selected.base_dir, binding) except AuthorityConfigurationError as exc: raise ClaudexError(f"claudex: {exc}") from exc if data_plane.sanitized_status.state != "ready": raise ClaudexError(f"claudex: {data_plane.sanitized_status.detail}") with tempfile.TemporaryDirectory(prefix="claudex-authority-") as temporary: config_dir = Path(temporary) os.chmod(config_dir, 0o700) auth_path = config_dir / "codex" / "auth.json" _atomic_write_json( auth_path, { "access": data_plane.proxy_key, "refresh": "", "expires": data_plane.grant_expires_at_ms, }, ) proxy_env = { **os.environ, "CCP_CONFIG_DIR": str(config_dir), "CCP_BIND_ADDRESS": "127.0.0.1", "CCP_CODEX_BASE_URL": data_plane.responses_url, "CCP_CODEX_SERVICE_TIER": "fast", "PORT": str(port), } log_path = config_dir / "proxy.log" with log_path.open("a", encoding="utf-8") as proxy_log: _revalidate_selected_account(selected) proxy = subprocess.Popen( [proxy_binary, "serve", "--no-monitor"], env=proxy_env, stdin=subprocess.DEVNULL, stdout=proxy_log, stderr=proxy_log, text=True, preexec_fn=lambda: _set_parent_death_signal(parent_pid), ) try: wait_for_proxy(proxy, port) yield port finally: _terminate(proxy) return proxy_auth_path = selected.proxy_config_dir / "codex" / "auth.json" reconcile_credentials(selected.native_auth_path, proxy_auth_path) proxy_env = { **os.environ, "CCP_CONFIG_DIR": str(selected.proxy_config_dir), "CCP_BIND_ADDRESS": "127.0.0.1", "CCP_CODEX_SERVICE_TIER": "fast", "PORT": str(port), } log_path = selected.proxy_config_dir / "proxy.log" log_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with log_path.open("a", encoding="utf-8") as proxy_log: _revalidate_selected_account(selected) proxy = subprocess.Popen( [proxy_binary, "serve", "--no-monitor"], env=proxy_env, stdin=subprocess.DEVNULL, stdout=proxy_log, stderr=proxy_log, text=True, preexec_fn=lambda: _set_parent_death_signal(parent_pid), ) try: wait_for_proxy(proxy, port) yield port finally: _terminate(proxy) reconcile_credentials(selected.native_auth_path, proxy_auth_path) def proxy_claude_env(selected: SelectedAccount, model: str, port: int) -> dict[str, str]: claude_env = dict(os.environ) claude_env.pop("ANTHROPIC_API_KEY", None) for variable in ( *PROXY_SLOT_VARIABLES, "ANTHROPIC_DEFAULT_FABLE_MODEL", "CLAUDE_CODE_SUBAGENT_MODEL", "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC", "CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY", ): claude_env.pop(variable, None) claude_env.update({ "ANTHROPIC_BASE_URL": f"http://127.0.0.1:{port}", "ANTHROPIC_AUTH_TOKEN": "unused", "ANTHROPIC_MODEL": model, "ANTHROPIC_SMALL_FAST_MODEL": SMALL_MODEL, **proxy_picker_env(), **_shared_claude_env(selected), }) return claude_env def hybrid_claude_env( selected: SelectedAccount, account: ClaudeAccount, base_url: str ) -> dict[str, str]: claude_env = dict(os.environ) for variable in ( "ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", *PROXY_SLOT_VARIABLES, "ANTHROPIC_DEFAULT_FABLE_MODEL", "CLAUDE_CODE_SUBAGENT_MODEL", "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC", ): claude_env.pop(variable, None) claude_env.update({ **account.env, "ANTHROPIC_BASE_URL": base_url, "CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY": "1", "ANTHROPIC_MODEL": HYBRID_MODEL, "ANTHROPIC_SMALL_FAST_MODEL": HYBRID_SMALL_MODEL, **_shared_claude_env(selected), }) return claude_env def gateway_catalog_models() -> list[tuple[str, str]]: """Codex catalog entries; entries after the session model never render.""" others = [item for item in CODEX_MODEL_LABELS.items() if item[0] != HYBRID_MODEL] return [*others, (HYBRID_MODEL, CODEX_MODEL_LABELS[HYBRID_MODEL])] def write_gateway_models(account_home: Path, base_url: str) -> None: _atomic_write_json( account_home / "cache" / "gateway-models.json", { "baseUrl": base_url, "fetchedAt": int(time.time() * 1000), "models": [ {"id": model, "display_name": label} for model, label in gateway_catalog_models() ], }, ) def _shared_claude_env(selected: SelectedAccount) -> dict[str, str]: environment = { "CLAUDE_CODE_ALWAYS_ENABLE_EFFORT": "1", "CLAUDE_CODE_MAX_TOOL_USE_CONCURRENCY": "3", "ENABLE_TOOL_SEARCH": "false", "CLAUDE_CODE_DISABLE_NONSTREAMING_FALLBACK": "1", "SYSTRAY_CODEX_ACCOUNT_SLUG": selected.slug, } if not _authority_selected(selected): environment["SYSTRAY_CODEX_ACCOUNT_HOME"] = str(selected.native_auth_path.parent) return environment def run_proxy(model: str, argv: list[str], accounts: AccountOverrides) -> int: claude_argv = default_claude_args(argv) selected = _resolve_selected_for_overrides(accounts) _revalidate_selected_account(selected) claude_binary = _require_binary("claude") settings_path = selected.proxy_config_dir / "claude-settings.json" _atomic_write_json(settings_path, {"availableModels": list(AVAILABLE_MODELS)}) claude_argv = model_settings_args(claude_argv, settings_path) with codex_proxy(selected) as port: _revalidate_selected_account(selected) result = subprocess.run( [claude_binary, *claude_argv], env=proxy_claude_env(selected, model, port), check=False, ) return result.returncode def run_hybrid(argv: list[str], accounts: AccountOverrides) -> int: from claudex_router import HybridRouter claude_argv = default_claude_args(argv) selected = _resolve_selected_for_overrides(accounts) claude_binary = _require_binary("claude") account = _resolve_claude_for_overrides(accounts) _revalidate_selected_account(selected) _revalidate_claude_account(account) settings_path = selected.proxy_config_dir / "claudex-hybrid-settings.json" _atomic_write_json(settings_path, {"availableModels": list(HYBRID_PICKER_MODELS)}) claude_argv = model_settings_args(claude_argv, settings_path) with codex_proxy(selected) as port: router = HybridRouter(port) router.start() try: write_gateway_models(account.home, router.base_url) claude_env = hybrid_claude_env(selected, account, router.base_url) _revalidate_selected_account(selected) _revalidate_claude_account(account) result = subprocess.run( [claude_binary, *claude_argv], env=claude_env, check=False ) return result.returncode finally: router.stop() def resumed_session_id(argv: list[str]) -> str | None: for index, argument in enumerate(argv): flag, _, inline = argument.partition("=") if flag not in {"--resume", "-r"}: continue candidate = inline or (argv[index + 1] if index + 1 < len(argv) else "") if SESSION_ID_PATTERN.fullmatch(candidate): return candidate return None def session_projects_dirs() -> list[Path]: roots = [Path.home() / ".claude" / "projects"] accounts_dir = runtime_dir() / "claude-accounts" if accounts_dir.is_dir(): roots.extend( account_dir / "CLAUDE_HOME" / "projects" for account_dir in accounts_dir.iterdir() ) seen: dict[Path, None] = {} for root in roots: if root.is_dir(): seen.setdefault(root.resolve(), None) return list(seen) def session_working_directory(session_id: str) -> Path | None: for projects_dir in session_projects_dirs(): for transcript in projects_dir.glob(f"*/{session_id}.jsonl"): recorded = _first_recorded_cwd(transcript) if recorded is not None and recorded.is_dir(): return recorded return None def _first_recorded_cwd(transcript: Path) -> Path | None: try: with transcript.open(encoding="utf-8") as handle: for line in handle: try: entry = json.loads(line) except json.JSONDecodeError: continue if isinstance(entry, dict) and isinstance(entry.get("cwd"), str): return Path(entry["cwd"]) except OSError: return None return None def relocate_to_resumed_session(argv: list[str]) -> None: session_id = resumed_session_id(argv) if session_id is None: return recorded = session_working_directory(session_id) if recorded is None or recorded == Path.cwd(): return os.chdir(recorded) print(f"claudex: resuming in {recorded}", file=sys.stderr) def run(argv: list[str]) -> int: seat_mode, argv = parse_seat_flag(argv) if seat_mode: accounts, argv = parse_seat_account_overrides(argv) if accounts.human_override_lock and accounts.claude is None: raise HumanOverrideError( "--human-override-lock requires an explicit --account" ) return run_seat(argv, accounts) accounts, argv = parse_account_overrides(argv) if accounts.human_override_lock and accounts.codex is None and accounts.claude is None: raise HumanOverrideError( "--human-override-lock requires an explicit --account" ) enforce_local_seat_binding_gate(existing_claude_account_slug(accounts.claude)) relocate_to_resumed_session(argv) model, claude_argv = parse_model(argv) if model is None: if should_run_codex_only_for_pinned_account(accounts): return run_proxy(PROXY_MODEL, claude_argv, accounts) return run_hybrid(claude_argv, accounts) if model in NATIVE_PICKER_MODELS: return run_native(model, claude_argv, accounts) return run_proxy(model, claude_argv, accounts) def main(argv: list[str]) -> int: try: return run(argv[1:]) except KeyboardInterrupt: return 130 except ClaudexCappedError as exc: print(exc, file=sys.stderr) print(exc.refusal_json()) return exc.exit_code except AccountLockError as exc: print(f"claudex: {exc}", file=sys.stderr) print(json.dumps(exc.refusal_payload(), separators=(",", ":"))) return exc.exit_code except ClaudexError as exc: print(exc, file=sys.stderr) return exc.exit_code except OSError as exc: print(f"claudex: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": os.environ.setdefault("STALL_GUARD_NOTIFY", "1") raise SystemExit(main(sys.argv))