#!/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)