merge: fix-governance-agent-worktree-restore-2026-09-22

scripts/agent_worktree.py

@@ -199,6 +199,33 @@ def assert_no_process_cwd(path):
             raise Refused(f"process {pid} still uses this worktree as cwd")
 
 
+def inspect_process_cwd(path):
+    """Structured process-cwd visibility for cleanup reports (never force-clears)."""
+    path = Path(path).resolve()
+    try:
+        for pid, cwd in iter_same_user_cwds():
+            try:
+                resolved = cwd.resolve()
+            except OSError:
+                return {'state': 'unreadable', 'unreadable': [str(pid)]}
+            if resolved == path or path in resolved.parents:
+                return {'state': 'occupied', 'pid': str(pid)}
+        return {'state': 'clear'}
+    except Refused as exc:
+        msg = str(exc)
+        if 'unsupported' in msg:
+            return {
+                'state': 'unsupported_platform',
+                'platform': sys.platform,
+                'message': msg,
+            }
+        if 'cannot verify cwd of same-user process' in msg:
+            # Extract pid list when present after the word process
+            return {'state': 'unreadable', 'unreadable': [msg], 'message': msg}
+        return {'state': 'unreadable', 'unreadable': [], 'message': msg}
+
+
+
 @contextmanager
 def removal_lock(main, path):
     # Every agentctl run/registration holds the shared side of this same lock.
@@ -259,10 +286,134 @@ def remove_checked(repo, main, branch, delete_branch, path):
     print(json.dumps({"removed": str(path), "branch": branch, "branch_deleted": delete_branch}))
 
 
+def inspect_worktree(repo, main, entry):
+    """Read-only cleanup readiness for one porcelain worktree entry."""
+    path = Path(entry.get("worktree", ""))
+    branch_ref = entry.get("branch", "")
+    report = {
+        "path": str(path),
+        "branch": branch_ref[len("refs/heads/"):] if branch_ref.startswith("refs/heads/") else branch_ref or None,
+        "verdict": "blocked",
+        "blockers": [],
+        "notes": [],
+    }
+    if "bare" in entry:
+        report["blockers"].append("bare_worktree")
+        return report
+    try:
+        safe_directory(main, path.resolve())
+    except Refused as exc:
+        report["blockers"].append(f"path:{exc}")
+        return report
+    if "locked" in entry:
+        report["blockers"].append("git_locked")
+    if "prunable" in entry:
+        report["blockers"].append("git_prunable")
+    if not path.is_dir():
+        report["blockers"].append("missing_directory")
+        return report
+    if not report["branch"]:
+        report["blockers"].append("detached_or_unmapped_branch")
+        return report
+    try:
+        actual = git(path, "symbolic-ref", "--quiet", "HEAD").stdout.strip()
+        if actual != f"refs/heads/{report['branch']}":
+            report["blockers"].append("branch_identity_mismatch")
+        head = git(path, "rev-parse", "HEAD").stdout.strip()
+        report["head"] = head
+        if git(main, "merge-base", "--is-ancestor", head, "refs/heads/dev", check=False).returncode:
+            report["blockers"].append("not_ancestor_of_dev")
+        else:
+            report["notes"].append("ancestor_of_dev")
+        for marker in ("MERGE_HEAD", "CHERRY_PICK_HEAD", "REVERT_HEAD", "rebase-merge", "rebase-apply", "BISECT_START"):
+            marker_path = Path(git(path, "rev-parse", "--path-format=absolute", "--git-path", marker).stdout.strip())
+            if marker_path.exists():
+                report["blockers"].append(f"unfinished_git:{marker}")
+        dirty = git(path, "status", "--porcelain=v1", "--untracked-files=all", "--ignored=matching").stdout
+        if dirty:
+            report["blockers"].append("dirty_or_ignored_files")
+            report["dirty_sample"] = dirty.splitlines()[:8]
+    except Refused as exc:
+        report["blockers"].append(f"git:{exc}")
+    check = control(repo, "worktree-check", "--path", str(path), check=False)
+    if check.returncode:
+        try:
+            payload = json.loads(check.stderr.strip().splitlines()[-1])
+            report["blockers"].append(f"registry:{payload.get('code', 'DENIED')}")
+            if payload.get("message"):
+                report["notes"].append(payload["message"])
+        except (ValueError, IndexError, json.JSONDecodeError):
+            report["blockers"].append("registry:worktree_check_failed")
+    else:
+        report["notes"].append("registry_clear")
+    process = inspect_process_cwd(path)
+    report["process"] = process
+    if process["state"] == "occupied":
+        report["blockers"].append(f"process_cwd:{process['pid']}")
+    elif process["state"] == "unreadable":
+        report["blockers"].append("process_cwd_unreadable")
+    elif process["state"] == "unsupported_platform":
+        report["blockers"].append("process_cwd_platform_pending")
+        report["notes"].append(process.get("message") or "platform process check pending")
+    key = hashlib.sha256(str(path.resolve()).encode()).hexdigest()
+    lock_path = main / ".locks" / "os" / f"worktree-{key}.lock"
+    if lock_path.exists():
+        fd = None
+        try:
+            fd = os.open(lock_path, os.O_RDWR | os.O_NOFOLLOW, 0o600)
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            fcntl.flock(fd, fcntl.LOCK_UN)
+            report["notes"].append("lifecycle_lock_idle")
+        except BlockingIOError:
+            report["blockers"].append("lifecycle_lock_busy")
+        except OSError as exc:
+            report["blockers"].append(f"lifecycle_lock:{exc}")
+        finally:
+            if fd is not None:
+                os.close(fd)
+    if not report["blockers"]:
+        report["verdict"] = "ready"
+    return report
+
+
+def cleanup_report(repo, main):
+    """Surface cleanup readiness without deleting anything."""
+    try:
+        control(repo, "status")
+        registry = "initialized"
+    except Exception:
+        registry = "unavailable"
+    items = []
+    for entry in worktrees(main):
+        path = entry.get("worktree")
+        if not path:
+            continue
+        resolved = Path(path).resolve()
+        if resolved == main.resolve():
+            continue
+        if resolved.parent != (main / ".worktrees").resolve():
+            items.append({
+                "path": str(resolved),
+                "branch": entry.get("branch"),
+                "verdict": "blocked",
+                "blockers": ["outside_dot_worktrees"],
+                "notes": ["not managed by agent-worktree cleanup"],
+            })
+            continue
+        items.append(inspect_worktree(repo, main, entry))
+    ready = sum(1 for item in items if item.get("verdict") == "ready")
+    print(json.dumps({
+        "status": "ok",
+        "registry": registry,
+        "worktrees": items,
+        "summary": {"total": len(items), "ready": ready, "blocked": len(items) - ready},
+    }, ensure_ascii=False, sort_keys=True))
+
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     commands = parser.add_subparsers(dest="command", required=True)
     commands.add_parser("list", help="read-only Git worktree listing")
+    commands.add_parser("cleanup-report", help="read-only cleanup readiness for .worktrees entries")
     for name in ("create", "remove"):
         child = commands.add_parser(name)
         child.add_argument("branch")
@@ -275,6 +426,9 @@ def main():
         if args.command == "list":
             print(git(root, "worktree", "list").stdout, end="")
             return 0
+        if args.command == "cleanup-report":
+            cleanup_report(repo, root)
+            return 0
         safe_branch(args.branch)
         with write_lock(repo, root):
             if args.command == "create":