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 stat
import sys
from typing import Any


HERE = Path(__file__).resolve().parent
CORE_FILES = (
    ".gitignore",
    "DESIGN.md",
    "README.md",
    "coord",
    "hook.py",
    "install",
    "session.py",
    "store.py",
    "workspace.py",
)
EXECUTABLES = {"coord", "install"}
PROJECT_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}")

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:
    return isinstance(item, dict) and isinstance(item.get("command"), str) and (
        ".coord/" in item["command"] or ".claude/coord/" in item["command"]
    )


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:
        os.write(fd, content)
        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 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}")
    selected_id = project_id(target, requested_id, check=check)
    drift: list[str] = []
    destination = target / ".coord"
    for name in CORE_FILES:
        write_if_changed(
            destination / name,
            (HERE / name).read_bytes(),
            executable=name in EXECUTABLES,
            check=check,
            drift=drift,
        )
    write_if_changed(
        destination / "project-id",
        f"{selected_id}\n".encode(),
        check=check,
        drift=drift,
    )

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

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

    codex = target / ".codex" / "hooks.json"
    expected_codex = (
        json.dumps(codex_hooks(read_json(codex)), indent=2, sort_keys=True) + "\n"
    ).encode()
    write_if_changed(codex, expected_codex, check=check, drift=drift)
    return drift


def main() -> int:
    parser = argparse.ArgumentParser(prog="coord-install")
    parser.add_argument("target", nargs="?", default=".")
    parser.add_argument("--project-id", default="")
    parser.add_argument("--check", action="store_true")
    args = parser.parse_args()
    try:
        drift = install(
            Path(args.target), requested_id=args.project_id, check=args.check
        )
    except ValueError as exc:
        print(f"coord-install: {exc}", file=sys.stderr)
        return 2
    if args.check:
        if drift:
            print(f"coord-install: drift ({len(drift)} files)")
            for path in drift:
                print(f"  {path}")
            return 1
        print("coord-install: installation matches")
        return 0
    print(f"coord-install: {'updated' if drift else 'already current'}")
    print(f"target {Path(args.target).expanduser().resolve()}")
    return 0


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