diff --git a/scripts/cleanup-worktrees-aggressive.py b/scripts/cleanup-worktrees-aggressive.py new file mode 100755 index 00000000..9cf86d7b --- /dev/null +++ b/scripts/cleanup-worktrees-aggressive.py @@ -0,0 +1,275 @@ +#!/usr/bin/env python3 +""" +Aggressive cleanup of stale Claude Code background-session worktrees. + +Unlike safe `cleanup-merged-worktrees.sh`, this queries Forgejo PR API to +detect squash-merged branches (where local SHA isn't ancestor of main +because Forgejo squash created a new commit). + +Decision matrix per worktree branch: + + PR state | Action + ----------------------|--------------------------------------------- + merged (any style) | REMOVE + closed (no merge) | REMOVE (work abandoned) + open | KEEP (active PR in progress) + no PR + age <14d | KEEP (recent WIP, no PR yet) + no PR + age >=14d | REMOVE (stale orphan) + +Requires: + - FORGEJO_TOKEN env var (or in ~/.claude/settings.json env block) + - FORGEJO_URL env var (default: https://git.gendsgn.ru) + +Usage: + python scripts/cleanup-worktrees-aggressive.py --dry-run # show plan + python scripts/cleanup-worktrees-aggressive.py # execute + AGE_DAYS=30 python scripts/cleanup-worktrees-aggressive.py # override threshold +""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +import time +import urllib.error +import urllib.request +from collections import Counter +from pathlib import Path + +OWNER = "lekss361" +REPO_NAME = "gendesign" +FORGEJO_URL = os.environ.get("FORGEJO_URL", "https://git.gendsgn.ru") +TOKEN = os.environ.get("FORGEJO_TOKEN") +AGE_DAYS = int(os.environ.get("AGE_DAYS", "14")) +DRY_RUN = "--dry-run" in sys.argv + +REPO_ROOT = Path(__file__).resolve().parent.parent + + +def run(cmd: list[str], cwd: Path | None = None, check: bool = True) -> str: + """Run a subprocess and return stdout, swallowing errors when check=False.""" + result = subprocess.run( + cmd, + cwd=cwd or REPO_ROOT, + capture_output=True, + text=True, + check=False, + ) + if check and result.returncode != 0: + raise RuntimeError(f"{' '.join(cmd)}\n{result.stderr}") + return result.stdout + + +def fetch_all_prs() -> dict[str, dict]: + """Returns {branch_name: {state, merged}} for all PRs across all states.""" + if not TOKEN: + sys.exit( + "ERROR: FORGEJO_TOKEN env var not set.\n" + "Hint: $env:FORGEJO_TOKEN = '' (PowerShell) or export it before running." + ) + + prs: dict[str, dict] = {} + page = 1 + api = f"{FORGEJO_URL}/api/v1/repos/{OWNER}/{REPO_NAME}/pulls" + + while page <= 30: + url = f"{api}?state=all&limit=50&page={page}&sort=newest" + req = urllib.request.Request( + url, headers={"Authorization": f"token {TOKEN}"} + ) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + batch = json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + sys.exit(f"ERROR: Forgejo API {url} → {e.code} {e.reason}") + except Exception as e: + sys.exit(f"ERROR: fetch {url} failed: {e}") + + if not batch: + break + + for pr in batch: + branch = pr.get("head", {}).get("ref") + if not branch: + continue + # Newest-wins: only set if not already seen (sort=newest gives + # most recent first, so first occurrence is the newest PR for + # that branch) + prs.setdefault( + branch, + { + "state": pr.get("state"), + "merged": bool(pr.get("merged")), + "number": pr.get("number"), + }, + ) + + page += 1 + + return prs + + +def get_worktrees() -> list[tuple[str, str]]: + """Returns list of (path, branch_short_name) tuples, skipping main.""" + out = run(["git", "worktree", "list", "--porcelain"]) + worktrees: list[tuple[str, str]] = [] + current = {} + for line in out.splitlines(): + if line.startswith("worktree "): + current = {"path": line[len("worktree ") :]} + elif line.startswith("branch "): + current["branch"] = line[len("branch ") :] + elif line == "" and current: + branch = current.get("branch", "") + if branch.startswith("refs/heads/"): + short = branch[len("refs/heads/") :] + worktrees.append((current["path"], short)) + current = {} + if current and "branch" in current: + branch = current["branch"] + if branch.startswith("refs/heads/"): + worktrees.append((current["path"], branch[len("refs/heads/") :])) + return worktrees + + +def branch_age_days(path: str) -> int | None: + """Returns age in days of branch HEAD, or None on error.""" + try: + ts = run(["git", "-C", path, "log", "-1", "--format=%ct"], check=False).strip() + if not ts: + return None + return (int(time.time()) - int(ts)) // 86400 + except Exception: + return None + + +def main() -> int: + if DRY_RUN: + print("DRY RUN — no worktrees will be removed\n") + + print("Fetching all PRs from Forgejo (paginated, up to 1500)...") + prs = fetch_all_prs() + print(f"Indexed {len(prs)} unique branch→PR mappings.\n") + + worktrees = get_worktrees() + print(f"Scanning {len(worktrees)} worktrees...\n") + + categories: Counter[str] = Counter() + to_remove: list[tuple[str, str, str]] = [] # (path, branch, reason) + repo_root_str = str(REPO_ROOT) + + for path, branch in worktrees: + # Don't touch the worktree we're standing in. + if os.path.normcase(os.path.normpath(path)) == os.path.normcase( + os.path.normpath(repo_root_str) + ): + categories["skipped"] += 1 + continue + + pr = prs.get(branch) + if pr: + if pr["merged"]: + cat = "merged" + reason = f"remove (merged PR #{pr['number']})" + elif pr["state"] == "closed": + cat = "closed" + reason = f"remove (closed PR #{pr['number']}, no merge)" + else: + cat = "open" + reason = f"keep (open PR #{pr['number']})" + else: + age = branch_age_days(path) + if age is None: + cat = "other" + reason = "keep (no PR, age unknown)" + elif age >= AGE_DAYS: + cat = "stale" + reason = f"remove (no PR, stale {age}d ≥ {AGE_DAYS}d)" + else: + cat = "recent" + reason = f"keep (no PR, recent {age}d)" + + print(f" [{cat:>6}] {branch} → {reason}") + categories[cat] += 1 + + if cat in {"merged", "closed", "stale"}: + to_remove.append((path, branch, reason)) + + if not DRY_RUN: + print(f"\nRemoving {len(to_remove)} worktree(s)...") + failed = 0 + for path, branch, _ in to_remove: + # `--force --force` (double flag) overrides Claude Code supervisor + # locks (`claude agent (pid X)` lock messages). + result = subprocess.run( + ["git", "worktree", "remove", "--force", "--force", path], + capture_output=True, + text=True, + cwd=REPO_ROOT, + ) + if result.returncode != 0: + failed += 1 + err = result.stderr.strip().splitlines()[-1] if result.stderr else "unknown" + print(f" FAILED: {branch} → {err}", file=sys.stderr) + run(["git", "worktree", "prune"], check=False) + if failed: + print(f"\nWARN: {failed} worktree(s) failed to remove (see stderr above)") + + # Phase 2: orphan-dir cleanup. After `git worktree remove`, git no longer + # tracks the dir, but filesystem may still have it (Windows file locks + # held by Claude Code supervisor, etc). Try to remove via shutil. + wt_root = REPO_ROOT / ".claude" / "worktrees" + if wt_root.is_dir(): + fs_dirs = {p.name for p in wt_root.iterdir() if p.is_dir()} + git_dirs = {Path(p).name for p, _ in get_worktrees()} + orphans = fs_dirs - git_dirs + if orphans: + print(f"\nFound {len(orphans)} orphan dir(s) (git-untracked, FS leftover)...") + orphan_removed = 0 + orphan_locked = 0 + for o in orphans: + try: + shutil.rmtree(wt_root / o, ignore_errors=False) + orphan_removed += 1 + except PermissionError: + orphan_locked += 1 + except Exception as e: + print(f" ORPHAN FAILED: {o} → {type(e).__name__}", file=sys.stderr) + print(f" Removed {orphan_removed}, locked (need Claude Code restart) {orphan_locked}") + if orphan_locked: + print( + f"\n Hint: {orphan_locked} dirs held by Claude Code supervisor " + f"(open file handles). After next Claude Code restart, re-run this " + f"script — they will be removable then." + ) + + print("\nSummary:") + print(f" Remove — merged: {categories['merged']}") + print(f" Remove — closed: {categories['closed']}") + print(f" Remove — stale: {categories['stale']} (no PR, age >={AGE_DAYS}d)") + print(f" Keep — open PR: {categories['open']}") + print( + f" Keep — recent: {categories['recent']} (no PR yet, age <{AGE_DAYS}d)" + ) + print(f" Keep — other: {categories['other']} (no PR, unknown age)") + print(f" Skipped (current): {categories['skipped']}") + + total_remove = categories["merged"] + categories["closed"] + categories["stale"] + print() + if DRY_RUN: + print( + f"DRY RUN — would remove {total_remove} worktree(s). " + "Re-run without --dry-run to execute." + ) + else: + after = len(get_worktrees()) + 1 # +1 for main + print(f"Worktrees remaining: {after}") + + return 0 + + +if __name__ == "__main__": + sys.exit(main())