weichen
b84667a697
Why: independent review found shared submodule metadata risk, quoted header leakage, stale JSON receipt blindness and missing abandon coverage. Evidence: SELF 92 governance regressions pass; the original review receipt and replay output are recorded in the task card. Impact: preserves external component worktrees, handles both evidence formats and adds targeted failure-path coverage; no product behavior changes. Refs: .task/feature-governance-reliability.md; .task/reviews/governance-reliability.json
149 lines
7.0 KiB
Python
149 lines
7.0 KiB
Python
#!/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 worktrees(repo):
|
|
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 any(not record[0].startswith(b"worktree ") for record in parsed):
|
|
raise RuntimeError(f"cannot identify registered worktrees in {repo}")
|
|
return parsed
|
|
|
|
|
|
def check_component_worktrees(repo, removal_roots):
|
|
common = Path(text(repo, "rev-parse", "--path-format=absolute", "--git-common-dir")).resolve()
|
|
if not any(common == root or root in common.parents for root in removal_roots):
|
|
return
|
|
current = Path(text(repo, "rev-parse", "--absolute-git-dir")).resolve()
|
|
for index, record in enumerate(worktrees(repo)):
|
|
other = Path(os.fsdecode(record[0][9:])).resolve()
|
|
# Git may report an absorbed submodule's primary checkout as its common
|
|
# directory. It is the current checkout only when no linked gitdir is used.
|
|
if other != repo.resolve() and not (index == 0 and current == common):
|
|
raise RuntimeError(f"registered component worktree {other} depends on metadata in {common}; preserve or relocate it before removal")
|
|
|
|
|
|
def check_children(repo, mode, removal_roots=()):
|
|
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
|
|
if mode == "finish":
|
|
check_component_worktrees(child, removal_roots)
|
|
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, removal_roots)
|
|
|
|
|
|
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")
|
|
parsed = worktrees(repo)
|
|
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)
|
|
removal_roots = ()
|
|
if mode == "finish":
|
|
check_finish_root(repo)
|
|
removal_roots = (repo, Path(text(repo, "rev-parse", "--absolute-git-dir")).resolve())
|
|
if mode in ("before-sync", "after-sync", "finish"):
|
|
check_clean(repo, ignored=(mode == "finish"))
|
|
if mode == "materialize":
|
|
materialize(repo)
|
|
else:
|
|
check_children(repo, mode, removal_roots)
|
|
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)
|