#!/usr/bin/env python3 """Atomic per-user installer for Codex Multi-Model Orchestrator.""" from __future__ import annotations import argparse import contextlib import datetime as dt import json import math import os import re import shlex import shutil import stat import subprocess import sys import tempfile import traceback from collections.abc import Iterable, Mapping from pathlib import Path from typing import Any PACKAGE_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PACKAGE_ROOT / "libexec")) from mmo_version import ( # noqa: E402 APP_SERVER_PROTOCOL_CODEX_VERSION, MMO_SCHEMA_VERSION, PACKAGE_VERSION, SWITCHYARD_BASELINE_VERSION, ) VERSION = PACKAGE_VERSION PAYLOAD_FILES = ( "LICENSE", "VERSION", "README.md", "CHANGELOG.md", "VALIDATION.md", "PACKAGE-MANIFEST.json", ) PAYLOAD_DIRECTORIES = ( "config", "docs", "evals", "libexec", "profiles", "scripts", "tests", ) RETAINED_OWNER_MANIFEST = ".codex-mmo-install-owner.json" def filtered_environment(*, extra: dict[str, str] | None = None) -> dict[str, str]: sensitive = re.compile( r"(?:^|_)(?:API_?KEY|KEY|TOKEN|SECRET|PASSWORD|PASSWD|CREDENTIAL|COOKIE|AUTH|PRIVATE)(?:_|$)", re.IGNORECASE, ) result = { key: value for key, value in os.environ.items() if not key.startswith("MMO_") and not sensitive.search(key) } if extra: result.update(extra) return result def toml_quote(value: str) -> str: try: value.encode("utf-8") except UnicodeEncodeError as exc: raise ValueError("generated TOML strings must contain valid Unicode scalar values") from exc return json.dumps(value, ensure_ascii=False, allow_nan=False).replace("\x7f", "\\u007F") def xdg_path(variable: str, fallback: Path) -> Path: """Resolve an XDG base directory, ignoring empty or relative values.""" raw = os.environ.get(variable) if raw: candidate = Path(raw) if candidate.is_absolute(): return candidate return fallback def default_install_root() -> Path: return xdg_path("XDG_DATA_HOME", Path.home() / ".local" / "share") / "codex-mmo" def default_config_root() -> Path: return xdg_path("XDG_CONFIG_HOME", Path.home() / ".config") / "codex-mmo" def default_state_root() -> Path: return xdg_path("XDG_STATE_HOME", Path.home() / ".local" / "state") / "codex-mmo" def default_bin_dir() -> Path: return xdg_path("XDG_BIN_HOME", Path.home() / ".local" / "bin") def atomic_write(path: Path, data: str, mode: int = 0o600) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) os.chmod(temporary, mode) os.replace(temporary, path) finally: with contextlib.suppress(FileNotFoundError): os.unlink(temporary) def remove_tree(path: Path) -> None: """Remove a tree that may contain immutable snapshot permissions.""" try: root_mode = path.lstat().st_mode except FileNotFoundError: return if stat.S_ISLNK(root_mode): path.unlink() return if not stat.S_ISDIR(root_mode): raise RuntimeError(f"refusing to recursively remove non-directory: {path}") for item in path.rglob("*"): with contextlib.suppress(OSError): mode = item.lstat().st_mode if not stat.S_ISLNK(mode): os.chmod(item, 0o700 if stat.S_ISDIR(mode) else 0o600) with contextlib.suppress(OSError): os.chmod(path, 0o700) shutil.rmtree(path) def validate_destination_paths(paths: Iterable[Path]) -> None: values = list(paths) protected = {Path("/"), Path.home().resolve()} for path in values: if path in protected or len(path.parts) <= 2: raise ValueError(f"refusing to install into broad protected path: {path}") for index, left in enumerate(values): for right in values[index + 1 :]: if left == right or left in right.parents or right in left.parents: raise ValueError(f"installation paths must not overlap: {left} and {right}") def _reject_json_constant(value: str) -> None: raise ValueError(f"non-standard JSON constant: {value}") def _finite_json_float(value: str) -> float: parsed = float(value) if not math.isfinite(parsed): raise ValueError(f"JSON number is outside the supported finite range: {value}") return parsed def _unique_json_object(pairs: list[tuple[str, object]]) -> dict[str, object]: result: dict[str, object] = {} for key, item in pairs: if key in result: raise ValueError(f"duplicate JSON object member: {key!r}") result[key] = item return result def _validate_json_unicode(value: object) -> None: if isinstance(value, str): try: value.encode("utf-8") except UnicodeEncodeError as exc: raise ValueError("JSON strings must contain valid Unicode scalar values") from exc elif isinstance(value, list): for item in value: _validate_json_unicode(item) elif isinstance(value, dict): for key, item in value.items(): _validate_json_unicode(key) _validate_json_unicode(item) def _strict_json_loads(value: str) -> object: parsed = json.loads( value, parse_constant=_reject_json_constant, parse_float=_finite_json_float, object_pairs_hook=_unique_json_object, ) _validate_json_unicode(parsed) return parsed def validate_existing_install(path: Path) -> None: """Refuse to replace a nonempty tree without this installer's ownership record.""" if not path.exists(): return if path.is_symlink() or not path.is_dir(): raise RuntimeError(f"existing install root is not a real directory: {path}") if next(path.iterdir(), None) is None: return manifest_path = path / "config" / "install-manifest.json" if manifest_path.is_symlink() or not manifest_path.is_file(): raise RuntimeError( f"refusing to replace a nonempty install root without its ownership manifest: {path}" ) try: data = _strict_json_loads(manifest_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError, ValueError) as exc: raise RuntimeError( f"existing install ownership manifest is invalid: {manifest_path}" ) from exc recorded = data.get("install_root") if isinstance(data, dict) else None if ( not isinstance(data, dict) or not isinstance(data.get("schema_version"), int) or isinstance(data.get("schema_version"), bool) or data.get("schema_version") != MMO_SCHEMA_VERSION or data.get("package") != "codex-multimodel-orchestrator" or not isinstance(recorded, str) or not Path(recorded).expanduser().is_absolute() or Path(recorded).expanduser().resolve(strict=False) != path ): raise RuntimeError(f"existing install ownership manifest does not match: {path}") def validate_wrapper_targets(bin_dir: Path) -> None: marker = "Installed by codex-multimodel-orchestrator" for name in ("codex-mmoctl", "codex-mmo", "codex-mmo-uninstall"): path = bin_dir / name if not path.exists() and not path.is_symlink(): continue if path.is_symlink() or not path.is_file(): raise FileExistsError(f"refusing to replace non-regular wrapper target: {path}") try: owned = marker in path.read_text(encoding="utf-8", errors="replace")[:500] except OSError as exc: raise FileExistsError(f"unable to verify existing wrapper ownership: {path}") from exc if not owned: raise FileExistsError(f"refusing to replace unowned executable: {path}") def _ignore(_directory: str, names: list[str]) -> set[str]: ignored = { name for name in names if name in {"__pycache__", ".pytest_cache", ".mypy_cache", ".ruff_cache"} } ignored.update(name for name in names if name.endswith((".pyc", ".pyo"))) return ignored def validate_payload_source() -> None: """Require a complete, static source tree before copying an installer payload.""" for name in PAYLOAD_FILES: source = PACKAGE_ROOT / name try: mode = source.lstat().st_mode except FileNotFoundError as exc: raise FileNotFoundError(f"required installer payload file is missing: {name}") from exc if stat.S_ISLNK(mode) or not stat.S_ISREG(mode): raise ValueError(f"installer payload file must be regular and not a symlink: {name}") for name in PAYLOAD_DIRECTORIES: source = PACKAGE_ROOT / name try: mode = source.lstat().st_mode except FileNotFoundError as exc: raise FileNotFoundError( f"required installer payload directory is missing: {name}" ) from exc if stat.S_ISLNK(mode) or not stat.S_ISDIR(mode): raise ValueError(f"installer payload directory must be real and not a symlink: {name}") for item in source.rglob("*"): relative = item.relative_to(PACKAGE_ROOT) item_mode = item.lstat().st_mode if stat.S_ISLNK(item_mode): raise ValueError(f"symbolic links are not allowed in installer payload: {relative}") if not (stat.S_ISREG(item_mode) or stat.S_ISDIR(item_mode)): raise ValueError(f"special files are not allowed in installer payload: {relative}") def copy_payload(destination: Path) -> None: validate_payload_source() destination.mkdir(parents=True, exist_ok=False) for name in PAYLOAD_FILES: source = PACKAGE_ROOT / name shutil.copy2(source, destination / name) for name in PAYLOAD_DIRECTORIES: source = PACKAGE_ROOT / name shutil.copytree(source, destination / name, ignore=_ignore) for path in destination.rglob("*"): if path.is_dir(): os.chmod(path, 0o755) elif path.suffix == ".py": os.chmod(path, 0o755 if os.access(path, os.X_OK) else 0o644) else: os.chmod(path, 0o644) def wrapper( python: Path, entrypoint: Path, fixed: Iterable[str] = (), *, environment: Mapping[str, str] | None = None, ) -> str: words = [ shlex.quote(str(python)), shlex.quote(str(entrypoint)), *[shlex.quote(item) for item in fixed], ] command = " ".join(words) exports = "".join( f"export {name}={shlex.quote(value)}\n" for name, value in sorted((environment or {}).items()) ) return f"""#!/usr/bin/env bash # Installed by codex-multimodel-orchestrator. set -euo pipefail {exports}exec {command} "$@" """ def main_wrapper(python: Path, entrypoint: Path) -> str: return wrapper( python, entrypoint, environment={"MMO_CLI_ENTRYPOINT": "codex-mmo"}, ) def install_optional_tools(args: argparse.Namespace) -> None: if args.install_codex: codex = shutil.which(args.codex_bin) observed = "" if codex: result = subprocess.run( [codex, "--version"], text=True, capture_output=True, timeout=10, check=False, env=filtered_environment(), ) if result.returncode == 0: observed = result.stdout.strip().removeprefix("codex-cli ").strip() if observed != APP_SERVER_PROTOCOL_CODEX_VERSION: npm = shutil.which("npm") if not npm: raise RuntimeError("--install-codex requires npm") subprocess.run( [ npm, "install", "-g", f"@openai/codex@{APP_SERVER_PROTOCOL_CODEX_VERSION}", ], check=True, env=filtered_environment(), ) codex = shutil.which(args.codex_bin) if not codex: raise RuntimeError( f"npm installation did not provide requested Codex binary {args.codex_bin!r}" ) result = subprocess.run( [codex, "--version"], text=True, capture_output=True, timeout=10, check=False, env=filtered_environment(), ) observed = ( result.stdout.strip().removeprefix("codex-cli ").strip() if result.returncode == 0 else "" ) if observed != APP_SERVER_PROTOCOL_CODEX_VERSION: raise RuntimeError( "npm installed Codex but the requested binary does not expose the reviewed " f"version {APP_SERVER_PROTOCOL_CODEX_VERSION}: observed " f"{observed or 'unavailable'} at {codex}" ) if args.install_switchyard and not shutil.which(args.switchyard_bin): cargo = shutil.which("cargo") if not cargo: raise RuntimeError("--install-switchyard requires Cargo") subprocess.run( [ cargo, "install", "--locked", "--version", SWITCHYARD_BASELINE_VERSION, "switchyard-server", ], check=True, env=filtered_environment(), ) if not shutil.which(args.switchyard_bin): raise RuntimeError( "Cargo installation did not provide requested Switchyard binary " f"{args.switchyard_bin!r}" ) def validate_stage(stage: Path, config_root: Path, state_root: Path) -> None: validation_state = Path(tempfile.mkdtemp(prefix=".install-validation-", dir=state_root)) env = filtered_environment( extra={ "MMO_INSTALL_ROOT": str(stage), "MMO_CONFIG_ROOT": str(config_root), "MMO_STATE_ROOT": str(validation_state), "PYTHONDONTWRITEBYTECODE": "1", "PYTHONPATH": str(stage / "libexec"), } ) code = r""" from mmo_catalog import local_inventory_report from mmo_profiles import builtin_profiles_root, resolve_profile from mmo_snapshot import compile_profile profile_directories = sorted( path for path in builtin_profiles_root().iterdir() if path.is_dir() and (path / "profile.toml").is_file() ) profiles = {} for directory in profile_directories: resolved = resolve_profile(directory) profile_id = resolved["profile"]["id"] if profile_id in profiles: raise RuntimeError(f"duplicate bundled profile id: {profile_id}") profiles[profile_id] = (directory, resolved) expected_profiles = { "adaptive-engineering", "access-efficient-escalation-lab", "bounded-research-organization-lab", "codex-harness-team", "competing-implementations-lab", "contract-first-refactoring", "high-confidence-debugging", "incident-hypothesis-triage", "research-backed-engineering", "route-resilience-lab", "secure-change", "visual-engineering", } if set(profiles) != expected_profiles: raise RuntimeError( f"bundled profile set differs from the release contract: {sorted(profiles)}" ) for profile_id in sorted(profiles): directory, resolved = profiles[profile_id] if resolved["profile"]["root"] not in resolved["agents"]: raise RuntimeError(f"profile root is missing from agents: {profile_id}") compile_profile(directory) report = local_inventory_report() if not report["passed"]: raise RuntimeError(f"local inventory validation failed: {report}") print(len(profiles)) """ try: result = subprocess.run( [sys.executable, "-c", code], env=env, cwd=stage, text=True, capture_output=True, check=False, ) finally: if validation_state.exists(): remove_tree(validation_state) if result.returncode: raise RuntimeError("staged package validation failed:\n" + result.stdout + result.stderr) def parser() -> argparse.ArgumentParser: options: dict[str, Any] = { "description": "Install or atomically replace one per-user Codex MMO runtime.", "epilog": ( "Existing schema-8 configuration and state are retained. Run the uninstaller " "with explicit purge flags when permanent removal is intended." ), "allow_abbrev": False, } if sys.version_info >= (3, 14): options["color"] = False value = argparse.ArgumentParser(**options) value.add_argument( "--version", action="version", version=VERSION, help="print the package version" ) value.add_argument( "--install-root", type=Path, default=default_install_root(), help="runtime payload destination (default: XDG data home/codex-mmo)", ) value.add_argument( "--config-root", type=Path, default=default_config_root(), help="operator configuration destination (default: XDG config home/codex-mmo)", ) value.add_argument( "--state-root", type=Path, default=default_state_root(), help="session and job state destination (default: XDG state home/codex-mmo)", ) value.add_argument( "--bin-dir", type=Path, default=default_bin_dir(), help="installed command directory (default: XDG_BIN_HOME or ~/.local/bin)", ) value.add_argument( "--codex-bin", default="codex", help="Codex executable name or path (default: codex)" ) value.add_argument( "--switchyard-bin", default="switchyard-server", help="Switchyard executable name or path (default: switchyard-server)", ) value.add_argument( "--install-codex", action="store_true", help="install the reviewed Codex CLI with npm when the requested binary is missing", ) value.add_argument( "--install-switchyard", action="store_true", help="install the pinned Switchyard release with Cargo when missing", ) value.add_argument( "--no-validate", action="store_true", help="skip staged package validation (development use only)", ) value.add_argument( "--debug", action="store_true", help="show a Python traceback when installation fails" ) return value def main() -> int: if sys.version_info < (3, 11): # noqa: UP036 - bootstrap a clear unsupported-Python error raise RuntimeError("Python 3.11 or newer is required") args = parser().parse_args() for name in ("install_root", "config_root", "state_root", "bin_dir"): setattr(args, name, getattr(args, name).expanduser().resolve()) validate_destination_paths((args.install_root, args.config_root, args.state_root, args.bin_dir)) validate_existing_install(args.install_root) validate_wrapper_targets(args.bin_dir) install_optional_tools(args) args.config_root.mkdir(parents=True, exist_ok=True, mode=0o700) args.state_root.mkdir(parents=True, exist_ok=True, mode=0o700) args.bin_dir.mkdir(parents=True, exist_ok=True, mode=0o755) (args.config_root / "profiles.d").mkdir(mode=0o700, exist_ok=True) (args.config_root / "catalog.d").mkdir(mode=0o700, exist_ok=True) (args.config_root / "tool-mcp.d").mkdir(mode=0o700, exist_ok=True) retained_owner_markers = tuple( root / RETAINED_OWNER_MANIFEST for root in (args.config_root, args.state_root) ) for marker in retained_owner_markers: if marker.is_symlink(): raise RuntimeError(f"refusing retained ownership marker symlink: {marker}") if marker.exists() and not marker.is_file(): raise RuntimeError(f"retained ownership marker is not a regular file: {marker}") settings_path = args.config_root / "settings.toml" if not settings_path.exists(): settings = (PACKAGE_ROOT / "config" / "settings.toml").read_text(encoding="utf-8") settings = settings.replace( 'codex_bin = "codex"', f"codex_bin = {toml_quote(args.codex_bin)}", ) settings = settings.replace( 'switchyard_bin = "switchyard-server"', f"switchyard_bin = {toml_quote(args.switchyard_bin)}", ) atomic_write(settings_path, settings, 0o600) credentials_path = args.config_root / "credentials.env" credentials_existed = credentials_path.exists() if not credentials_existed: atomic_write( credentials_path, (PACKAGE_ROOT / "config" / "credentials.env.example").read_text(encoding="utf-8"), 0o600, ) stage_parent = args.install_root.parent stage_parent.mkdir(parents=True, exist_ok=True, mode=0o755) stage = Path(tempfile.mkdtemp(prefix=f".{args.install_root.name}.staging-", dir=stage_parent)) backup: Path | None = None old_install: Path | None = None wrapper_paths = tuple( args.bin_dir / name for name in ("codex-mmoctl", "codex-mmo", "codex-mmo-uninstall") ) previous_wrappers: dict[Path, tuple[str, int] | None] = {} try: remove_tree(stage) copy_payload(stage) runtime = { "schema_version": MMO_SCHEMA_VERSION, "package": "codex-multimodel-orchestrator", "package_version": VERSION, "install_root": str(args.install_root), "config_root": str(args.config_root), "state_root": str(args.state_root), "bin_dir": str(args.bin_dir), "python_bin": str(Path(sys.executable).resolve()), "installed_at": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"), } atomic_write( stage / "config" / "runtime.json", json.dumps(runtime, indent=2, sort_keys=True, allow_nan=False) + "\n", 0o644, ) manifest = { **runtime, "source_root": str(PACKAGE_ROOT), "wrappers": [ str(args.bin_dir / "codex-mmo"), str(args.bin_dir / "codex-mmoctl"), str(args.bin_dir / "codex-mmo-uninstall"), ], } atomic_write( stage / "config" / "install-manifest.json", json.dumps(manifest, indent=2, sort_keys=True, allow_nan=False) + "\n", 0o644, ) if not args.no_validate: validate_stage(stage, args.config_root, args.state_root) python = Path(sys.executable).resolve() wrapper_contents = { args.bin_dir / "codex-mmoctl": wrapper( python, args.install_root / "libexec" / "mmoctl.py", environment={"MMO_CLI_ENTRYPOINT": "codex-mmoctl"}, ), args.bin_dir / "codex-mmo": main_wrapper( python, args.install_root / "libexec" / "mmoctl.py" ), args.bin_dir / "codex-mmo-uninstall": wrapper( python, args.install_root / "scripts" / "uninstall.py", ( "--install-root", str(args.install_root), "--config-root", str(args.config_root), "--state-root", str(args.state_root), "--bin-dir", str(args.bin_dir), ), ), } for path in wrapper_paths: previous_wrappers[path] = ( (path.read_text(encoding="utf-8"), path.stat().st_mode & 0o777) if path.is_file() else None ) try: # Wrapper targets use the stable install path, so they can be # staged before the brief atomic payload swap and rolled back if # any write or activation step fails. for path, contents in wrapper_contents.items(): atomic_write(path, contents, 0o755) if args.install_root.exists(): stamp = dt.datetime.now(dt.UTC).strftime("%Y%m%d-%H%M%S-%f") backup_path = args.state_root / "backups" / f"install-{stamp}" backup_path.parent.mkdir(parents=True, exist_ok=True) shutil.copytree(args.install_root, backup_path, symlinks=True) backup = backup_path old_path = args.install_root.with_name( f".{args.install_root.name}.previous-{os.getpid()}-{stamp}" ) os.replace(args.install_root, old_path) old_install = old_path try: os.replace(stage, args.install_root) except BaseException: if old_install and old_install.exists() and not args.install_root.exists(): os.replace(old_install, args.install_root) raise except BaseException: for path, previous in previous_wrappers.items(): if previous is None: path.unlink(missing_ok=True) else: atomic_write(path, previous[0], previous[1]) raise if old_install and old_install.exists(): try: remove_tree(old_install) except (OSError, RuntimeError) as exc: print( f"warning: installed successfully but could not remove {old_install}: {exc}", file=sys.stderr, ) # Retained ownership markers authorize a later purge only while the # payload is absent. Once a fresh payload is active its embedded # manifest is authoritative, and leaving an older marker behind would # retain an obsolete MMO schema in the active config/state roots. for marker in retained_owner_markers: marker.unlink(missing_ok=True) finally: if stage.exists(): with contextlib.suppress(OSError): remove_tree(stage) print(f"Codex Multi-Model Orchestrator {VERSION} installed.") print(f" executable: {args.bin_dir / 'codex-mmo'}") print(f" control: {args.bin_dir / 'codex-mmoctl'}") print(f" install: {args.install_root}") print(f" config: {args.config_root}") print(f" state: {args.state_root}") if backup: print(f" backup: {backup}") print() print(f"Configure provider credentials in {credentials_path}") print("Authenticate built-in Codex models with: codex-mmo auth login") print("Validate with: codex-mmo validate --all-profiles") print("Live-check with: codex-mmo doctor --live") print("Launch with: codex-mmo") path_entries = { Path(item).expanduser().resolve(strict=False) for item in os.environ.get("PATH", "").split(os.pathsep) if item } if args.bin_dir not in path_entries: print( f"warning: {args.bin_dir} is not on PATH; add it before invoking codex-mmo", file=sys.stderr, ) return 0 def cli_main() -> int: try: return main() except KeyboardInterrupt: return 130 except Exception as exc: print(f"error: {exc}", file=sys.stderr) if "--debug" in sys.argv[1:]: print(f"exception: {type(exc).__name__}", file=sys.stderr) traceback.print_exc(file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(cli_main())