#!/usr/bin/env python3
# Copyright 2026 Lusoris
# SPDX-License-Identifier: EUPL-1.2
"""Preview/install paused, hash-bound local train adapters; never control PIDs."""

from __future__ import annotations

import argparse
import fcntl
import hashlib
import json
import os
import shlex
import shutil
import stat
import sys
import tempfile
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from scripts.lib import git_env
from scripts.lib.safe_subprocess import CommandFailed
from scripts.lib.safe_subprocess import run as run_command

SOURCE_PATH = "scripts/dev/merge_train_guard.py"
COMPOSER_PATH = "scripts/ci/squash_message.py"
RUNTIME_FILES = ("train.sh", "rebase-clean.sh", "watchdog.sh", "merge_train_operator.py")
CONTROL_FILES = (*RUNTIME_FILES, "policy.json", "hold.txt", "PAUSED")


def sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def git(source: Path, *args: str) -> str:
    executable = shutil.which("git")
    if executable is None:
        raise ValueError("git is unavailable")
    result = run_command(
        [executable, "-C", str(source), *args],
        allowed_executables=(executable,),
        env=git_env.drop(),
        capture_output=True,
        text=True,
        check=True,
        timeout_seconds=60,
    )
    return result.stdout


def snapshot(path: Path) -> dict[str, Any] | None:
    try:
        mode = path.lstat().st_mode
    except FileNotFoundError:
        return None
    if not stat.S_ISREG(mode):
        raise ValueError(f"refusing non-regular runtime target: {path}")
    return {"sha256": sha256(path.read_bytes()), "mode": stat.S_IMODE(mode)}


def load_committed_blob(source: Path, head: str, relative_path: str, label: str) -> bytes:
    """Load committed Git blob for a relative path; raises ValueError if on-disk file differs."""
    committed = git(source, "show", f"{head}:{relative_path}").encode()
    if (source / relative_path).read_bytes() != committed:
        raise ValueError(f"{label} source must match its committed Git blob")
    return committed


def plan(
    source: Path, root: Path, state: Path, branches: list[str], worktrees: list[str]
) -> tuple[dict[str, Any], bytes]:
    """Compute installation plan and gateway bytes; raises ValueError on invalid policy or drift.

    Returns:
        Tuple of (plan_dict, committed_gateway_bytes).
    """
    head = git(source, "rev-parse", "HEAD").strip()
    committed = load_committed_blob(source, head, SOURCE_PATH, "gateway")
    committed_composer = load_committed_blob(source, head, COMPOSER_PATH, "composer")
    old = {name: snapshot(state / name) for name in CONTROL_FILES}
    policy: dict[str, Any] = {"schema": 1, "protected_branches": [], "protected_worktrees": []}
    if old["policy.json"] is not None:
        policy = json.loads((state / "policy.json").read_text())
    if not isinstance(policy, dict) or policy.get("schema") != 1:
        raise ValueError("unsupported existing policy; preserve it for manual review")
    for key, added in (("protected_branches", branches), ("protected_worktrees", worktrees)):
        if not isinstance(policy.get(key), list) or not all(
            isinstance(item, str) for item in policy[key]
        ):
            raise ValueError(f"invalid policy list: {key}")
        policy[key] = sorted(set(policy[key]) | set(added))
    for path in policy["protected_worktrees"]:
        if not Path(path).is_absolute():
            raise ValueError("protected worktree paths must be absolute")
    result = {
        "schema": 1,
        "source_commit": head,
        "source": str(source),
        "repository_root": str(root),
        "state": str(state),
        "gateway_sha256": sha256(committed),
        "composer_sha256": sha256(committed_composer),
        "before": old,
        "policy": policy,
        "installer_sha256": sha256(Path(__file__).read_bytes()),
    }
    return result, committed


def plan_hash(value: dict[str, Any]) -> str:
    return sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode())


def launch_code(
    gateway: Path,
    expected: str,
    composer_expected: str,
    root: Path,
    state: Path,
) -> str:
    composer = gateway.parent / "squash_message.py"
    return (
        "import hashlib,runpy,sys; from pathlib import Path; "
        f"p=Path({str(gateway)!r}); "
        f"actual=hashlib.sha256(p.read_bytes()).hexdigest(); expected={expected!r}; "
        "actual==expected or sys.exit('refused: installed gateway hash changed'); "
        f"cp=Path({str(composer)!r}); "
        f"cactual=hashlib.sha256(cp.read_bytes()).hexdigest(); cexpected={composer_expected!r}; "
        "cactual==cexpected or sys.exit('refused: installed composer hash changed'); "
        f"sys.argv=[str(p),*sys.argv[1:],'--repo-root',{str(root)!r},'--state-dir',{str(state)!r}]; "
        "runpy.run_path(str(p),run_name='__main__')"
    )


def adapters(
    gateway: Path,
    expected: str,
    composer_expected: str,
    root: Path,
    state: Path,
) -> dict[str, bytes]:
    """Generate runtime adapter script contents bound to gateway and composer hashes.

    Args:
        expected: Expected SHA-256 digest of the gateway script.
        composer_expected: Expected SHA-256 digest of the composer script.
    """
    code = launch_code(gateway, expected, composer_expected, root, state)
    launcher = f"exec python3 -c {shlex.quote(code)}"
    train = f'#!/bin/sh\n# Generated by install_merge_train_guard.py; default read-only.\nset -eu\n{launcher} cycle "$@"\n'
    rebase = f'#!/bin/sh\n# Explicit PR and --apply required by the gateway.\nset -eu\n{launcher} rebase "$@"\n'
    operator = (
        "#!/usr/bin/env python3\n# Retired unrestricted operator; inspection only.\n"
        "import sys\n"
        "if len(sys.argv) != 1:\n    raise SystemExit('operator is inspection-only; no arguments accepted')\n"
        "sys.argv = [sys.argv[0], 'cycle']\n" + code + "\n"
    )
    train_path = shlex.quote(str(state / "train.sh"))
    watchdog = (
        "#!/bin/sh\n# Observe only; process start/stop belongs to the operator.\nset -eu\n"
        'if [ "$#" -ne 0 ]; then\n\techo "watchdog is inspection-only" >&2\n\texit 2\nfi\n'
        f"while :; do\n\tif ! {train_path}; then echo 'guarded observer refused this pass' >&2; fi\n\tsleep 60\ndone\n"
    )
    return {
        name: content.encode()
        for name, content in zip(RUNTIME_FILES, (train, rebase, watchdog, operator), strict=True)
    }


def atomic_write(path: Path, data: bytes, mode: int) -> None:
    temporary: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as stream:
            temporary = Path(stream.name)
            stream.write(data)
            stream.flush()
            os.fsync(stream.fileno())
        temporary.chmod(mode)
        temporary.replace(path)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def install(
    value: dict[str, Any],
    committed: bytes,
    expected: str,
    composer_committed: bytes | None = None,
) -> Path:
    """Apply the previewed plan under the mutation lock.

    Raises ValueError when the plan, gateway or composer bytes differ from the preview, and
    BlockingIOError when another holder has the mutation lock.

    Args:
        expected: Expected SHA-256 digest of the preview plan.

    Returns:
        Path to the migration backup directory.
    """
    if plan_hash(value) != expected:
        raise ValueError("preview plan changed; inspect a fresh plan before applying")
    if sha256(committed) != value["gateway_sha256"]:
        raise ValueError("gateway bytes do not match the preview")
    if composer_committed is None:
        source = Path(value["source"])
        head = str(value["source_commit"])
        composer_committed = git(source, "show", f"{head}:{COMPOSER_PATH}").encode()
    if sha256(composer_committed) != value["composer_sha256"]:
        raise ValueError("composer bytes do not match the preview")
    state = Path(value["state"])
    state.mkdir(parents=True, exist_ok=True)
    with (state / "mutation.lock").open("a") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        return install_locked(value, committed, expected, state, composer_committed)


def prepare_release(
    release: Path,
    gateway_bytes: bytes,
    gateway_sha: str,
    composer_bytes: bytes,
    composer_sha: str,
) -> Path:
    """Write the read-only gateway and composer release files, keeping matching ones.

    Raises ValueError when a release directory is a symlink or an existing release file has a
    different hash.

    Returns:
        Path to the installed gateway script.
    """
    for directory in (release.parent, release):
        if directory.is_symlink():
            raise ValueError(f"refusing symlink release directory: {directory}")
        directory.mkdir(exist_ok=True)
    gateway = release / "merge_train_guard.py"
    existing = snapshot(gateway)
    if existing is not None and existing["sha256"] != gateway_sha:
        raise ValueError("release path contains unexpected evidence; refusing overwrite")
    if existing is None:
        atomic_write(gateway, gateway_bytes, 0o444)
    composer = release / "squash_message.py"
    existing_composer = snapshot(composer)
    if existing_composer is not None and existing_composer["sha256"] != composer_sha:
        raise ValueError("release path contains unexpected composer evidence; refusing overwrite")
    if existing_composer is None:
        atomic_write(composer, composer_bytes, 0o444)
    return gateway


def release_name(value: dict[str, Any]) -> str:
    """Name the release directory after both the gateway and the composer hash.

    A change to either script then installs into a new immutable directory instead of
    meeting an older file there that prepare_release refuses to overwrite.

    Returns:
        SHA-256 hex digest of the two script hashes.
    """
    return sha256(f"{value['gateway_sha256']}:{value['composer_sha256']}".encode())


def install_locked(
    value: dict[str, Any],
    committed: bytes,
    expected: str,
    state: Path,
    composer_committed: bytes,
) -> Path:
    """Back up the runtime targets, pause the train, then install release files and adapters.

    Raises ValueError when a runtime target changed since the preview.

    Returns:
        Path to the migration backup directory.
    """
    backup = Path(tempfile.mkdtemp(prefix="migration-", dir=state))
    atomic_write(backup / "plan.json", (json.dumps(value, indent=2) + "\n").encode(), 0o644)
    for name, previous in value["before"].items():
        path = state / name
        if snapshot(path) != previous:
            raise ValueError(f"runtime target changed during installation: {path}")
        if previous is not None:
            shutil.copy2(path, backup / name)
    # Install no live bypass: leave paused even if a later step fails.
    if not (state / "PAUSED").exists():
        atomic_write(
            state / "PAUSED", b"Runtime migration: review before enabling mutations.\n", 0o644
        )
    release = state / "releases" / release_name(value)
    gateway = prepare_release(
        release,
        committed,
        value["gateway_sha256"],
        composer_committed,
        value["composer_sha256"],
    )
    targets = adapters(
        gateway,
        value["gateway_sha256"],
        value["composer_sha256"],
        Path(value["repository_root"]),
        state,
    )
    targets["policy.json"] = (json.dumps(value["policy"], indent=2) + "\n").encode()
    if value["before"]["hold.txt"] is None:
        targets["hold.txt"] = b"# Operator holds; #1213 is always excluded.\n"
    for name, data in targets.items():
        if snapshot(state / name) != value["before"][name]:
            raise ValueError(f"runtime target changed before replacement: {name}")
        atomic_write(state / name, data, 0o755 if name in RUNTIME_FILES else 0o644)
    receipt = {
        "plan": value,
        "plan_sha256": expected,
        "backup": str(backup),
        "after": {name: snapshot(state / name) for name in CONTROL_FILES},
    }
    atomic_write(backup / "receipt.json", (json.dumps(receipt, indent=2) + "\n").encode(), 0o644)
    return backup


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--source-root", type=Path, default=Path(__file__).resolve().parents[2])
    parser.add_argument("--repo-root", type=Path, required=True)
    parser.add_argument("--state-dir", type=Path, required=True)
    parser.add_argument("--protect-branch", action="append", default=[])
    parser.add_argument("--protect-worktree", action="append", default=[])
    parser.add_argument("--apply", action="store_true")
    parser.add_argument("--expect-plan", help="sha256 printed by the reviewed preview")
    args = parser.parse_args(argv)
    try:
        value, committed = plan(
            args.source_root.resolve(),
            args.repo_root.resolve(),
            args.state_dir.resolve(),
            args.protect_branch,
            args.protect_worktree,
        )
        if not args.apply:
            print(json.dumps({"plan_sha256": plan_hash(value), "plan": value}, indent=2))
        else:
            if not args.expect_plan:
                raise ValueError("--apply requires the reviewed --expect-plan hash")
            backup = install(value, committed, args.expect_plan)
            print(f"Installed paused adapters; originals and receipt: {backup}")
        return 0
    except (OSError, ValueError, TypeError, KeyError, CommandFailed) as exc:
        print(f"merge-train install refused: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())