Files
ISEP/scripts/worktree
T
isep-hand 7cacfdcf11 一個 repo 只有一個資料夾:分身住 repo 裡面的 .worktrees/,收工由同一支閘收掉(inkstone/ISEP#147)
leo 2026-09-07:「一個 repo 只看得到一個資料夾」「禁止這樣產出一大堆資料夾」。
根因:line-needs-own-worktree.sh 教工人開在 repo 旁邊,「收工 remove」只寫在閘訊息裡、沒機制驗,
每條線留一份 ⇒ tech_projects/ 長出 ISEP-wt-117/InkStoneCo-wt-112…一排。規則在文字裡不在機器上。

- scripts/worktree(新):open/close/list/sweep 四個動詞。分身住 <repo>/.worktrees/<票 repo>-<N>
  (走 .git/info/exclude,不動被追蹤的檔);「推了沒」只有這一份判準(三態:有/沒有/問不到,
  問不到不當成沒推也不收);sweep 預設只列不動,--apply 只收「推了且乾淨」,--skip 是操作者點名。
- hooks/line-needs-own-worktree.sh:擋下時印 `scripts/worktree open`(絕對路徑,貼了就跑);
  同一支多掛 PostToolUse Agent|Task——從【工單】認出分身,推了收、沒推點名分支+路徑(exit 2),
  不 force、不刪。+0 支、+1 條(61 支/86 條,當場數的)。
- docs/governance/worktree-placement.md(新)+ principles 一行:禁令與閘互指。
- 測試:line-needs-own-worktree.test.sh 77→110(G 群:閘印的 open 那行原樣跑、旁邊不多任何東西;
  H 群:PostToolUse 推了收/沒推點名/安靜的方向);scripts/test-worktree.sh 42 條(七個分身的 fixture,
  名字照 leo 看到的那排)。舊閘跑新測試 90/110,紅的 20 條正是這次改到的行為。
- permissions-allow.json 加 worktree 四種形狀(出路在雲端被分類器擋就走不通)。
- README/plugin.json/hooks-inventory 的數字在這棵樹上實數:61 支、86 條、62 支腳本。

待總管定版(改了會被載入的東西)。沒碰 hooks/lib/mainline.py。

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0178ef1fGw3XeZtpN7LaZrm4
2026-09-07 07:13:21 +00:00

554 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""worktree — 一條線的分身住在 repo **裡面**`.worktrees/`),收工時被收掉。
leo 2026-09-07inkstone/ISEP#147,一字不改):
「我看到的是一個 repo,你會產生大量的同一個頭不同尾巴的 repo,全部拆開就更亂了,
為什麼要產出新的 repo?這個 repo 的目的就是收納所有相關的內容,這樣散開造成凌亂」
「另外要在 ISEP 裡寫明,禁止這樣產出一大堆資料夾」
那些不是 repo,是 `hooks/line-needs-own-worktree.sh`inkstone/ISEP#109)教工人開在
repo **旁邊**的 git worktree——「收工時 `worktree remove`」只寫在閘訊息裡,沒有任何機制驗,
每個工人留一份,`tech_projects/` 就長成 `ISEP-wt-117``ISEP-wt115``InkStoneCo-wt-112`…。
**規則在文字裡,不在機器上。** 本工具把兩端都放到機器上:
worktree open <owner/repo#N> [--repo <路徑>] [-b <分支>]
開一份:<repo>/.worktrees/<repo名>-<N>,並把 `/.worktrees/` 寫進該 repo 的
.git/info/exclude(不動任何被追蹤的檔)。冪等:已經開過就印出路徑。
worktree close <owner/repo#N> [--repo <路徑>] [--scan <目錄>]... [--quiet]
收:分支推上遠端且工作樹乾淨 ⇒ remove+prune;否則離開碼 2、印出分支與路徑。
**不 force、不刪任何沒推的 commit。**
worktree list [<目錄>...]
列:每個 repo 的分身、分支、推了沒、乾不乾淨。
worktree sweep <目錄> [--apply] [--skip <repo名>]...
leo 在 Mac 上跑一次的整理工具:逐個 repo 列分身;預設只列不動,
`--apply` 才收——而且只收「推了且乾淨」的;沒推的列出來等他決定。
`--skip` 是操作者點名不碰的 repo(例:地端正在處理的線),不是判準。
「推了沒」的判準(跟 hooks/unpushed-police.sh 同一套,三態不是兩態):
0 遠端有這顆 commit 1 遠端沒有 2 問不到(沒 remote/網路不通)
2 一律不收——把「離線」當成「你沒推」再收掉,就是拿雜訊懲罰謹慎;
但 2 也一律**點名**,因為它同樣是留在那裡的一份分身。
判準是「這個目錄是不是某個 repo 的 worktree、它的分支推了沒、工作樹乾不乾淨」——
三件都是 git 自己回答的事實,沒有任何一條看名字、看關鍵字。
"""
import os
import re
import subprocess
import sys
WT_DIR = ".worktrees"
EXCLUDE_LINE = "/" + WT_DIR + "/"
TICKET_RE = re.compile(r"^([A-Za-z0-9_.-]+)/([A-Za-z0-9_.-]+)#(\d+)$")
GOVERNANCE = "docs/governance/worktree-placement.md"
GIT_ENV = dict(os.environ, GIT_TERMINAL_PROMPT="0", GIT_ASKPASS="/bin/echo",
GIT_SSH_COMMAND="ssh -oBatchMode=yes", LC_ALL="C")
def run(args, cwd=None, timeout=None):
try:
p = subprocess.run(args, cwd=cwd, env=GIT_ENV, capture_output=True, text=True, timeout=timeout)
except (OSError, subprocess.TimeoutExpired):
return 124, "", ""
return p.returncode, p.stdout, p.stderr
def git(repo, *args, timeout=None):
return run(["git", "-C", repo] + list(args), timeout=timeout)
def out(repo, *args):
rc, o, _ = git(repo, *args)
return o.strip() if rc == 0 else ""
def die(msg, code=1):
sys.stderr.write(msg.rstrip() + "\n")
sys.exit(code)
def self_path():
"""印給人複製的指令一律用這個(mistakes.md「閘印出來的那條出路跑不動」:相對路徑會對貼的人的 cwd 解析)。"""
return os.path.realpath(__file__)
def parse_ticket(s):
m = TICKET_RE.match((s or "").strip())
if not m:
die("票要寫成 owner/repo#N(例:inkstone/ISEP#147),收到的是:%r" % (s,), 1)
return m.group(1), m.group(2), m.group(3)
def wt_name(repo, num):
return "%s-%s" % (repo, num)
# ── git 事實 ────────────────────────────────────────────────────────────
def common_dir(path):
"""這個路徑所屬 repo 的共用 .git(主工作目錄那一份);不是 repo 回 None。"""
rc, o, _ = git(path, "rev-parse", "--path-format=absolute", "--git-common-dir")
if rc != 0:
rc, o, _ = git(path, "rev-parse", "--git-common-dir")
if rc != 0:
return None
o = os.path.join(path, o.strip()) if not os.path.isabs(o.strip()) else o.strip()
return os.path.realpath(o.strip())
def main_top(path):
"""主工作目錄(大家共用的那一份)的絕對路徑;bare/不是 repo 回 None。"""
cd = common_dir(path)
if not cd:
return None
top = os.path.dirname(cd) if os.path.basename(cd) == ".git" else None
if not top:
return None
rc, o, _ = git(top, "rev-parse", "--show-toplevel")
return os.path.realpath(o.strip()) if rc == 0 else None
def is_main_worktree(path):
rc, gd, _ = git(path, "rev-parse", "--path-format=absolute", "--git-dir")
rc2, cd, _ = git(path, "rev-parse", "--path-format=absolute", "--git-common-dir")
if rc != 0 or rc2 != 0:
return False
return os.path.realpath(gd.strip()) == os.path.realpath(cd.strip())
def list_linked(main):
"""`git worktree list --porcelain`:主工作目錄以外的每一份。"""
rc, o, _ = git(main, "worktree", "list", "--porcelain")
if rc != 0:
return []
entries, cur = [], None
for line in o.splitlines() + [""]:
if not line:
if cur:
entries.append(cur)
cur = None
continue
k, _, v = line.partition(" ")
if k == "worktree":
cur = {"path": os.path.realpath(v), "raw": v, "branch": "", "sha": "", "prunable": "", "bare": False}
elif cur is None:
continue
elif k == "HEAD":
cur["sha"] = v
elif k == "branch":
cur["branch"] = v[len("refs/heads/"):] if v.startswith("refs/heads/") else v
elif k == "prunable":
cur["prunable"] = v or "prunable"
elif k == "bare":
cur["bare"] = True
mt = os.path.realpath(main)
return [e for e in entries if e["path"] != mt and not e["bare"]]
def ls_remote_heads(repo, remote, cache={}):
key = (repo, remote)
if key not in cache:
rc, o, _ = git(repo, "ls-remote", "--heads", remote, timeout=15)
cache[key] = [ln.split() for ln in o.splitlines() if ln.strip()] if rc == 0 and o.strip() else None
return cache[key]
def remote_has_commit(repo, sha):
"""0 遠端有/1 遠端沒有/2 問不到(沿用 hooks/unpushed-police.sh 的三態)。"""
if not sha:
return 2
if out(repo, "branch", "-r", "--contains", sha):
return 0
asked = False
for rm in out(repo, "remote").split():
heads = ls_remote_heads(repo, rm)
if heads is None:
continue
asked = True
for rsha, _ref in heads:
if rsha == sha:
return 0
rc, _, _ = git(repo, "cat-file", "-e", rsha + "^{commit}")
if rc == 0:
rc2, _, _ = git(repo, "merge-base", "--is-ancestor", sha, rsha)
if rc2 == 0:
return 0
return 1 if asked else 2
def classify(main, e):
"""一份分身的事實:dirtypushedverdict。verdict ∈ collect|unpushed|dirty|unknown|prunable。"""
r = dict(e)
r["main"] = main
if e["prunable"] or not os.path.isdir(e["path"]):
r["dirty"] = 0
r["pushed"] = 2
r["verdict"] = "prunable"
return r
rc, st, _ = git(e["path"], "status", "--porcelain", "--untracked-files=normal")
r["dirty"] = len([ln for ln in st.splitlines() if ln.strip()]) if rc == 0 else 0
sha = e["sha"] or out(e["path"], "rev-parse", "HEAD")
r["sha"] = sha
r["pushed"] = remote_has_commit(main, sha)
if r["dirty"]:
r["verdict"] = "dirty"
elif r["pushed"] == 0:
r["verdict"] = "collect"
elif r["pushed"] == 1:
r["verdict"] = "unpushed"
else:
r["verdict"] = "unknown"
return r
def ensure_exclude(main):
cd = common_dir(main)
if not cd:
return False
info = os.path.join(cd, "info")
os.makedirs(info, exist_ok=True)
f = os.path.join(info, "exclude")
try:
lines = open(f, encoding="utf-8", errors="replace").read().splitlines() if os.path.exists(f) else []
except OSError:
lines = []
if EXCLUDE_LINE in [ln.strip() for ln in lines]:
return True
with open(f, "a", encoding="utf-8") as fh:
if lines and not lines[-1].endswith("\n"):
fh.write("\n")
fh.write("# 一條線的分身住這裡,不住 repo 旁邊(inkstone/ISEP#147scripts/worktree 寫的)\n")
fh.write(EXCLUDE_LINE + "\n")
return True
def repos_under(root):
"""root 自己(若是主工作目錄)+它的直接子目錄+ matrix/products/polaris 的子目錄裡的主工作目錄。"""
found, seen = [], set()
def add(p):
if not os.path.isdir(p):
return
rp = os.path.realpath(p)
if rp in seen:
return
seen.add(rp)
if os.path.isdir(os.path.join(rp, ".git")) and is_main_worktree(rp):
found.append(rp)
root = os.path.realpath(root)
add(root)
try:
kids = sorted(os.listdir(root))
except OSError:
kids = []
for k in kids:
if k.startswith(".") and k != WT_DIR:
continue
p = os.path.join(root, k)
add(p)
if k in ("matrix", "products", "polaris") and os.path.isdir(p):
try:
for kk in sorted(os.listdir(p)):
add(os.path.join(p, kk))
except OSError:
pass
return found
def fmt_pushed(v):
return {0: "已推上遠端", 1: "**沒推**", 2: "問不到遠端"}[v]
def describe(r):
br = r["branch"] or ("detached %s" % r["sha"][:7])
if r["verdict"] == "prunable":
return "%s ← 登記指向空氣(目錄已不在)" % r["path"]
bits = [fmt_pushed(r["pushed"])]
if r["dirty"]:
bits.append("**%d 個未 commit/未追蹤檔**" % r["dirty"])
return "%s\n 分支 %s%s%s" % (r["path"], br, r["sha"][:7], "、".join(bits))
def push_hint(r):
rm = (out(r["main"], "remote").split() or ["origin"])[0]
br = r["branch"] or "<分支>"
return "git -C %s push %s %s" % (r["path"], rm, br)
def remove_one(r):
"""只在 verdict==collect 時呼叫。不 force:git 自己拒絕的(有東西沒交代)就照實回報。"""
rc, _, err = git(r["main"], "worktree", "remove", r["path"])
git(r["main"], "worktree", "prune")
return rc == 0, err.strip()
# ── 動詞 ────────────────────────────────────────────────────────────────
def take_opt(argv, name, default=None, many=False):
vals, rest, i = [], [], 0
while i < len(argv):
if argv[i] == name and i + 1 < len(argv):
vals.append(argv[i + 1])
i += 2
continue
rest.append(argv[i])
i += 1
if many:
return vals, rest
return (vals[-1] if vals else default), rest
def cmd_open(argv):
repo_opt, argv = take_opt(argv, "--repo")
branch, argv = take_opt(argv, "-b")
if not argv:
die("用法:worktree open <owner/repo#N> [--repo <路徑>] [-b <分支>]")
owner, repo, num = parse_ticket(argv[0])
main = main_top(repo_opt or os.getcwd())
if not main:
die("%s 不是任何 git repo 的一部分。用 --repo 指到要開分身的那個 repo。" % (repo_opt or os.getcwd()))
name = wt_name(repo, num)
dest = os.path.join(main, WT_DIR, name)
ensure_exclude(main)
for e in list_linked(main):
if e["path"] == os.path.realpath(dest):
print(dest)
sys.stderr.write("已經開過了(分支 %s)。直接 cd 進去做事。\n" % (e["branch"] or "?"))
return 0
os.makedirs(os.path.join(main, WT_DIR), exist_ok=True)
branch = branch or "feat/%s" % num
rc_b, _, _ = git(main, "rev-parse", "--verify", "-q", "refs/heads/" + branch)
args = ["worktree", "add", dest] + ([branch] if rc_b == 0 else ["-b", branch])
rc, o, err = git(main, *args)
if rc != 0:
die("git worktree add 失敗:\n%s" % err.strip(), 1)
print(dest)
sys.stderr.write(
"🌱 分身開在 repo 裡面:%s(分支 %s\n"
" cd %s\n"
" 收工:python3 %s close %s/%s#%s --repo %s\n" % (dest, branch, dest, self_path(), owner, repo, num, main))
return 0
def find_by_ticket(roots, name):
hits = []
for root in roots:
for main in repos_under(root):
for e in list_linked(main):
if os.path.basename(e["path"]) == name:
hits.append((main, e))
# 同一份分身可能從兩個 root 都走得到;去重
uniq, seen = [], set()
for main, e in hits:
if e["path"] in seen:
continue
seen.add(e["path"])
uniq.append((main, e))
return uniq
def cmd_close(argv):
repo_opt, argv = take_opt(argv, "--repo")
scans, argv = take_opt(argv, "--scan", many=True)
quiet = "--quiet" in argv
argv = [a for a in argv if a != "--quiet"]
if not argv:
die("用法:worktree close <owner/repo#N> [--repo <路徑>] [--scan <目錄>]... [--quiet]")
owner, repo, num = parse_ticket(argv[0])
name = wt_name(repo, num)
roots = list(scans)
if repo_opt:
roots.append(repo_opt)
if not roots:
roots = [main_top(os.getcwd()) or os.getcwd()]
hits = find_by_ticket(roots, name)
if not hits:
if not quiet:
sys.stderr.write("沒有找到 %s/%s#%s 的分身(找的是 .worktrees/%s,範圍:%s)。\n" % (owner, repo, num, name, "、".join(roots)))
return 0
bad = []
for main, e in hits:
r = classify(main, e)
if r["verdict"] == "prunable":
git(main, "worktree", "prune")
sys.stderr.write("🧹 %s 的登記指向空氣,已 prune。\n" % r["path"])
continue
if r["verdict"] == "collect":
ok, err = remove_one(r)
if ok:
sys.stderr.write("🧹 收掉 %s(分支 %s 已在遠端,工作樹乾淨)。\n" % (r["path"], r["branch"] or r["sha"][:7]))
continue
r["verdict"] = "dirty"
r["git_said"] = err
bad.append(r)
if not bad:
return 0
lines = ["🌳 %s/%s#%s 的分身**沒有收掉**——不是靜默留著,是這裡點名:" % (owner, repo, num), ""]
for r in bad:
lines.append(" ❌ " + describe(r))
if r["verdict"] == "unpushed":
lines.append(" 沒推就不收(不刪任何沒推的 commit)。推了再收:")
lines.append(" %s" % push_hint(r))
elif r["verdict"] == "dirty":
lines.append(" 有東西沒交代。commit 並推、或 `git -C %s stash`/丟掉,再收:" % r["path"])
if r.get("git_said"):
lines.append(" git 說:%s" % r["git_said"].splitlines()[-1])
elif r["verdict"] == "unknown":
lines.append(" 問不到遠端(沒 remote/網路不通)——**不當成沒推**,也不收。網路回來再收:")
lines.append(" python3 %s close %s/%s#%s --repo %s" % (self_path(), owner, repo, num, r["main"]))
lines.append("")
lines.append(" 規約:%sinkstone/ISEP#147" % GOVERNANCE)
sys.stderr.write("\n".join(lines) + "\n")
return 2
def gather(roots, skip):
"""回 (rows, orphans, skipped_rows)。rows=每份分身的 classifyorphans=看起來是 worktree 但沒人登記它的目錄。"""
rows, skipped, registered = [], [], set()
mains = []
for root in roots:
for m in repos_under(root):
if m not in mains:
mains.append(m)
for main in mains:
linked = list_linked(main)
for e in linked:
registered.add(e["path"])
if os.path.basename(main) in skip:
for e in linked:
r = dict(e)
r["main"] = main
r["verdict"] = "skipped"
r["dirty"] = 0
r["pushed"] = 2
skipped.append(r)
continue
for e in linked:
rows.append(classify(main, e))
orphans = []
for root in roots:
try:
kids = sorted(os.listdir(root))
except OSError:
kids = []
for k in kids:
p = os.path.realpath(os.path.join(root, k))
gitfile = os.path.join(p, ".git")
if not os.path.isfile(gitfile) or p in registered:
continue
try:
gd = open(gitfile, encoding="utf-8", errors="replace").read().strip()
except OSError:
gd = ""
gd = gd[len("gitdir:"):].strip() if gd.startswith("gitdir:") else gd
owner_main = ""
m = re.search(r"^(.*)/\.git/worktrees/[^/]+/?$", gd)
if m:
owner_main = m.group(1)
if owner_main and os.path.basename(owner_main) in skip:
skipped.append({"path": p, "main": owner_main, "verdict": "skipped", "branch": "", "sha": "", "dirty": 0, "pushed": 2})
continue
orphans.append({"path": p, "gitdir": gd, "main": owner_main})
return rows, orphans, skipped
def cmd_list(argv):
roots = argv or [os.getcwd()]
rows, orphans, _ = gather(roots, set())
if not rows and not orphans:
print("沒有任何分身。")
return 0
for r in rows:
print(" %s %s" % ({"collect": "🟢", "unpushed": "🔴", "dirty": "🟡", "unknown": "❔", "prunable": "👻"}[r["verdict"]], describe(r)))
for o in orphans:
print(" ❓ %s ← .git 檔指向 %s,但沒有任何 repo 登記它" % (o["path"], o["gitdir"] or "?"))
return 0
def cmd_sweep(argv):
skip, argv = take_opt(argv, "--skip", many=True)
apply = "--apply" in argv
argv = [a for a in argv if a != "--apply"]
if not argv:
die("用法:worktree sweep <目錄> [--apply] [--skip <repo名>]...")
root = os.path.realpath(argv[0])
if not os.path.isdir(root):
die("%s 不是目錄" % root)
skip = set(skip)
rows, orphans, skipped = gather([root], skip)
groups = {"collect": [], "unpushed": [], "dirty": [], "unknown": [], "prunable": []}
for r in rows:
groups[r["verdict"]].append(r)
print("🔎 %s 底下的分身(%s" % (root, "**--apply:會收**" if apply else "只列不動;加 --apply 才收"))
print()
removed, failed = [], []
if groups["collect"]:
print("🟢 推了且乾淨(%d)——%s" % (len(groups["collect"]), "收掉" if apply else "可以收"))
for r in groups["collect"]:
print(" " + describe(r))
if apply:
ok, err = remove_one(r)
if ok:
removed.append(r)
print(" 🧹 已收")
else:
failed.append(r)
print(" ❌ git 拒絕:%s" % (err.splitlines()[-1] if err else "?"))
print()
if groups["prunable"]:
print("👻 登記指向空氣(%d)——%s" % (len(groups["prunable"]), "prune" if apply else "可以 prune"))
for r in groups["prunable"]:
print(" " + describe(r))
if apply:
git(r["main"], "worktree", "prune")
print(" 🧹 已 prune")
print()
if groups["unpushed"]:
print("🔴 **沒推**%d)——不動,你決定:推了再收、或確認作廢後自己刪" % len(groups["unpushed"]))
for r in groups["unpushed"]:
print(" " + describe(r))
print(" 推:%s" % push_hint(r))
print()
if groups["dirty"]:
print("🟡 有未 commit/未追蹤的東西(%d)——不動" % len(groups["dirty"]))
for r in groups["dirty"]:
print(" " + describe(r))
print()
if groups["unknown"]:
print("❔ 問不到遠端(%d)——不當成沒推、也不收" % len(groups["unknown"]))
for r in groups["unknown"]:
print(" " + describe(r))
print()
if skipped:
print("⏭️ --skip 點名不碰(%d" % len(skipped))
for r in skipped:
print(" %s%s" % (r["path"], os.path.basename(r["main"]) if r["main"] else "?"))
print()
if orphans:
print("❓ 看起來是分身、但沒有任何 repo 登記它(%d)——不動,只能人看" % len(orphans))
for o in orphans:
print(" %s ← .git 檔指向 %s" % (o["path"], o["gitdir"] or "?"))
print()
left = len(groups["unpushed"]) + len(groups["dirty"]) + len(groups["unknown"]) + len(orphans) + len(failed)
if apply:
print("── 收了 %d 份,留 %d 份等你決定 ──" % (len(removed), left + (0 if apply else len(groups["collect"]))))
else:
print("── 可收 %d 份,%d 份要你決定;沒有動任何東西 ──" % (len(groups["collect"]) + len(groups["prunable"]), left))
return 2 if left else 0
def main(argv):
if not argv or argv[0] in ("-h", "--help", "help"):
print(__doc__)
return 0
verb, rest = argv[0], argv[1:]
fn = {"open": cmd_open, "close": cmd_close, "list": cmd_list, "sweep": cmd_sweep}.get(verb)
if not fn:
die("不認識的動詞 %r。有:opencloselistsweep--help 看全文)" % verb)
return fn(rest)
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))