`install.py`

167 lines
#!/usr/bin/env python3
"""Prepare, inspect, apply, or roll back a fixed list of local script adapters."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import stat
import subprocess
import sys

from terminal_status import atomic_write

PROJECT = Path(__file__).resolve().parent
USER_ROOT = Path.home()
PLAN = PROJECT / "verification/install-plan.json"
BUNDLE = PROJECT / "verification/install"


def digest(data):
    return hashlib.sha256(data).hexdigest()


def replacements():
    files = {}
    for color in ("green", "yellow", "purple", "orange", "pink"):
        path = f".claude/skills/{color}dot/{color}dot.sh"
        files[path] = (
            '#!/usr/bin/env bash\n'
            '# Shared semantic status; see Projects/terminal-status/README.md.\n'
            f'exec bash "$HOME/Projects/terminal-status/integrations/dot.sh" claude {color} "$@"\n')
    mapping = {
        ".claude/skills/colordots/paint-lib.sh": "paint-lib.sh",
        ".claude/hooks/dot-floor.sh": "dot-floor.sh",
        ".claude/skills/color/scripts/working-state.sh": "working-state.sh",
        ".claude/skills/color/scripts/paint-tab.sh": "paint-tab.sh",
        ".claude/skills/color/scripts/tab-state.sh": "tab-state.sh",
        ".agents/skills/color/scripts/semantic-dot.sh": "semantic-dot.sh",
        ".agents/skills/color/scripts/repaint-dot.sh": "repaint-dot.sh",
        ".claude/skills/tab-ticket/scripts/tab-ticket-set.sh": "tab-ticket-set.sh",
        ".claude/skills/tab-ticket/scripts/tab-ticket-title.sh": "tab-ticket-title.sh",
        ".claude/skills/dot/current.sh": "current-dot.sh",
        ".claude/skills/lightgreendot/lightgreendot.sh": "lightgreendot.sh",
    }
    for target, source in mapping.items():
        files[target] = (PROJECT / "integrations" / source).read_text()

    scanner = ".claude/skills/allcolordots/allcolordots.sh"
    baseline = BUNDLE / "before" / scanner
    text = (baseline if baseline.exists() else USER_ROOT / scanner).read_text()
    start = text.index("# Codex rewrites its title")
    end = text.index("# ---- JUMP:")
    text = text[:start] + text[end:]
    start = text.index("# ---- collect normalized")
    end = text.index("# ---- roster of ONE color")
    text = text[:start] + (
        '# ---- Shared state scanner: presentation never overrides canonical state ----\n'
        'collect_rows(){\n'
        '  python3 "$HOME/Projects/terminal-status/terminal_status.py" scan --tsv\n'
        '}\n\n') + text[end:]
    start = text.index("# Prefer semantic LIVE")
    end = text.index("# Usage:")
    text = text[:start] + (
        '# Status comes from the shared session-bound terminal-status engine.\n'
        '# Labels and dot files are presentation; exact main agent ownership is required.\n'
        '#\n') + text[end:]
    text = text.replace('CODEX_DOTS="$HOME/.codex/tab-dots"\n', '')
    text = text.replace('set -u\n', 'set -u\nset -o pipefail\n')
    text = "\n".join(line + ' || return $?' if 'local want=' in line and 'collect_rows' in line else line for line in text.split('\n'))
    text = text.replace('print_roster yellow; exit 0', 'print_roster yellow; exit $?')
    text = text.replace('print_roster "${2:-yellow}"; exit 0', 'print_roster "${2:-yellow}"; exit $?')
    text = text.replace("    exit 0 ;;", "    exit $? ;;")
    text = text.replace('rows="$(collect_rows)"', 'rows="$(collect_rows)" || exit $?')
    files[scanner] = text

    # Keep the existing SessionStart hook and initialize semantic status afterward.
    for target in (".claude/skills/color/scripts/color.sh",
                   ".agents/skills/color/scripts/color.sh"):
        text = (USER_ROOT / target).read_text()
        marker = "# Shared terminal status follows the cosmetic session color."
        if marker not in text:
            text += (
                '\n' + marker + '\n'
                'python3 "$HOME/Projects/terminal-status/terminal_status.py" start --quiet || true\n')
        files[target] = text
    return files


def prepare(extend=False):
    if PLAN.exists() and not extend:
        raise SystemExit("Plan already exists; inspect it instead of replacing the baseline")
    old = {item["relative"]: item for item in json.loads(PLAN.read_text())} if extend else {}
    entries = []
    for relative, content in replacements().items():
        target = USER_ROOT / relative
        if target.is_symlink():
            raise SystemExit("Refusing to replace symlink: " + str(target))
        original = target.read_bytes()
        before = BUNDLE / "before" / relative
        after = BUNDLE / "after" / relative
        if relative in old:
            if digest(original) not in (old[relative]["after_sha256"], old[relative]["before_sha256"], old[relative].get("upgrade_sha256")):
                raise SystemExit("Concurrent adapter edit: " + relative)
        else:
            atomic_write(before, original.decode())
        atomic_write(after, content)
        subprocess.run(["bash", "-n", str(after)], check=True)
        entries.append({
            "relative": relative, "target": str(target),
            "before": str(before), "after": str(after),
            "before_sha256": digest(before.read_bytes()),
            "after_sha256": digest(content.encode()),
            "upgrade_sha256": digest(original),
            "mode": stat.S_IMODE(target.stat().st_mode),
        })
    atomic_write(PLAN, json.dumps(entries, indent=2) + "\n")
    print(f"Prepared {len(entries)} script changes: {PLAN}")


def change(rollback=False):
    entries = json.loads(PLAN.read_text())
    expected = "after_sha256" if rollback else "before_sha256"
    desired = "before_sha256" if rollback else "after_sha256"
    source_key = "before" if rollback else "after"
    # Check every preimage before the first mutation; preserve concurrent edits.
    for item in entries:
        target = Path(item["target"])
        if target.is_symlink():
            raise SystemExit("Refusing to replace symlink: " + str(target))
        current = digest(target.read_bytes())
        allowed = (item[expected], item[desired], item.get("upgrade_sha256")) if not rollback else (item[expected], item[desired])
        if current not in allowed:
            raise SystemExit("Concurrent edit; no changes applied: " + str(target))
        if digest(Path(item[source_key]).read_bytes()) != item[desired]:
            raise SystemExit("Changed install bundle; no changes applied")
    for item in entries:
        target = Path(item["target"])
        if digest(target.read_bytes()) == item[desired]:
            continue
        atomic_write(target, Path(item[source_key]).read_text())
        os.chmod(target, item["mode"])
    print(("Restored" if rollback else "Installed") + f" {len(entries)} script adapters")


def check():
    failed = []
    for item in json.loads(PLAN.read_text()):
        if digest(Path(item["target"]).read_bytes()) != item["after_sha256"]:
            failed.append(item["target"])
    if failed:
        print(json.dumps({"ok": False, "drift": failed}, indent=2))
        return 1
    print(json.dumps({"ok": True, "adapters": len(json.loads(PLAN.read_text()))}))
    return 0


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("action", choices=["prepare", "extend", "apply", "rollback", "check"])
    args = parser.parse_args()
    if args.action in ("prepare", "extend"):
        prepare(extend=args.action == "extend")
    elif args.action == "check":
        sys.exit(check())
    else:
        change(rollback=args.action == "rollback")