Agent-System-Dev/scripts/check-worktree.py

#!/usr/bin/env python3
"""Fail closed before task synchronization, verification, or worktree disposal."""
import os
from pathlib import Path
import subprocess
import sys


def git(repo, *args):
    result = subprocess.run(["git", "-C", str(repo), *args], stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE)
    if result.returncode:
        raise RuntimeError(f"Git check failed in {repo}: {' '.join(args)}")
    return result.stdout


def text(repo, *args):
    return os.fsdecode(git(repo, *args)).strip()


def check_operations(repo):
    for marker in ("index.lock", "HEAD.lock", "MERGE_HEAD", "CHERRY_PICK_HEAD",
                   "REVERT_HEAD", "rebase-merge", "rebase-apply", "BISECT_LOG"):
        path = Path(text(repo, "rev-parse", "--path-format=absolute", "--git-path", marker))
        if path.exists():
            raise RuntimeError(f"active Git operation/lock in {repo}: {marker}; owner must finish it")


def check_clean(repo, ignored=False, allow_gitlink_drift=False):
    args = ["status", "--porcelain=v1", "--untracked-files=all",
            "--ignore-submodules=all" if allow_gitlink_drift else "--ignore-submodules=none"]
    if ignored:
        args.append("--ignored=matching")
    status = git(repo, *args)
    if status:
        raise RuntimeError(f"unsaved, untracked{', or ignored' if ignored else ''} files in {repo}; preserve them before continuing")


def submodules(repo):
    for item in git(repo, "ls-files", "--stage", "-z").split(b"\0"):
        if not item:
            continue
        meta, path = item.split(b"\t", 1)
        mode, sha, stage = meta.split()
        if stage != b"0":
            raise RuntimeError(f"unmerged index in {repo}; resolve conflicts first")
        if mode == b"160000":
            yield repo / os.fsdecode(path), sha.decode("ascii")


def check_children(repo, mode):
    for child, expected in submodules(repo):
        if child.is_symlink():
            raise RuntimeError(f"submodule path is a symbolic link: {child}")
        if not (child / ".git").exists():
            if mode in ("verify", "after-sync", "finish"):
                raise RuntimeError(f"uninitialized submodule: {child}; initialize and inspect it first")
            continue
        actual = text(child, "rev-parse", "HEAD")
        check_operations(child)
        check_clean(child, ignored=(mode == "finish"), allow_gitlink_drift=(mode == "before-update"))
        branch = text(child, "branch", "--show-current")
        if mode == "before-update" and branch and actual != expected:
            raise RuntimeError(f"submodule {child} owns branch {branch}; keep that work and explicitly reconcile pin {expected} before retrying task-sync")
        if mode in ("verify", "after-sync", "finish") and actual != expected:
            raise RuntimeError(f"submodule SHA drift: {child}: expected {expected}, actual {actual}")
        if mode == "finish" and git(child, "rev-list", "--all", "--reflog", "--not", expected).strip():
            raise RuntimeError(f"submodule has commits outside its pinned history: {child}; preserve branches, stash and reflogs before removal")
        check_children(child, mode)


def materialize(repo):
    """Check out one level, then inspect the newly exposed nested pins before moving them."""
    for child, expected in submodules(repo):
        check_children(repo, "before-update")
        result = subprocess.run(["git", "-C", str(repo), "submodule", "update", "--init",
                                 "--checkout", "--", str(child.relative_to(repo))])
        if result.returncode:
            raise RuntimeError(f"component materialization failed: {child}; preserve and reconcile before retrying")
        materialize(child)


def check_finish_root(repo):
    if Path.cwd().resolve() != repo:
        raise RuntimeError("task-finish must run at the registered worktree root, not a subdirectory")
    records = git(repo, "worktree", "list", "--porcelain", "-z").split(b"\0\0")
    parsed = [record.split(b"\0") for record in records if record]
    if not parsed or not parsed[0][0].startswith(b"worktree "):
        raise RuntimeError("cannot identify the main checkout")
    main = Path(os.fsdecode(parsed[0][0][9:])).resolve()
    if main == repo:
        raise RuntimeError("task-finish must run in a linked task worktree, never the main checkout")
    matches = [r for r in parsed if r[0].startswith(b"worktree ")
               and Path(os.fsdecode(r[0][9:])).resolve() == repo]
    if len(matches) != 1:
        raise RuntimeError("worktree is not registered exactly once; preserve it for owner inspection")
    if any(field.startswith((b"locked", b"prunable", b"bare", b"detached")) for field in matches[0][1:]):
        raise RuntimeError("locked, detached, bare, or prunable worktree must be preserved")
    common = Path(text(repo, "rev-parse", "--path-format=absolute", "--git-common-dir"))
    if any(common.rglob("*.lock")):
        raise RuntimeError("Git lock exists in shared repository metadata; wait for its owner")


def main():
    mode = sys.argv[1] if len(sys.argv) > 1 else "verify"
    if mode not in ("verify", "before-sync", "before-update", "materialize", "after-sync", "finish"):
        raise RuntimeError("usage: check-worktree.py [verify|before-sync|before-update|materialize|after-sync|finish]")
    repo = Path(text(Path.cwd(), "rev-parse", "--show-toplevel")).resolve()
    check_operations(repo)
    if mode == "finish":
        check_finish_root(repo)
    if mode in ("before-sync", "after-sync", "finish"):
        check_clean(repo, ignored=(mode == "finish"))
    if mode == "materialize":
        materialize(repo)
    else:
        check_children(repo, mode)
    print(f"[worktree] {mode}: recursive component checks passed")


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)