Codex
116 lines
4.7 KiB
Python
116 lines
4.7 KiB
Python
#!/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())
|