Light1YT
86e9ea2937
Многоагентный аудит + имплементация: один воркер на файл, точечные правки. Верификация: py_compile (47/47 .py) + tsc --noEmit (0 ошибок). Unit-тесты не прогонялись (окружение не поднято: rollup native dep / нет pytest-venv). Полностью исправлено (169): #1336, #1337, #1339, #1340, #1341, #1342, #1343, #1345, #1346, #1348, #1349, #1350, #1351, #1354, #1356, #1358, #1359, #1360, #1362, #1364, #1365, #1366, #1367, #1368, #1369, #1370, #1371, #1372, #1373, #1374, #1375, #1376, #1377, #1378, #1379, #1380, #1381, #1382, #1384, #1385, #1386, #1387, #1388, #1389, #1390, #1391, #1392, #1394, #1395, #1396, #1397, #1399, #1400, #1401, #1402, #1403, #1404, #1408, #1409, #1410, #1411, #1412, #1413, #1414, #1415, #1416, #1417, #1418, #1420, #1423, #1425, #1426, #1427, #1428, #1429, #1430, #1431, #1432, #1433, #1434, #1435, #1437, #1438, #1439, #1440, #1441, #1442, #1443, #1444, #1445, #1446, #1447, #1448, #1449, #1450, #1451, #1452, #1453, #1454, #1455, #1456, #1457, #1458, #1459, #1460, #1461, #1462, #1463, #1464, #1465, #1466, #1467, #1468, #1469, #1471, #1472, #1473, #1474, #1476, #1478, #1479, #1481, #1482, #1483, #1484, #1485, #1487, #1488, #1489, #1490, #1491, #1492, #1493, #1494, #1495, #1496, #1497, #1499, #1500, #1501, #1502, #1504, #1505, #1506, #1507, #1510, #1514, #1515, #1516, #1517, #1518, #1519, #1521, #1522, #1523, #1524, #1525, #1526, #1527, #1528, #1529, #1531, #1532, #1533, #1534, #1535, #1536, #1537, #1538 Частично (9, in-file часть, остаток cross-file): #1361, #1419, #1422, #1424, #1470, #1475, #1477, #1480, #1498 Требуют cross-file (3, не тронуты): #1338, #1363, #1421 Пропущено (1): #1539 Не входило в партию: 22 needs-Leha issue (нужны решения владельца). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
275 lines
9.9 KiB
Python
Executable file
275 lines
9.9 KiB
Python
Executable file
#!/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 True:
|
|
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())
|