chore(claude): aggressive worktree cleanup via Forgejo PR API

Сложение к safe cleanup-merged-worktrees.sh — script на Python который
запрашивает Forgejo PR API чтобы поймать squash-merged branches (где local
SHA не ancestor remote main).

Decision matrix per worktree branch:
- merged (any style)    → REMOVE
- closed (no merge)     → REMOVE (abandoned)
- open                  → KEEP (active PR)
- no PR + age <14d      → KEEP (recent WIP)
- no PR + age >=14d     → REMOVE (stale orphan)

Two-phase removal:
1. git worktree remove --force --force (overrides Claude supervisor locks)
2. shutil.rmtree on orphan dirs (git-untracked but FS-leftover)

Graceful PermissionError handling — dirs locked by running Claude Code
supervisor printed with hint to re-run after restart.

Real run results (single pass, before this commit):
- Before: 102 worktrees, 14GB
- After:  13 git worktrees + 23 FS orphans (PermissionError-locked), 3.7GB
- Removed: 82 git entries (80 merged + 2 closed), -10.3GB (-74%)
- Remaining 12 keep (recent WIP, no PR yet) + 1 current branch
- 23 FS orphans cleanable after next Claude Code restart

Requires:
- FORGEJO_TOKEN env (already set via ~/.claude/settings.json)
- python 3 stdlib only (urllib + json + shutil — no jq/curl/external deps)

Usage:
  python scripts/cleanup-worktrees-aggressive.py --dry-run   # show plan
  python scripts/cleanup-worktrees-aggressive.py             # execute
  AGE_DAYS=30 python scripts/...                             # override threshold
This commit is contained in:
lekss361 2026-05-24 12:30:15 +03:00
parent 643064ee0c
commit bd73ff2209

View file

@ -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 = '<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 <id> (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())