bjd-cms-content/install/install.py

#!/usr/bin/env python3
"""Install or verify one immutable BJD CMS Content release."""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import uuid


VERSION_PATTERN = re.compile(r"^[0-9A-Za-z.+-]+$")


class InstallFailure(RuntimeError):
    pass


def require(condition: bool, message: str) -> None:
    if not condition:
        raise InstallFailure(message)


def load_release(release_root: Path) -> tuple[str, dict[str, object]]:
    try:
        version = (release_root / "VERSION").read_text(encoding="ascii").strip()
        metadata = json.loads((release_root / "RELEASE.json").read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as error:
        raise InstallFailure(f"release metadata cannot be read: {error}") from None

    require(bool(VERSION_PATTERN.fullmatch(version)), "release VERSION is invalid")
    require(isinstance(metadata, dict), "RELEASE.json must contain an object")
    require(metadata.get("release_version") == version, "release version metadata does not match VERSION")
    require(metadata.get("transport") == "stdio", "release transport must be stdio")
    require(metadata.get("skill_required") is True, "release must include its complete skill")
    require(sys.version_info >= (3, 10), "Python 3.10 or newer is required")

    required = (
        release_root / "plugin/bjd-cms-content/scripts/server.py",
        release_root / "plugin/bjd-cms-content/scripts/client.py",
        release_root / "skill/bjd-cms-content/SKILL.md",
    )
    require(all(path.is_file() for path in required), "release plugin or skill is incomplete")
    return version, metadata


def validate_prefix(raw_prefix: str) -> Path:
    prefix = Path(raw_prefix).expanduser()
    require(prefix.is_absolute(), "installation prefix must be absolute")
    prefix = prefix.resolve(strict=False)
    require(prefix != Path(prefix.anchor), "refusing a broad filesystem-root installation prefix")
    return prefix


def copy_version(release_root: Path, destination: Path, staging: Path) -> None:
    staging.mkdir(mode=0o755)
    shutil.copytree(release_root / "plugin", staging / "plugin")
    shutil.copytree(release_root / "skill", staging / "skill")
    shutil.copytree(release_root / "install", staging / "install")
    shutil.copy2(release_root / "VERSION", staging / "VERSION")
    shutil.copy2(release_root / "RELEASE.json", staging / "RELEASE.json")
    os.replace(staging, destination)


def native_smoke(destination: Path, metadata: dict[str, object]) -> dict[str, object]:
    plugin = destination / "plugin/bjd-cms-content"
    client = plugin / "scripts/client.py"
    environment = os.environ.copy()
    for name in (
        "AI_CONTENT_API_TOKEN",
        "BJD_CMS_SKILL_ROOT",
        "BJD_CMS_CONFIG_FILE",
        "BJD_CMS_ENV_FILE",
        "BJD_CMS_MCP_TRACE_FILE",
        "CMS_ROOT",
    ):
        environment.pop(name, None)
    completed = subprocess.run(
        [sys.executable, str(client), "native-smoke"],
        cwd=plugin,
        env=environment,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
        timeout=150,
    )
    if completed.returncode != 0:
        detail = (completed.stderr or completed.stdout or "native-smoke failed").strip()
        raise InstallFailure(detail[:1000])
    try:
        result = json.loads(completed.stdout)
    except json.JSONDecodeError:
        raise InstallFailure("native-smoke did not return JSON") from None
    require(isinstance(result, dict) and result.get("ok") is True, "native-smoke did not succeed")
    require(result.get("server_version") == metadata.get("server_version"), "server version mismatch")
    require(result.get("client_version") == metadata.get("client_version"), "client version mismatch")
    require(result.get("tool_count") == metadata.get("tool_count"), "tool count mismatch")
    require(result.get("bundled_pair_verified") is True, "bundled client/server pair is not verified")
    require(result.get("initialize") is True, "MCP initialize failed")
    require(result.get("ping") is True, "MCP ping failed")
    require(result.get("tools_list") is True, "MCP tools/list failed")
    return result


def update_windows_current(prefix: Path, version: str) -> Path:
    current = prefix / "current.txt"
    require(not current.exists() or current.is_file(), "refusing to replace a non-file current.txt")
    temporary = prefix / f".current-{os.getpid()}-{uuid.uuid4().hex}.txt"
    try:
        temporary.write_text(str(prefix / "versions" / version) + os.linesep, encoding="utf-8")
        os.replace(temporary, current)
    finally:
        temporary.unlink(missing_ok=True)
    return current


def update_posix_current(prefix: Path, version: str) -> Path:
    current = prefix / "current"
    require(not current.exists() or current.is_symlink(), "refusing to replace a non-symlink current path")
    temporary = prefix / f".current-{os.getpid()}-{uuid.uuid4().hex}"
    try:
        temporary.symlink_to(f"versions/{version}")
        os.replace(temporary, current)
    finally:
        temporary.unlink(missing_ok=True)
    return current


def update_current(prefix: Path, version: str) -> Path:
    if os.name == "nt":
        return update_windows_current(prefix, version)
    return update_posix_current(prefix, version)


def install(prefix: Path, release_root: Path) -> dict[str, object]:
    version, metadata = load_release(release_root)
    versions = prefix / "versions"
    destination = versions / version
    staging = versions / f".install-{version}-{os.getpid()}-{uuid.uuid4().hex}"
    versions.mkdir(parents=True, exist_ok=True)

    created = False
    if destination.exists() or destination.is_symlink():
        require(destination.is_dir() and not destination.is_symlink(), "installed version path is invalid")
        installed_version, installed_metadata = load_release(destination)
        require(installed_version == version, "installed version directory does not match the release")
        require(installed_metadata == metadata, "installed release metadata differs from the selected release")
        action = "already_current"
    else:
        action = "installed"
        try:
            copy_version(release_root, destination, staging)
            created = True
        finally:
            if staging.exists():
                shutil.rmtree(staging)

    try:
        smoke = native_smoke(destination, metadata)
        current = update_current(prefix, version)
    except Exception:
        if created and destination.is_dir() and not destination.is_symlink():
            shutil.rmtree(destination)
        raise

    return {
        "ok": True,
        "action": action,
        "release_version": version,
        "plugin_path": str(destination / "plugin/bjd-cms-content"),
        "skill_path": str(destination / "skill/bjd-cms-content"),
        "current_pointer": str(current),
        "python_executable": sys.executable,
        "native_smoke": {
            key: smoke.get(key)
            for key in (
                "server_version",
                "client_version",
                "tool_count",
                "bundled_pair_verified",
                "initialize",
                "ping",
                "tools_list",
            )
        },
        "next_action": "register_or_update_local_stdio_mcp_without_self_restart",
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--prefix", required=True, help="absolute per-user installation prefix")
    args = parser.parse_args()
    release_root = Path(__file__).resolve().parent.parent
    result = install(validate_prefix(args.prefix), release_root)
    print(json.dumps(result, ensure_ascii=False, separators=(",", ":")))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (InstallFailure, OSError, subprocess.SubprocessError) as error:
        print(
            json.dumps(
                {"ok": False, "error": str(error), "next_action": "stop_and_report"},
                ensure_ascii=False,
                separators=(",", ":"),
            )
        )
        raise SystemExit(1)