#!/usr/bin/env python3 """Create durable WIP refs without mutating source worktrees or indexes.""" from __future__ import annotations import argparse import datetime as dt import hashlib import json import os from pathlib import Path, PurePosixPath import re import subprocess import tempfile REF_ROOT = "refs/wip-preserve/20260812" VALUABLE_SUFFIXES = { ".c", ".cc", ".css", ".csv", ".go", ".h", ".html", ".java", ".js", ".json", ".jsx", ".md", ".mjs", ".php", ".py", ".rb", ".rs", ".scss", ".sh", ".sql", ".svg", ".toml", ".ts", ".tsx", ".txt", ".vue", ".xml", ".yaml", ".yml", } EXCLUDED_PARTS = { ".cache", ".git", ".idea", ".next", ".parcel-cache", ".pytest_cache", ".turbo", ".worktrees", "__pycache__", "backup", "backups", "build", "cache", "coverage", "dist", "logs", "node_modules", "runtime", "tmp", "vendor", } SENSITIVE_NAMES = { ".env", ".env.local", ".env.production", ".npmrc", ".pypirc", "admin.json", "auth.json", "credentials.json", "secrets.json", } SENSITIVE_RE = re.compile(r"(^|[-_.])(auth|credential|private[-_]?key|secret|token)([-_.]|$)", re.I) UNSAFE_SCRIPT = "plugins/international-press-zone/tools/tmp/.ctx-mode-yE4PFU/script.sh" def git(repo: Path, *args: str, env: dict[str, str] | None = None, check: bool = True) -> bytes: merged = os.environ.copy() if env: merged.update(env) result = subprocess.run(["git", "-C", str(repo), *args], env=merged, capture_output=True) if check and result.returncode: raise RuntimeError(f"git {' '.join(args)} failed ({result.returncode}): {result.stderr.decode(errors='replace').strip()}") return result.stdout def slug(value: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip("/")) return cleaned.strip(".-") or "detached" def status(repo: Path) -> bytes: return git(repo, "status", "--porcelain=v2", "-z", "--untracked-files=all") def status_paths(raw: bytes) -> tuple[set[str], set[str], set[str]]: tracked: set[str] = set() untracked: set[str] = set() types: set[str] = set() records = raw.split(b"\0") i = 0 while i < len(records): record = records[i] i += 1 if not record: continue text = record.decode("utf-8", "surrogateescape") kind = text[:1] if kind == "1": fields = text.split(" ", 8) tracked.add(fields[8]) xy = fields[1] if xy[0] != ".": types.add("staged") if xy[1] != ".": types.add("dirty") elif kind == "2": fields = text.split(" ", 9) tracked.add(fields[9]) if i < len(records): tracked.add(records[i].decode("utf-8", "surrogateescape")) i += 1 xy = fields[1] if xy[0] != ".": types.add("staged") if xy[1] != ".": types.add("dirty") elif kind == "u": tracked.add(text.split(" ", 10)[10]) types.add("conflicted") elif kind == "?": untracked.add(text[2:]) types.add("untracked") return tracked, untracked, types def exclusion_reason(path: str, *, untracked: bool) -> str | None: pure = PurePosixPath(path) lower_parts = {part.lower() for part in pure.parts} name = pure.name.lower() if path == UNSAFE_SCRIPT: return "security-unsafe generated script; evidence only" if lower_parts & EXCLUDED_PARTS: return "dependency/generated/cache/log/runtime/backup/nested-worktree class" if name in SENSITIVE_NAMES or name.startswith(".env.") or SENSITIVE_RE.search(name): return "secret/auth/env-sensitive path" if name.endswith((".log", ".pid", ".pyc", ".swp", ".tmp", ".bak", "~")): return "cache/log/runtime/backup file" if untracked and pure.suffix.lower() not in VALUABLE_SUFFIXES: return "untracked non-source/non-doc extension" return None def worktrees(repo: Path) -> list[dict[str, str]]: rows: list[dict[str, str]] = [] current: dict[str, str] = {} for line in git(repo, "worktree", "list", "--porcelain").decode().splitlines() + [""]: if not line: if current: rows.append(current) current = {} continue key, _, value = line.partition(" ") current[key] = value return rows def update_ref(repo: Path, ref: str, oid: str) -> None: existing = git(repo, "rev-parse", "--verify", "--quiet", ref, check=False).decode().strip() if existing and existing != oid: raise RuntimeError(f"ref collision: {ref} already resolves to {existing}, expected {oid}") if not existing: git(repo, "update-ref", "-m", "wip-preserve 2026-08-12", ref, oid, "") def commit_snapshot(repo: Path, wt: Path, head: str, included_untracked: list[str], excluded_tracked: list[str], label: str) -> tuple[str, list[str]]: common_raw = Path(git(repo, "rev-parse", "--git-common-dir").decode().strip()) common = common_raw if common_raw.is_absolute() else repo / common_raw common = common.resolve() fd, index_name = tempfile.mkstemp(prefix="wip-preserve-index-", dir=common) os.close(fd) os.unlink(index_name) env = {"GIT_INDEX_FILE": index_name, "GIT_WORK_TREE": str(wt)} try: git(repo, "read-tree", head, env=env) git(repo, "add", "-u", "--", ":/", env=env) for path in excluded_tracked: git(repo, "reset", "-q", head, "--", path, env=env) for path in included_untracked: git(repo, "add", "--", path, env=env) expected = sorted(filter(None, git(repo, "diff", "--cached", "--name-only", "-z", head, env=env).decode("utf-8", "surrogateescape").split("\0"))) if not expected: return head, [] tree = git(repo, "write-tree", env=env).decode().strip() message = f"WIP preserve 2026-08-12: {label}\n" result = subprocess.run( ["git", "-C", str(repo), "commit-tree", tree, "-p", head], input=message.encode(), capture_output=True, env={**os.environ, **env}, check=True, ) return result.stdout.decode().strip(), expected finally: try: os.unlink(index_name) except FileNotFoundError: pass def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("repo", type=Path) parser.add_argument("output", type=Path) args = parser.parse_args() repo = Path(git(args.repo, "rev-parse", "--show-toplevel").decode().strip()) remote_url = git(repo, "remote", "get-url", "origin").decode().strip() remote_head = git(repo, "symbolic-ref", "refs/remotes/origin/HEAD").decode().strip() baseline = git(repo, "rev-parse", remote_head).decode().strip() rows = worktrees(repo) before = {row["worktree"]: hashlib.sha256(status(Path(row["worktree"]))).hexdigest() for row in rows} manifest: dict[str, object] = { "generated_at_utc": dt.datetime.now(dt.timezone.utc).isoformat(), "repository": str(repo), "canonical_remote": remote_url, "remote_default": remote_head, "baseline": baseline, "ref_root": REF_ROOT, "worktrees": [], "branch_only": [], "stashes": [], "unsafe_script_evidence": None, "existing_protected_namespaces_mutated": False, } checked_out: set[str] = set() for row in rows: wt = Path(row["worktree"]) head = row["HEAD"] branch = row.get("branch", "detached") if branch != "detached": checked_out.add(branch) label = slug(branch.removeprefix("refs/heads/") if branch != "detached" else wt.name) tip_ref = f"{REF_ROOT}/tips/worktree/{label}" update_ref(repo, tip_ref, head) raw = status(wt) tracked, untracked, wip_types = status_paths(raw) exclusions: dict[str, str] = {} included_untracked: list[str] = [] for path in sorted(tracked): reason = exclusion_reason(path, untracked=False) if reason: exclusions[path] = reason for path in sorted(untracked): reason = exclusion_reason(path, untracked=True) if reason: exclusions[path] = reason else: included_untracked.append(path) snapshot, expected = commit_snapshot(repo, wt, head, included_untracked, sorted(path for path in tracked if path in exclusions), label) snapshot_ref = None if snapshot != head: snapshot_ref = f"{REF_ROOT}/snapshots/{label}" update_ref(repo, snapshot_ref, snapshot) actual = sorted(filter(None, git(repo, "diff-tree", "--no-commit-id", "--name-only", "-r", "-z", head, snapshot).decode("utf-8", "surrogateescape").split("\0"))) if actual != expected: raise RuntimeError(f"snapshot path mismatch for {wt}: expected {expected!r}, got {actual!r}") manifest["worktrees"].append({ "path": str(wt), "branch": branch, "head": head, "wip_types": sorted(wip_types), "tip_ref": tip_ref, "snapshot_ref": snapshot_ref, "expected_snapshot_paths": expected, "excluded_paths": exclusions, "status_sha256_before": before[str(wt)], "refs_verified": True, "recovery": f"git switch -c recover-{label} {snapshot_ref or tip_ref}", }) local_branches = git(repo, "for-each-ref", "--format=%(refname)%00%(objectname)", "refs/heads").decode().splitlines() for item in local_branches: ref, oid = item.split("\0", 1) if ref in checked_out: continue preserve_ref = f"{REF_ROOT}/tips/branch-only/{slug(ref.removeprefix('refs/heads/'))}" update_ref(repo, preserve_ref, oid) manifest["branch_only"].append({"branch": ref, "head": oid, "tip_ref": preserve_ref, "verified": True, "recovery": f"git branch {ref.removeprefix('refs/heads/')} {preserve_ref}"}) stash_lines = git(repo, "stash", "list", "--format=%gd%x00%H%x00%gs").decode().splitlines() for index, item in enumerate(stash_lines): source, oid, subject = item.split("\0", 2) ref = f"{REF_ROOT}/stashes/{index:03d}-{slug(subject)[:80]}" update_ref(repo, ref, oid) manifest["stashes"].append({"source": source, "head": oid, "subject": subject, "ref": ref, "verified": True, "recovery": f"git stash apply {ref}"}) unsafe = repo / UNSAFE_SCRIPT if unsafe.exists() and unsafe.is_file(): manifest["unsafe_script_evidence"] = {"path": UNSAFE_SCRIPT, "sha256": hashlib.sha256(unsafe.read_bytes()).hexdigest(), "size": unsafe.stat().st_size, "provenance": "untracked generated tools/tmp context-mode script", "content_preserved": False} after = {row["worktree"]: hashlib.sha256(status(Path(row["worktree"]))).hexdigest() for row in rows} changed = {path: {"before": before[path], "after": after[path]} for path in before if before[path] != after[path]} if changed: raise RuntimeError(f"source worktree status fingerprints changed: {json.dumps(changed, sort_keys=True)}") manifest["status_fingerprints_unchanged"] = True manifest["counts"] = {"worktrees": len(rows), "branch_only": len(manifest["branch_only"]), "stashes": len(manifest["stashes"]), "refs": len(rows) + len(manifest["branch_only"]) + len(manifest["stashes"]) + sum(1 for row in manifest["worktrees"] if row["snapshot_ref"])} args.output.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") if __name__ == "__main__": main()