218 lines
8.1 KiB
Python
Executable File
218 lines
8.1 KiB
Python
Executable File
#!/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)
|