claude-harness/docs/retro/scripts/retro_leftovers.py

#!/usr/bin/env python3
"""Хвосты чеклиста закрытия по текущему состоянию репозиториев проекта или workspace (B12, #29).

Запуск: python3 retro_leftovers.py <папка проекта или workspace> [--fetch] [--markdown]
Репозитории: сам git-проект или каталоги из таблицы «| Каталог | Репозиторий | Роль |» в CLAUDE.md workspace.
Хвост: слитая в origin/main ветка на сервере или локально, worktree на ней или prunable, .superpowers/sdd/*,
неархивный openspec/changes/<id>/ в origin/main, основной checkout позади origin/main или с изменениями.
Ничего не меняет; --fetch делает только git fetch --prune origin. Спека: openspec/specs/harness-retro/spec.md.
"""
import collections
import glob
import json
import os
import re
import subprocess
import sys

TABLE = re.compile(r"^\|\s*Каталог\s*\|\s*Репозиторий\s*\|\s*Роль\s*\|")
PROTECTED = ("main", "master")


def git(repo, *args):
    env = {k: v for k, v in os.environ.items() if k not in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE")}
    r = subprocess.run(["git", "-C", repo] + list(args), stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                       universal_newlines=True, env=env)
    return r.stdout if r.returncode == 0 else None


def lines(s):
    return [l.strip() for l in (s or "").splitlines() if l.strip()]


def repos_of(project):
    project = os.path.abspath(project)
    if os.path.exists(os.path.join(project, ".git")):
        return [project]
    md, out, on = os.path.join(project, "CLAUDE.md"), [], False
    if not os.path.isfile(md):
        return out
    for line in open(md, encoding="utf-8", errors="replace"):
        if TABLE.match(line):
            on = True
            continue
        if not on:
            continue
        if not line.startswith("|"):
            break
        d = line.strip().strip("|").split("|")[0].strip()
        if not d or set(d) <= set("-: "):
            continue
        p = os.path.join(project, d.rstrip("/"))
        if os.path.exists(os.path.join(p, ".git")):
            out.append(p)
    return out


def worktrees(repo):
    out, cur = [], {}
    for l in (git(repo, "worktree", "list", "--porcelain") or "").splitlines() + [""]:
        if not l.strip():
            if cur:
                out.append(cur)
            cur = {}
            continue
        k, _, v = l.partition(" ")
        cur[k] = v or True
    return out


def leftovers(repo, fetch=False):
    items = []

    def add(kind, item, hint):
        items.append({"kind": kind, "repo": repo, "item": item, "hint": hint})

    if fetch:
        git(repo, "fetch", "--prune", "-q", "origin")
    sha = lambda ref: (git(repo, "rev-parse", "-q", "--verify", ref) or "").strip()
    tip = sha("origin/main")
    if not tip:
        add("no-origin-main", "origin/main", "нет origin/main: git fetch origin или проверить remote")
        return items
    for b in lines(git(repo, "branch", "-r", "--merged", "origin/main")):
        if "->" in b or not b.startswith("origin/") or b[len("origin/"):] in PROTECTED + ("HEAD",):
            continue
        if sha(b) != tip:
            add("remote-branch", b, "git push origin --delete " + b[len("origin/"):])
    stale = set()
    for b in lines(git(repo, "branch", "--merged", "origin/main", "--format=%(refname:short)")):
        if b not in PROTECTED and sha("refs/heads/" + b) != tip:
            stale.add(b)
    for l in lines(git(repo, "for-each-ref", "--format=%(refname:short) %(upstream:track)", "refs/heads")):
        name, _, track = l.partition(" ")
        if "gone" in track and name not in PROTECTED:
            stale.add(name)
    for b in sorted(stale):
        add("local-branch", b, "git branch -D " + b)
    wts = worktrees(repo)
    main_wt = wts[0]["worktree"] if wts else repo
    done = [main_wt]  # SDD workspace — хвост в основном checkout и в worktree закрытой задачи, не в активном
    for w in wts[1:]:
        br = str(w.get("branch") or "")
        if br.startswith("refs/heads/"):
            br = br[len("refs/heads/"):]
        if w.get("prunable") or br in stale:
            add("worktree", w["worktree"], "ExitWorktree, затем git worktree remove " + w["worktree"])
            done.append(w["worktree"])
    for w in done:
        for d in sorted(glob.glob(os.path.join(w, ".superpowers", "sdd", "*"))):
            if os.path.isdir(d):
                add("sdd-workspace", d, "rm -rf " + d)
    for c in lines(git(repo, "ls-tree", "-d", "--name-only", "origin/main", "openspec/changes/")):
        name = c.rstrip("/").rsplit("/", 1)[-1]
        if name != "archive":
            add("openspec-change", c, "openspec archive %s — отдельным PR" % name)
    behind = (git(main_wt, "rev-list", "--count", "HEAD..origin/main") or "0").strip()
    if behind not in ("", "0"):
        add("main-behind", main_wt, "git pull --ff-only (позади origin/main на %s)" % behind)
    dirty = [l for l in (git(main_wt, "status", "--porcelain") or "").splitlines() if l.strip()]
    if dirty:
        add("main-dirty", main_wt, "не трогать, перечислить: " + ", ".join(l[3:] for l in dirty[:5]))
    return items


def to_markdown(items):
    if not items:
        return "Хвостов закрытия нет."
    out = ["| Вид | Репозиторий | Что | Чем убрать |", "|---|---|---|---|"]
    for i in items:
        out.append("| %s | %s | `%s` | `%s` |" % (i["kind"], os.path.basename(i["repo"]), i["item"], i["hint"]))
    return "\n".join(out)


def main(argv):
    args = argv[1:]
    pos = [a for a in args if not a.startswith("--")]
    if not pos:
        print(__doc__.strip().splitlines()[2])
        return 1
    repos = repos_of(pos[0])
    if not repos:
        sys.stderr.write("retro_leftovers: в %s нет git-репозитория и таблицы репозиториев workspace\n" % pos[0])
        return 1
    items = [i for r in repos for i in leftovers(r, "--fetch" in args)]
    if "--markdown" in args:
        print(to_markdown(items))
    else:
        print(json.dumps({"repos": repos, "items": items,
                          "totals": dict(collections.Counter(i["kind"] for i in items))}, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))