#!/usr/bin/env python3 """Preview or install maintained files; preserve unrelated files and back up edits.""" from __future__ import annotations import argparse from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import shutil import tempfile import uuid ROOT = Path(__file__).resolve().parents[1] def digest(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def plan(repo: Path, home: Path, include_prompt: bool = False) -> list: source = repo / "skills/codex-subagent-router" files = [(p, Path("skills/codex-subagent-router") / p.relative_to(source)) for p in sorted(source.rglob("*")) if p.is_file() and "__pycache__" not in p.parts and p.suffix != ".pyc"] if not (source / "SKILL.md").is_file(): raise ValueError("source SKILL.md missing") if include_prompt: files.append((repo / "prompts/AGENTS.md", Path("AGENTS.md"))) changes = [] for src, relative in files: if src.is_symlink(): raise ValueError("refusing symlink source: " + str(relative)) dest = home / relative for part in [dest, *dest.parents]: if part == home: break if part.is_symlink(): raise ValueError("refusing symlink target: " + str(relative)) if dest.exists() and not dest.is_file(): raise ValueError("target is not a regular file: " + str(relative)) before = dest.read_bytes() if dest.exists() else None after = src.read_bytes() if before != after: changes.append({"path": relative, "before": before, "after": after}) return changes def install(repo: Path, home: Path, include_prompt: bool = False, apply: bool = False) -> dict: home = home.expanduser().resolve() changes = plan(repo, home, include_prompt) result = {"changed_paths": [str(x["path"]) for x in changes], "applied": False, "backup": None} if not apply or not changes: return result # Check all baselines before any managed file is changed. for change in changes: dest = home / change["path"] current = dest.read_bytes() if dest.exists() else None if current != change["before"]: raise ValueError("target changed during planning: " + str(change["path"])) stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex[:8] backup = home / "backups/codex-subagent-router" / stamp backup.mkdir(parents=True, exist_ok=False) manifest = [] for change in changes: before = change["before"] if before is not None: saved = backup / change["path"] saved.parent.mkdir(parents=True, exist_ok=True) saved.write_bytes(before) manifest.append({"path": str(change["path"]), "existed": before is not None, "before_sha256": digest(before) if before is not None else None, "installed_sha256": digest(change["after"])}) (backup / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") for change in changes: dest = home / change["path"] dest.parent.mkdir(parents=True, exist_ok=True) temporary = None try: with tempfile.NamedTemporaryFile(dir=dest.parent, delete=False) as stream: temporary = Path(stream.name) stream.write(change["after"]) if dest.exists(): shutil.copymode(dest, temporary) temporary.replace(dest) finally: if temporary is not None and temporary.exists(): temporary.unlink() result.update(applied=True, backup=str(backup)) return result def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--codex-home", type=Path, default=Path(os.environ.get("CODEX_HOME") or "~/.codex")) parser.add_argument("--include-prompt", action="store_true", help="also replace the global AGENTS.md") parser.add_argument("--apply", action="store_true", help="write files; default is preview only") parser.add_argument("--check", action="store_true", help="exit 1 when maintained files differ") args = parser.parse_args() if args.apply and args.check: parser.error("--apply and --check are mutually exclusive") try: result = install(ROOT, args.codex_home, args.include_prompt, args.apply) print(json.dumps(result, ensure_ascii=False, indent=2)) return 1 if args.check and result["changed_paths"] else 0 except (OSError, ValueError) as error: print("Install failed: " + str(error)) return 1 if __name__ == "__main__": raise SystemExit(main())