coord/install

#!/usr/bin/env python3
"""Install or verify a vendored, harness-neutral coordinator."""

from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import re
import secrets
import shlex
import stat
import sys
from typing import Any


HERE = Path(__file__).resolve().parent
CORE_FILES = (
    ".gitignore",
    "AGENTS.md",
    "CLAUDE.md",
    "DESIGN.md",
    "README.md",
    "VERSION",
    "coord",
    "hook.py",
    "install",
    "session.py",
    "store.py",
    "workspace.py",
)
EXECUTABLES = {"coord", "install"}
SKILL_FILES = (
    "SKILL.md",
    "agents/openai.yaml",
)
PROJECT_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}")
VERSION_RE = re.compile(
    r"(?:0|[1-9]\d*)\.(?:0|[1-9]\d*)\.(?:0|[1-9]\d*)"
)

CLAUDE_COMMANDS = {
    "SessionStart": ('python3 "$CLAUDE_PROJECT_DIR/.coord/session.py"', 10),
    "SessionEnd": ('python3 "$CLAUDE_PROJECT_DIR/.coord/session.py"', 5),
    "PreToolUse": ('python3 "$CLAUDE_PROJECT_DIR/.coord/hook.py"', 10),
}
CODEX_DISCOVERY = (
    'root="$PWD"; while [ "$root" != / ] && [ ! -f '
    '"$root/.coord/{name}.py" ]; do root="${{root%/*}}"; '
    '[ -n "$root" ] || root=/; done; '
    '[ -f "$root/.coord/{name}.py" ] || exit 0; '
    'python3 "$root/.coord/{name}.py"'
)


def legacy_wrapper(name: str) -> str:
    if name == "coord":
        return """#!/usr/bin/env python3
import os
from pathlib import Path
import sys

target = Path(__file__).resolve().parents[2] / ".coord" / "coord"
os.execv(str(target), [str(target), *sys.argv[1:]])
"""
    if name in {"hook.py", "session.py"}:
        return f"""#!/usr/bin/env python3
from pathlib import Path
import runpy

target = Path(__file__).resolve().parents[2] / ".coord" / "{name}"
runpy.run_path(str(target), run_name="__main__")
"""
    if name in {"store.py", "workspace.py"}:
        module = f"_coord_{name.removesuffix('.py')}_compat"
        return f"""from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
import sys

target = Path(__file__).resolve().parents[2] / ".coord" / "{name}"
spec = spec_from_file_location("{module}", target)
assert spec is not None and spec.loader is not None
implementation = module_from_spec(spec)
sys.modules[spec.name] = implementation
spec.loader.exec_module(implementation)
globals().update({{
    key: value for key, value in vars(implementation).items()
    if key not in {{"__name__", "__loader__", "__package__", "__spec__"}}
}})
"""
    raise ValueError(f"unsupported compatibility wrapper: {name}")


def read_json(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"refusing invalid JSON at {path}: {exc}") from exc
    if not isinstance(value, dict):
        raise ValueError(f"refusing non-object JSON at {path}")
    return value


def is_coord_hook(item: object) -> bool:
    if not isinstance(item, dict) or not isinstance(item.get("command"), str):
        return False
    command = item["command"]
    return any(
        endpoint in command
        for endpoint in (
            ".coord/hook.py",
            ".coord/session.py",
            ".claude/coord/hook.py",
            ".claude/coord/session.py",
        )
    )


def merged_hooks(
    current: dict[str, Any], desired: dict[str, tuple[str, int]]
) -> dict[str, Any]:
    result = json.loads(json.dumps(current))
    hooks = result.setdefault("hooks", {})
    if not isinstance(hooks, dict):
        raise ValueError("refusing hook configuration whose 'hooks' value is not an object")
    for event, (command, timeout) in desired.items():
        existing = hooks.get(event, [])
        if not isinstance(existing, list):
            raise ValueError(f"refusing hook event {event!r} whose value is not a list")
        retained = []
        for group in existing:
            if not isinstance(group, dict):
                retained.append(group)
                continue
            entries = group.get("hooks")
            if not isinstance(entries, list):
                retained.append(group)
                continue
            kept_entries = [entry for entry in entries if not is_coord_hook(entry)]
            if kept_entries:
                kept_group = dict(group)
                kept_group["hooks"] = kept_entries
                retained.append(kept_group)
        installed: dict[str, Any] = {
            "hooks": [{"type": "command", "command": command, "timeout": timeout}]
        }
        if event == "PreToolUse":
            installed["matcher"] = ".*"
        retained.append(installed)
        hooks[event] = retained
    return result


def codex_hooks(current: dict[str, Any]) -> dict[str, Any]:
    desired = {
        event: (
            CODEX_DISCOVERY.format(
                name="hook" if event == "PreToolUse" else "session"
            ),
            timeout,
        )
        for event, (_command, timeout) in CLAUDE_COMMANDS.items()
    }
    return merged_hooks(current, desired)


def write_if_changed(
    path: Path,
    content: bytes,
    *,
    executable: bool = False,
    check: bool,
    drift: list[str],
) -> None:
    observed = path.read_bytes() if path.is_file() else None
    expected_mode = 0o755 if executable else 0o644
    observed_executable = path.is_file() and bool(
        stat.S_IMODE(path.stat().st_mode) & stat.S_IXUSR
    )
    mode_ok = path.is_file() and observed_executable == executable
    if observed == content and mode_ok:
        return
    drift.append(str(path))
    if check:
        return
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.coord-new-{secrets.token_hex(6)}")
    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, expected_mode)
    try:
        remaining = memoryview(content)
        while remaining:
            written = os.write(fd, remaining)
            if written <= 0:
                raise OSError(f"short write while creating {temporary}")
            remaining = remaining[written:]
        os.fsync(fd)
    finally:
        os.close(fd)
    os.replace(temporary, path)
    os.chmod(path, expected_mode)


def project_id(target: Path, requested: str, *, check: bool) -> str:
    canonical = target / ".coord" / "project-id"
    legacy = target / ".claude" / "coord" / "project-id"
    existing = []
    for path in (canonical, legacy):
        if path.is_file():
            existing.append((path, path.read_text(encoding="utf-8").strip()))
    values = {value for _path, value in existing}
    if len(values) > 1:
        detail = ", ".join(f"{path}={value!r}" for path, value in existing)
        raise ValueError(f"refusing ambiguous existing project IDs: {detail}")
    selected = next(iter(values), "")
    if requested and selected and requested != selected:
        raise ValueError(
            f"refusing project ID change from {selected!r} to {requested!r}"
        )
    selected = selected or requested
    if not selected:
        if check:
            raise ValueError("installation has no project ID")
        raise ValueError("new installation requires --project-id")
    if not PROJECT_ID_RE.fullmatch(selected) or ".." in selected:
        raise ValueError("project ID must be one safe immutable slug")
    return selected


def source_version() -> str:
    try:
        version = (HERE / "VERSION").read_text(encoding="utf-8").strip()
    except OSError as exc:
        raise ValueError(f"cannot read source VERSION: {exc}") from exc
    if not VERSION_RE.fullmatch(version):
        raise ValueError(f"refusing invalid source VERSION: {version!r}")
    return version


def installed_version(target: Path) -> str:
    marker = target / ".coord" / "VERSION"
    if not marker.is_file():
        return "unversioned"
    try:
        version = marker.read_text(encoding="utf-8").strip()
    except OSError:
        return "unreadable"
    return version if VERSION_RE.fullmatch(version) else "invalid"


def install(target: Path, *, requested_id: str, check: bool) -> list[str]:
    target = target.expanduser().resolve()
    if not target.is_dir():
        raise ValueError(f"target is not a directory: {target}")
    source_version()
    selected_id = project_id(target, requested_id, check=check)
    destination = target / ".coord"
    planned: list[tuple[Path, bytes, bool]] = []
    for name in CORE_FILES:
        planned.append(
            (
                destination / name,
                (HERE / name).read_bytes(),
                name in EXECUTABLES,
            )
        )
    for root in (
        ".coord/skills/coord",
        ".agents/skills/coord",
        ".claude/skills/coord",
    ):
        for name in SKILL_FILES:
            planned.append(
                (
                    target / root / name,
                    (HERE / "skills" / "coord" / name).read_bytes(),
                    False,
                )
            )
    planned.append(
        (destination / "project-id", f"{selected_id}\n".encode(), False)
    )

    legacy = target / ".claude" / "coord"
    for name in ("coord", "hook.py", "session.py", "store.py", "workspace.py"):
        planned.append(
            (legacy / name, legacy_wrapper(name).encode(), name == "coord")
        )
    planned.extend(
        (
            (legacy / "project-id", f"{selected_id}\n".encode(), False),
            (
                legacy / "DESIGN.md",
                b"Canonical design: [`../../.coord/DESIGN.md`](../../.coord/DESIGN.md).\n",
                False,
            ),
            (legacy / ".gitignore", b"state/\n__pycache__/\n", False),
        )
    )

    settings = target / ".claude" / "settings.json"
    expected_claude = (
        json.dumps(
            merged_hooks(read_json(settings), CLAUDE_COMMANDS),
            indent=2,
            sort_keys=True,
        )
        + "\n"
    ).encode()

    codex = target / ".codex" / "hooks.json"
    expected_codex = (
        json.dumps(codex_hooks(read_json(codex)), indent=2, sort_keys=True) + "\n"
    ).encode()
    planned.extend(
        (
            (settings, expected_claude, False),
            (codex, expected_codex, False),
        )
    )

    # All source reads, identity checks, wrapper generation and hook JSON
    # validation complete before the first target replacement.
    drift: list[str] = []
    for path, content, executable in planned:
        write_if_changed(
            path,
            content,
            executable=executable,
            check=check,
            drift=drift,
        )
    return drift


def main() -> int:
    try:
        release = source_version()
    except ValueError as exc:
        print(f"coord-install: {exc}", file=sys.stderr)
        return 2
    parser = argparse.ArgumentParser(prog="coord-install")
    parser.add_argument("target", nargs="?", default=".")
    parser.add_argument("--project-id", default="")
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument("--check", action="store_true", help="check for drift")
    mode.add_argument(
        "--doctor",
        action="store_true",
        help="diagnose installation version and drift without changing files",
    )
    mode.add_argument(
        "--repair",
        action="store_true",
        help="repair managed files and hooks while preserving unrelated content",
    )
    parser.add_argument(
        "--version", action="version", version=f"coord-install {release}"
    )
    args = parser.parse_args()
    target = Path(args.target).expanduser().resolve()
    observed_version = installed_version(target)
    try:
        drift = install(
            target,
            requested_id=args.project_id,
            check=args.check or args.doctor,
        )
    except ValueError as exc:
        print(f"coord-install: {exc}", file=sys.stderr)
        return 2
    if args.check or args.doctor:
        if args.doctor:
            print(f"coord-install: source version {release}")
            print(f"coord-install: installed version {observed_version}")
        if drift:
            print(f"coord-install: drift ({len(drift)} files)")
            for path in drift:
                print(f"  {path}")
            if args.doctor:
                command = " ".join(
                    shlex.quote(value)
                    for value in (str(HERE / "install"), str(target), "--repair")
                )
                print(f"coord-install: repair with: {command}")
            return 1
        print(f"coord-install: installation matches {release}")
        return 0
    if args.repair and drift:
        action = "repaired"
    elif drift:
        action = "updated"
    else:
        action = "already current"
    print(f"coord-install: {action} {release}")
    print(f"target {target}")
    return 0


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