codex-mmo/scripts/uninstall.py

#!/usr/bin/env python3
"""Uninstall Codex MMO without deleting user state unless explicitly requested."""

from __future__ import annotations

import argparse
import contextlib
import json
import math
import os
import shutil
import stat
import sys
import tempfile
import traceback
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 MMO_SCHEMA_VERSION, PACKAGE_VERSION  # noqa: E402

OWNER_MANIFEST = ".codex-mmo-install-owner.json"


def _reject_json_constant(value: str) -> Any:
    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, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    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: Any) -> 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) -> Any:
    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 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 defaults() -> tuple[Path, Path, Path, Path]:
    install = xdg_path("XDG_DATA_HOME", Path.home() / ".local" / "share") / "codex-mmo"
    config = xdg_path("XDG_CONFIG_HOME", Path.home() / ".config") / "codex-mmo"
    state = xdg_path("XDG_STATE_HOME", Path.home() / ".local" / "state") / "codex-mmo"
    bin_dir = xdg_path("XDG_BIN_HOME", Path.home() / ".local" / "bin")
    return install, config, state, bin_dir


def owned_wrapper(path: Path) -> bool:
    if not path.is_file():
        return False
    try:
        return (
            "Installed by codex-multimodel-orchestrator"
            in path.read_text(encoding="utf-8", errors="replace")[:500]
        )
    except OSError:
        return False


def absolute_path(path: Path) -> Path:
    """Return a lexical absolute path without following the deletion target."""

    return Path(os.path.abspath(path.expanduser()))


def remove_tree(path: Path) -> None:
    """Remove an owned tree even when compiled snapshots are read-only."""

    try:
        mode = path.lstat().st_mode
    except FileNotFoundError:
        return
    if stat.S_ISLNK(mode):
        path.unlink()
        return
    if not stat.S_ISDIR(mode):
        raise RuntimeError(f"refusing to recursively remove non-directory: {path}")
    if path.resolve(strict=False) != path:
        raise RuntimeError(f"refusing to remove a path reached through a symlink: {path}")
    for item in path.rglob("*"):
        with contextlib.suppress(OSError):
            item_mode = item.lstat().st_mode
            if not stat.S_ISLNK(item_mode):
                os.chmod(item, 0o700 if stat.S_ISDIR(item_mode) else 0o600)
    with contextlib.suppress(OSError):
        os.chmod(path, 0o700)
    shutil.rmtree(path)


def validate_removal_target(path: Path, label: str) -> None:
    protected = {Path("/"), Path.home().resolve()}
    if path in protected or len(path.parts) <= 2:
        raise ValueError(f"refusing to remove broad protected {label}: {path}")
    if path.resolve(strict=False) != path:
        raise ValueError(f"refusing {label} reached through a symlink: {path}")


def _option_was_supplied(name: str) -> bool:
    return any(item == name or item.startswith(f"{name}=") for item in sys.argv[1:])


def _trusted_manifest(path: Path, install_root: Path) -> dict[str, Any] | None:
    if path.is_symlink() or not path.is_file():
        return None
    try:
        data = _strict_json_loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError, ValueError):
        return None
    schema_version = data.get("schema_version") if isinstance(data, dict) else None
    if (
        not isinstance(data, dict)
        or not isinstance(schema_version, int)
        or isinstance(schema_version, bool)
        or schema_version != MMO_SCHEMA_VERSION
    ):
        return None
    recorded = data.get("install_root")
    if (
        not isinstance(recorded, str)
        or not Path(recorded).expanduser().is_absolute()
        or absolute_path(Path(recorded)) != install_root
    ):
        return None
    if data.get("package") != "codex-multimodel-orchestrator":
        return None
    return data


def _write_owner_manifest(root: Path, data: dict[str, Any]) -> None:
    """Persist deletion ownership when config/state survives package removal."""

    if not root.is_dir() or root.is_symlink():
        return
    destination = root / OWNER_MANIFEST
    fd, raw_temporary = tempfile.mkstemp(prefix=f".{OWNER_MANIFEST}.", dir=root)
    temporary = Path(raw_temporary)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle:
            json.dump(data, handle, indent=2, sort_keys=True, allow_nan=False)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(temporary, 0o600)
        os.replace(temporary, destination)
    finally:
        with contextlib.suppress(FileNotFoundError):
            temporary.unlink()


def main() -> int:
    install, config, state, bin_dir = defaults()
    parser_options: dict[str, Any] = {
        "description": "Remove the owned Codex MMO runtime while retaining user data by default.",
        "epilog": (
            "Configuration, credentials, sessions, jobs, evaluations, and logs remain unless "
            "their explicit purge flags are supplied. Ownership manifests guard every removal."
        ),
        "allow_abbrev": False,
    }
    if sys.version_info >= (3, 14):
        parser_options["color"] = False
    parser = argparse.ArgumentParser(**parser_options)
    parser.add_argument(
        "--version", action="version", version=PACKAGE_VERSION, help="print the package version"
    )
    parser.add_argument(
        "--install-root", type=Path, default=install, help="owned runtime payload to remove"
    )
    parser.add_argument(
        "--config-root",
        type=Path,
        default=config,
        help="configuration root recorded by the installer",
    )
    parser.add_argument(
        "--state-root", type=Path, default=state, help="state root recorded by the installer"
    )
    parser.add_argument(
        "--bin-dir", type=Path, default=bin_dir, help="directory containing installed wrappers"
    )
    parser.add_argument(
        "--purge-config",
        action="store_true",
        help="also permanently remove configuration and credentials",
    )
    parser.add_argument(
        "--purge-state",
        action="store_true",
        help="also permanently remove sessions, jobs, evaluations, and logs",
    )
    parser.add_argument(
        "--debug", action="store_true", help="show a Python traceback when removal fails"
    )
    args = parser.parse_args()
    for name in ("install_root", "config_root", "state_root", "bin_dir"):
        setattr(args, name, absolute_path(getattr(args, name)))

    install_manifest = args.install_root / "config" / "install-manifest.json"
    # A retained config/state marker authorizes a later purge only after the
    # payload root is gone. If anything has appeared at the install path again,
    # require that live tree's own manifest so a stale marker cannot authorize
    # deletion of unrelated replacement content.
    manifests: tuple[Path, ...]
    if args.install_root.exists() or args.install_root.is_symlink():
        manifests = (install_manifest,)
    else:
        manifests = (
            args.config_root / OWNER_MANIFEST,
            args.state_root / OWNER_MANIFEST,
        )
    data = next(
        (
            candidate
            for path in manifests
            if (candidate := _trusted_manifest(path, args.install_root)) is not None
        ),
        None,
    )
    if data is None:
        raise RuntimeError(
            "refusing to remove an install root without its valid current-generation "
            "ownership manifest: "
            f"{args.install_root}"
        )
    for option, attribute in (
        ("--config-root", "config_root"),
        ("--state-root", "state_root"),
        ("--bin-dir", "bin_dir"),
    ):
        value = data.get(attribute)
        if not isinstance(value, str) or not Path(value).expanduser().is_absolute():
            raise RuntimeError(f"ownership manifest is missing {attribute}")
        recorded = absolute_path(Path(value))
        if _option_was_supplied(option):
            if getattr(args, attribute) != recorded:
                raise RuntimeError(
                    f"refusing {option} outside the ownership manifest: "
                    f"{getattr(args, attribute)} != {recorded}"
                )
        else:
            setattr(args, attribute, recorded)

    validate_removal_target(args.install_root, "install root")
    validate_removal_target(args.config_root, "configuration root")
    validate_removal_target(args.state_root, "state root")
    if args.bin_dir.resolve(strict=False) != args.bin_dir:
        raise ValueError(f"refusing binary directory reached through a symlink: {args.bin_dir}")

    # Preserve a location-bound ownership record before removing the only
    # installed copy. This supports a later explicit purge without relaxing
    # deletion checks for arbitrary directories.
    if not args.purge_config:
        _write_owner_manifest(args.config_root, data)
    if not args.purge_state:
        _write_owner_manifest(args.state_root, data)

    for name in ("codex-mmo", "codex-mmoctl", "codex-mmo-uninstall"):
        path = args.bin_dir / name
        if owned_wrapper(path):
            path.unlink()
    remove_tree(args.install_root)
    if args.purge_config:
        remove_tree(args.config_root)
    if args.purge_state:
        remove_tree(args.state_root)

    print("Codex Multi-Model Orchestrator removed.")
    if not args.purge_config and args.config_root.exists():
        print(f"Configuration retained at {args.config_root}; use --purge-config to remove it.")
    if not args.purge_state and args.state_root.exists():
        print(f"State retained at {args.state_root}; use --purge-state to remove it.")
    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())