併 origin/main（#134 ticket pick／claim）解衝突：盤點數字在合併後的樹重數——61 支、85 條、54 支腳本（inkstone/ISEP#130）
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DTZ9QtvjY7MNxjfQbexAm7
This commit is contained in:
isep-hand
Executable
+224
@@ -0,0 +1,224 @@
|
||||
#!/bin/bash
|
||||
# 雲端工人抓票規則的測試（inkstone/ISEP#131）
|
||||
#
|
||||
# 票上的驗收第 1 條：「`scripts/ticket pick` 給定主線檔，回一張符合規則的票；
|
||||
# 池子裡塞一張 `Human`、一張有 assignee、一張 backlog，三張都不會被挑到（測試綠）。」
|
||||
#
|
||||
# 這支測四件事：
|
||||
# ① `pickable()` 純函式——規則本體，一格一條，不打網路
|
||||
# ② `pick` 走整條路（把 api() 換成錄音機餵一個假池子）：挑到誰、順序、主線抓盡才抓逾期、
|
||||
# 沒有可抓的票時離開碼 1、沒有主線時離開碼 2 且訊息講得出出路
|
||||
# ③ `claim` 一個動作三件事（指派／s/doing／【身份】留言），以及該擋的擋在**打 API 之前**
|
||||
# ④ 不該擋的（誤攔比漏擋更該修）
|
||||
#
|
||||
# 全程離線：假池子在下面的 POOL；TICKET_HOST 指到連不上的位址，任何一條真的走到網路會當場炸，
|
||||
# 不會靜靜變成假綠。不開票、不改票、不留任何東西在 Gitea 上。
|
||||
cd "$(dirname "$0")/.." || exit 1
|
||||
T=scripts/ticket
|
||||
TMP=$(mktemp -d)
|
||||
trap 'rm -rf "$TMP"' EXIT
|
||||
PASS=0; FAIL=0
|
||||
export TICKET_HOST=http://127.0.0.1:9 GITEA_TOKEN=x
|
||||
export ISEP_COUNTDOWN_STATE_DIR="$TMP/state"; mkdir -p "$ISEP_COUNTDOWN_STATE_DIR"
|
||||
export ISEP_COUNTDOWN_NOW=$(python3 -c 'import datetime as d;print(int(d.datetime(2026,9,7,0,0,tzinfo=d.timezone.utc).timestamp()))')
|
||||
unset ISEP_MAINLINE_FILE ISEP_IDENTITY CLAUDE_PROJECT_DIR
|
||||
|
||||
MAIN="$TMP/mainline.json"
|
||||
cat > "$MAIN" <<'JSON'
|
||||
{"ref":"inkstone/Arcrun#48","owner":"inkstone","repo":"Arcrun","id":48,
|
||||
"title":"主線 M","description":"測試用主線","due_on":"2026-09-12T23:59:59Z",
|
||||
"open_issues":3,"closed_issues":0,"members":[],"set_at":"2026-09-07 07:30"}
|
||||
JSON
|
||||
|
||||
# ── 假池子：三個 repo、三個 open milestone、十一張票 ─────────────────────
|
||||
# 主線 M（未逾期，主線）／逾期 M（2026-08-19 到期）／未來 M（2026-10-01 到期，不是主線）
|
||||
cat > "$TMP/pool.py" <<'PY'
|
||||
import json, os
|
||||
MS = {"main": {"id": 48, "title": "主線 M", "due_on": "2026-09-12T23:59:59Z"},
|
||||
"late": {"id": 49, "title": "逾期 M", "due_on": "2026-08-19T23:59:59Z"},
|
||||
"future": {"id": 50, "title": "未來 M", "due_on": "2026-10-01T23:59:59Z"}}
|
||||
def I(repo, n, labels, ms=None, assignees=(), created="2026-09-01T00:00:00Z", pr=False, title=None):
|
||||
d = {"number": n, "title": title or f"票 {repo}#{n}", "state": "open",
|
||||
"labels": [{"name": l} for l in labels],
|
||||
"assignees": [{"login": a} for a in assignees],
|
||||
"milestone": MS[ms] if ms else None, "created_at": created,
|
||||
"repository": {"full_name": f"inkstone/{repo}"}}
|
||||
if pr: d["pull_request"] = {"merged": False}
|
||||
return d
|
||||
POOL = {
|
||||
"ISEP": [I("ISEP", 1, ["s/todo"], "main"), # ★ 可抓
|
||||
I("ISEP", 2, ["s/todo", "Human"], "main"), # 票上第 1 條：Human
|
||||
I("ISEP", 3, ["s/todo"], "main", assignees=("someone",)), # 票上第 1 條：有 assignee
|
||||
I("ISEP", 4, ["s/backlog"], None), # 票上第 1 條：backlog
|
||||
I("ISEP", 5, ["s/todo"], None), # 沒里程碑
|
||||
I("ISEP", 10, ["s/todo"], "main", pr=True), # PR 不是票
|
||||
I("ISEP", 11, ["s/todo", "human/exec"], "main")], # 執行者是人
|
||||
"Arcrun": [I("Arcrun", 6, ["s/todo", "p/high"], "main", created="2026-08-27T00:00:00Z"), # ★ 主線裡最優先
|
||||
I("Arcrun", 7, ["s/todo"], "late"), # 逾期里程碑：主線抓盡才輪到
|
||||
I("Arcrun", 8, ["s/todo"], "future")], # 不是主線也沒逾期：永遠不抓
|
||||
"mira": [I("mira", 9, ["s/todo", "s/doing"], "main")], # 09-07 實查 mira#6 的形狀：有人在做
|
||||
}
|
||||
MSN = {"ISEP": ["main", "late", "future"], "Arcrun": ["main", "late", "future"], "mira": ["main"]}
|
||||
DROP = set(json.loads(os.environ.get("DROP", "[]"))) # 測「主線抓盡」用：把某幾張從池子拿掉
|
||||
def fake_api(path, payload=None, method=None):
|
||||
rec.append((method or ("POST" if payload is not None else "GET"), path, payload))
|
||||
if path.startswith("/orgs/inkstone/repos"):
|
||||
return [{"name": r} for r in POOL]
|
||||
if path == "/user":
|
||||
return {"login": "claude-code"}
|
||||
if "/milestones?" in path:
|
||||
repo = path.split("/")[3]
|
||||
return [MS[k] for k in MSN.get(repo, [])]
|
||||
if "/issues?" in path:
|
||||
repo = path.split("/")[3]
|
||||
import urllib.parse
|
||||
q = urllib.parse.parse_qs(path.split("?", 1)[1])
|
||||
want = q.get("milestones", [""])[0]
|
||||
return [i for i in POOL[repo] if (i["milestone"] or {}).get("title") == want
|
||||
and f"{repo}#{i['number']}" not in DROP]
|
||||
if path.endswith("/labels?limit=100"):
|
||||
return [{"name": n, "id": k} for k, n in
|
||||
enumerate(["s/todo", "s/doing", "s/review", "p/high", "Human"], 1)]
|
||||
import re
|
||||
m = re.match(r"/repos/inkstone/(\w+)/issues/(\d+)$", path)
|
||||
if m and (method or "GET") == "GET":
|
||||
return next(i for i in POOL[m.group(1)] if i["number"] == int(m.group(2)))
|
||||
return {}
|
||||
rec = []
|
||||
PY
|
||||
|
||||
# probe <python 片段> —— 把 scripts/ticket 當模組載進來、api() 換成假池子，印出片段的值
|
||||
probe() {
|
||||
python3 - "$1" "$2" <<'PY'
|
||||
import sys, os, json, io, contextlib, importlib.util as u
|
||||
from importlib.machinery import SourceFileLoader as L
|
||||
spec = u.spec_from_loader("t", L("t", "scripts/ticket"))
|
||||
m = u.module_from_spec(spec); spec.loader.exec_module(m)
|
||||
ns = {}
|
||||
exec(open(os.environ["POOL_PY"]).read(), ns)
|
||||
m.api = ns["fake_api"]; rec = ns["rec"]
|
||||
argv = json.loads(sys.argv[2]) if sys.argv[2] else None
|
||||
buf = io.StringIO(); rc = None
|
||||
if argv is not None:
|
||||
with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf):
|
||||
try:
|
||||
getattr(m, argv[0])(argv[1:])
|
||||
except SystemExit as e:
|
||||
rc = e.code
|
||||
print(eval(sys.argv[1], {"m": m, "rec": rec, "rc": rc, "out": buf.getvalue(), "I": ns["I"], "json": json}))
|
||||
PY
|
||||
}
|
||||
export POOL_PY="$TMP/pool.py"
|
||||
|
||||
p() { # p <說明> <want> <python 片段> [argv json]
|
||||
got=$(probe "$3" "${4:-}" 2>&1)
|
||||
if [ "$got" = "$2" ]; then PASS=$((PASS+1)); printf ' ✅ '; else FAIL=$((FAIL+1)); printf ' ❌ '; fi
|
||||
printf 'want=%s got=%s %s\n' "$2" "$got" "$1"
|
||||
}
|
||||
|
||||
echo "── ① pickable()：規則本體，一格一條（純函式，不打網路）──"
|
||||
p "★ s/todo＋沒人＋沒 Human＋主線 → 可抓" "True" "m.pickable(I('ISEP',1,['s/todo'],'main'),'主線 M')[0]"
|
||||
p "★ 掛 Human → 不可抓（票上第 1 條）" "False" "m.pickable(I('ISEP',2,['s/todo','Human'],'main'),'主線 M')[0]"
|
||||
p "★ 有 assignee → 不可抓（票上第 1 條）" "False" "m.pickable(I('ISEP',3,['s/todo'],'main',assignees=('x',)),'主線 M')[0]"
|
||||
p "★ s/backlog、沒里程碑 → 不可抓（票上第 1 條）" "False" "m.pickable(I('ISEP',4,['s/backlog'],None),'主線 M')[0]"
|
||||
p "s/todo 但沒里程碑 → 不可抓（不抓沒里程碑的）" "False" "m.pickable(I('ISEP',5,['s/todo'],None),'主線 M')[0]"
|
||||
p "human/exec（執行者是人）→ 不可抓" "False" "m.pickable(I('ISEP',11,['s/todo','human/exec'],'main'),'主線 M')[0]"
|
||||
p "s/todo＋s/doing 同時掛著（mira#6 的形狀）→ 不可抓，有人在做" "False" "m.pickable(I('mira',9,['s/todo','s/doing'],'main'),'主線 M')[0]"
|
||||
p "hub（scope 容器，09-07 實查 ISEP#31–35 的形狀）→ 不可抓" "False" "m.pickable(I('ISEP',12,['s/todo','hub'],'main'),'主線 M')[0]"
|
||||
p "PR 不是票 → 不可抓" "False" "m.pickable(I('ISEP',10,['s/todo'],'main',pr=True),'主線 M')[0]"
|
||||
p "里程碑不是要求的那個 → 不可抓" "False" "m.pickable(I('Arcrun',8,['s/todo'],'future'),'主線 M')[0]"
|
||||
p "里程碑標題只差空白／大小寫 → 仍算同一個（norm 只有 mainline.py 那一份）" "True" "m.pickable(I('ISEP',1,['s/todo'],'main'),' 主線 m ')[0]"
|
||||
p "理由講得出「為什麼不能抓」（不是只回 False）" "True" "'Human' in m.pickable(I('ISEP',2,['s/todo','Human'],'main'),'主線 M')[1]"
|
||||
p "順序：p/high 先於沒標，沒標先於 p/low" "True" \
|
||||
"m.pick_order(I('a',1,['s/todo','p/high'],'main')) < m.pick_order(I('a',2,['s/todo'],'main')) < m.pick_order(I('a',3,['s/todo','p/low'],'main'))"
|
||||
p "順序：同優先序看開票時間，舊的先" "True" \
|
||||
"m.pick_order(I('a',9,['s/todo'],'main',created='2026-08-01T00:00:00Z')) < m.pick_order(I('a',1,['s/todo'],'main',created='2026-09-01T00:00:00Z'))"
|
||||
|
||||
echo
|
||||
echo "── ② pick 走整條路（假池子 11 張，三個 repo）──"
|
||||
ARGS="[\"cmd_pick\",\"--mainline\",\"$MAIN\"]"
|
||||
ALL="[\"cmd_pick\",\"--mainline\",\"$MAIN\",\"--all\",\"--json\"]"
|
||||
p "★ 給定主線檔 → 回一張，而且是主線裡最優先的 Arcrun#6（p/high）" "True" "'inkstone/Arcrun#6' in out and rc is None" "$ARGS"
|
||||
p "只回一張（不是整份清單）" "1" "out.count(' ● ')" "$ARGS"
|
||||
p "★ --all：主線裡可抓的恰好是 Arcrun#6、ISEP#1 兩張——Human／assignee／backlog 三張都不在" \
|
||||
"['inkstone/Arcrun#6', 'inkstone/ISEP#1']" "[r['ref'] for r in json.loads(out) if r['group']=='主線']" "$ALL"
|
||||
p "★ --all：逾期里程碑那張 Arcrun#7 排在主線之後（主線抓盡才輪到）" \
|
||||
"['inkstone/Arcrun#7']" "[r['ref'] for r in json.loads(out) if r['group']!='主線']" "$ALL"
|
||||
p "沒逾期也不是主線的 Arcrun#8 不在清單裡" "False" "'inkstone/Arcrun#8' in out" "$ALL"
|
||||
p "backlog（ISEP#4）／沒里程碑（ISEP#5）不在清單裡" "False" "'ISEP#4' in out or 'ISEP#5' in out" "$ALL"
|
||||
p "沒有任何一條真的走到網路（api 全被錄音機接住）" "True" "all(p.startswith('/') for _,p,_ in rec)" "$ALL"
|
||||
p "pick 只讀不寫（沒有 POST／PATCH／PUT）" "True" "all(mm=='GET' for mm,_,_ in rec)" "$ALL"
|
||||
p "訊息印出下一步（認領那一行用絕對路徑，貼到別的 cwd 也叫得到正本）" "True" "'claim inkstone/Arcrun#6' in out and m.self_path() in out" "$ARGS"
|
||||
|
||||
echo
|
||||
echo " （主線抓盡 → 才輪到逾期里程碑）"
|
||||
export DROP='["Arcrun#6","ISEP#1"]'
|
||||
p "★ 主線兩張都被拿走後 → 回逾期里程碑的 Arcrun#7" "True" "'inkstone/Arcrun#7' in out and rc is None" "$ARGS"
|
||||
export DROP='["Arcrun#6","ISEP#1","Arcrun#7"]'
|
||||
p "★ 主線與逾期都抓盡 → 離開碼 1（真的沒有，不是故障）" "1" "rc" "$ARGS"
|
||||
p "訊息明講「不要去抓 backlog」（出路是收工回報，不是換規則）" "True" "'backlog' in out and '收工' in out" "$ARGS"
|
||||
p "抓盡時**沒有**回 Arcrun#8（未來里程碑）" "False" "'Arcrun#8' in out" "$ARGS"
|
||||
unset DROP
|
||||
|
||||
echo
|
||||
echo " （沒有主線 ≠ 去抓別的）"
|
||||
NOML="[\"cmd_pick\",\"--mainline\",\"$TMP/none.json\"]"
|
||||
p "主線檔讀不到 → 離開碼 2（讀不到 ≠ 沒有）" "2" "rc" "$NOML"
|
||||
p "…而且一通 API 都沒打" "0" "len(rec)" "$NOML"
|
||||
p "沒給 --mainline、狀態檔也沒有、cwd 也沒有 → 離開碼 2，訊息給出兩條出路" "True" \
|
||||
"rc==2 and 'mainline set' in out and 'system-dev/mainline.json' in out" "[\"cmd_pick\"]"
|
||||
|
||||
echo
|
||||
echo "── ③ claim：一個動作三件事 ──"
|
||||
CL="[\"cmd_claim\",\"inkstone/Arcrun#6\",\"--mainline\",\"$MAIN\",\"--name\",\"isep-hand\"]"
|
||||
p "★ 指派給本 token 的使用者（欄位，不是一句話）" "[['claude-code']]" \
|
||||
"[pl['assignees'] for mm,p,pl in rec if mm=='PATCH' and p=='/repos/inkstone/Arcrun/issues/6']" "$CL"
|
||||
p "★ s/todo 換成 s/doing，p/high 留著（不是整組蓋掉）" "True" \
|
||||
"[sorted(pl['labels'])==[2,4] for mm,p,pl in rec if mm=='PUT' and p.endswith('/issues/6/labels')]==[True]" "$CL"
|
||||
p "★ 留言第一行是【身份】isep-hand／inkstone/Arcrun／-" "True" \
|
||||
"[pl['body'].split(chr(10))[0] for mm,p,pl in rec if mm=='POST' and p.endswith('/issues/6/comments')]==['【身份】isep-hand／inkstone/Arcrun／-']" "$CL"
|
||||
p "留言講了做完要走 handback（不推 main、不出貨）" "True" \
|
||||
"any('handback' in pl['body'] and '不推 main' in pl['body'] for mm,p,pl in rec if mm=='POST' and p.endswith('/comments'))" "$CL"
|
||||
p "三件事都做了，離開碼 0" "None" "rc" "$CL"
|
||||
p "--as 指定登入名時用它" "[['cloud-bot']]" \
|
||||
"[pl['assignees'] for mm,p,pl in rec if mm=='PATCH']" "[\"cmd_claim\",\"inkstone/Arcrun#6\",\"--mainline\",\"$MAIN\",\"--as\",\"cloud-bot\"]"
|
||||
p "逾期里程碑的票（Arcrun#7）也認領得了——主線抓盡時輪到它" "None" "rc" \
|
||||
"[\"cmd_claim\",\"inkstone/Arcrun#7\",\"--mainline\",\"$MAIN\"]"
|
||||
p "pick --claim：挑到誰就認領誰（Arcrun#6）" "True" \
|
||||
"any(mm=='PATCH' and p.endswith('/issues/6') for mm,p,pl in rec)" "[\"cmd_pick\",\"--mainline\",\"$MAIN\",\"--claim\",\"--name\",\"isep-hand\"]"
|
||||
|
||||
echo
|
||||
echo " （該擋的：規則在 claim 這一端再驗一次，擋在**寫入之前**）"
|
||||
blocked() { # blocked <說明> <ref> [extra argv...]
|
||||
what=$1; ref=$2; shift 2
|
||||
extra=$(printf '%s\n' "$@" | python3 -c "import sys,json;print(','.join(json.dumps(l.rstrip()) for l in sys.stdin if l.strip()))")
|
||||
argv="[\"cmd_claim\",\"$ref\",\"--mainline\",\"$MAIN\"${extra:+,$extra}]"
|
||||
p "$what → 離開碼 2，且沒有任何寫入" "True" \
|
||||
"rc==2 and not any(mm in ('PATCH','PUT','POST') for mm,_,_ in rec)" "$argv"
|
||||
}
|
||||
blocked "掛 Human 的 ISEP#2" inkstone/ISEP#2
|
||||
blocked "已指派的 ISEP#3" inkstone/ISEP#3
|
||||
blocked "backlog 的 ISEP#4" inkstone/ISEP#4
|
||||
blocked "沒逾期也不是主線的 Arcrun#8" inkstone/Arcrun#8
|
||||
blocked "有人在做的 mira#9（s/doing 也掛著）" inkstone/mira#9
|
||||
p "擋下時訊息講得出理由與出路（pick）" "True" "'Human' in out and 'pick' in out" \
|
||||
"[\"cmd_claim\",\"inkstone/ISEP#2\",\"--mainline\",\"$MAIN\"]"
|
||||
p "身份名不在名單上 → 在打任何 API 之前就擋（一通都沒打）" "True" "rc==2 and len(rec)==0" \
|
||||
"[\"cmd_claim\",\"inkstone/Arcrun#6\",\"--mainline\",\"$MAIN\",\"--name\",\"路人\"]"
|
||||
|
||||
echo
|
||||
echo "── ④ 動詞接得上（不帶假池子，走到網路才炸）──"
|
||||
out=$(python3 "$T" pick --mainline "$MAIN" 2>&1); got=$?
|
||||
if [ "$got" = 2 ] && printf '%s' "$out" | grep -q '拿不到 Gitea'; then PASS=$((PASS+1)); printf ' ✅ '; else FAIL=$((FAIL+1)); printf ' ❌ '; fi
|
||||
printf 'want=2 got=%s pick 連不上 Gitea → 離開碼 2 並說「讀不到 ≠ 沒有」（不是離開碼 1 的假「沒票」）\n' "$got"
|
||||
out=$(python3 "$T" 2>&1)
|
||||
if printf '%s' "$out" | grep -q "ticket pick" && printf '%s' "$out" | grep -q "ticket claim"; then PASS=$((PASS+1)); printf ' ✅ '
|
||||
else FAIL=$((FAIL+1)); printf ' ❌ '; fi
|
||||
printf '不帶參數的說明裡列得出 pick／claim（撈不到的東西等於不存在）\n'
|
||||
out=$(python3 "$T" usage pick 2>&1); got=$?
|
||||
if [ "$got" = 0 ] && printf '%s' "$out" | grep -q -- '--mainline'; then PASS=$((PASS+1)); printf ' ✅ '; else FAIL=$((FAIL+1)); printf ' ❌ '; fi
|
||||
printf 'ticket usage pick 印得出用法\n'
|
||||
|
||||
echo
|
||||
echo "通過 $PASS 條，失敗 $FAIL 條"
|
||||
[ "$FAIL" = 0 ] || exit 1
|
||||
+335
-1
@@ -28,6 +28,12 @@ leo 2026-08-16 三句話，本工具就是它們的機械化：
|
||||
|
||||
ticket usage <動詞> [--example] 用法的唯一真相源——**閘印的那一行從這裡來**
|
||||
|
||||
雲端工人抓票（inkstone/ISEP#131，規則機械可判，不靠 prompt 叮嚀）：
|
||||
|
||||
ticket pick [--mainline <主線檔>] [--all] [--json] 回一張現在可以抓的票（主線抓盡才抓逾期）
|
||||
ticket claim <票> [--as <登入名>] [--name <身份>] 認領＝指派＋s/doing＋【身份】留言，一個動作
|
||||
ticket pick --claim … 上面兩件連著做
|
||||
|
||||
票的寫法：`owner/repo#N`，例：`inkstone/InkStoneCo#44`
|
||||
"""
|
||||
import json
|
||||
@@ -837,6 +843,17 @@ def subtask_example(parent=None, draft=None):
|
||||
# 🔴 `{bin}` 一定要留成佔位符、由 usage_text() 現填 self_path()：
|
||||
# 用法裡寫死 `ticket` 或 `scripts/ticket`，讀的人照著貼就又踩回這張票的坑。
|
||||
USAGE = {
|
||||
"pick":
|
||||
"用法：{bin} pick [--mainline <主線 json 檔>] [--all] [--json] [--claim [--as <登入名>] [--name <身份>]]\n"
|
||||
" 回**一張**現在可以抓的票（主線里程碑抓盡才輪到逾期里程碑；不抓 backlog、不抓沒里程碑的）\n"
|
||||
" --mainline 不給時依序找：$ISEP_MAINLINE_FILE → ~/.claude/isep-countdown/mainline.json"
|
||||
" → $CLAUDE_PROJECT_DIR/system-dev/mainline.json\n"
|
||||
" 離開碼：0 ＝ 有票（印在 stdout）；1 ＝ 真的沒有可抓的票；2 ＝ 讀不到（讀不到 ≠ 沒有）",
|
||||
"claim":
|
||||
"用法：{bin} claim <owner/repo#N> [--as <登入名，預設本 token 的使用者>] [--name <身份名，預設 subagent>]"
|
||||
" [--mainline <主線 json 檔>]\n"
|
||||
" 認領＝指派給自己 ＋ 換成 s/doing ＋ 第一行【身份】的留言，**一個動作**。\n"
|
||||
" 抓票規則在這裡再驗一次：不是 s/todo、已有人、掛 Human、不在主線也不逾期 ⇒ 擋下（離開碼 2）",
|
||||
"subtask":
|
||||
"用法：{bin} subtask <母票 owner/repo#N> --title \"<User Story>\" -F <內文檔>\n"
|
||||
" [--repo <收件 repo，預設跟母票同一個>] [--label <s/xxx，預設 s/todo>]\n"
|
||||
@@ -1193,10 +1210,327 @@ def cmd_loose(argv):
|
||||
return loose
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════
|
||||
# inkstone/ISEP#131 — 雲端工人抓票：規則是機械可判的，不靠 prompt 叮嚀
|
||||
#
|
||||
# 票上的原話：Routine 08-06 起讀 `journeys.md` 本 sprint 段，08-19 到期沒續，
|
||||
# **連續兩週讀到殘骸只能空轉**。09-07 把 `cloud-worker.md` 步驟 1 改成從 Gitea 主線
|
||||
# 里程碑抓票（inkstone/InkStoneCo#118）——但那是一段給人讀的字，讀漏一句就抓錯。
|
||||
# 這一段把它做成一個指令，規則寫在 `pickable()` 一支純函式裡，離線測得動。
|
||||
#
|
||||
# 規則（全部是 Gitea 欄位，沒有一格在猜文字）：
|
||||
# 可抓 s/* 狀態恰好是 {s/todo} ＋ 沒有 assignee ＋ 沒有 Human／human/* ＋ 不是 hub ＋ 有里程碑
|
||||
# 且里程碑 ＝ 主線；主線抓盡才抓「逾期的 open 里程碑」；backlog／沒里程碑一律不抓
|
||||
# 認領 指派 ＋ s/doing ＋【身份】留言，一個動作（`claim`）
|
||||
# 完成 分支 ＋ 證據 ＋ s/review ＋ `handback`——那半本來就有，不重做
|
||||
#
|
||||
# 🔴 為什麼是「s/* 恰好等於 {s/todo}」而不是「有 s/todo」：2026-09-07 實查
|
||||
# `inkstone/mira#6` 同時掛著 s/doing 與 s/todo（exclusive 只擋 UI，不擋 API）。
|
||||
# 只看「有 s/todo」會把一張別人正在做的票抓走。要求的是**狀態在場且唯一**，
|
||||
# 仍然是「要求某個東西在場」那個形狀，不是黑名單。
|
||||
#
|
||||
# 🔴 Human 那一格用的是**標籤**（唯一識別碼），不是票上有沒有 👤 這個字——
|
||||
# cloud-worker.md 寫過「票上沒有 👤」，那是文字判準，leo 2026-08-17 已證偽那條路。
|
||||
# `human/exec`（執行者是人）與 `Human` 同組（labels.yaml「人的介入」），一起算。
|
||||
# ══════════════════════════════════════════════════════════════════════════
|
||||
|
||||
_ML = None
|
||||
|
||||
|
||||
def _mainline_lib():
|
||||
"""`hooks/lib/mainline.py`——「主線」這件事的唯一存放處。norm() 也只有那一份。"""
|
||||
global _ML
|
||||
if _ML is None:
|
||||
import importlib.machinery
|
||||
import importlib.util
|
||||
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "hooks", "lib", "mainline.py")
|
||||
loader = importlib.machinery.SourceFileLoader("isep_mainline_lib", path)
|
||||
spec = importlib.util.spec_from_file_location("isep_mainline_lib", path, loader=loader)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
loader.exec_module(mod)
|
||||
_ML = mod
|
||||
return _ML
|
||||
|
||||
|
||||
def _now():
|
||||
"""現在。測試用 ISEP_COUNTDOWN_NOW（epoch 秒）把時鐘定住——跟 countdown 同一個開關。"""
|
||||
from datetime import datetime, timezone
|
||||
override = os.environ.get("ISEP_COUNTDOWN_NOW", "").strip()
|
||||
if override:
|
||||
try:
|
||||
return datetime.fromtimestamp(float(override), tz=timezone.utc)
|
||||
except Exception:
|
||||
pass
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _label_names(issue):
|
||||
out = []
|
||||
for l in issue.get("labels") or []:
|
||||
out.append(l["name"] if isinstance(l, dict) else str(l))
|
||||
return out
|
||||
|
||||
|
||||
def human_gated(labels):
|
||||
"""掛著「人的介入」那一組標籤（labels.yaml：`Human`／`human/*`）。"""
|
||||
return any(n == "Human" or n.startswith("human/") for n in labels)
|
||||
|
||||
|
||||
def pickable(issue, milestone_title):
|
||||
"""這張票現在能不能被雲端工人抓走。回 (ok, 理由)。**純函式，不打網路。**
|
||||
|
||||
`milestone_title`＝要求它掛在哪個里程碑底下（主線，或輪到的那個逾期里程碑）。
|
||||
理由是給人看的：`pick --all` 與 `claim` 擋下時都印它，一張票為什麼不能抓要說得出來。
|
||||
"""
|
||||
if issue.get("pull_request"):
|
||||
return False, "這是 PR，不是票"
|
||||
labels = _label_names(issue)
|
||||
st = sorted(n for n in labels if n.startswith("s/"))
|
||||
if st != ["s/todo"]:
|
||||
return False, "狀態不是單一的 s/todo（現在是 %s）" % (st or "沒有 s/*")
|
||||
if issue.get("assignees") or issue.get("assignee"):
|
||||
who = [a.get("login") for a in (issue.get("assignees") or []) if isinstance(a, dict)] \
|
||||
or [(issue.get("assignee") or {}).get("login")]
|
||||
return False, "已經指派給 %s" % "、".join(str(w) for w in who if w)
|
||||
if human_gated(labels):
|
||||
return False, "掛著 %s——要 leo 親手做的" % "、".join(n for n in labels if n == "Human" or n.startswith("human/"))
|
||||
if "hub" in labels:
|
||||
# labels.yaml：hub ＝「聚合一批 leaf 的 scope 容器，本身不掛 milestone、不對應 PR」
|
||||
# 2026-09-07 實查：ISEP#31–35 五張 hub 掛在逾期里程碑上、s/todo、沒人——
|
||||
# 照欄位規則會被抓走，但它們不是一件可以做完的事。容器不是任務。
|
||||
return False, "是 hub（scope 容器，不是一件可以做完的事）"
|
||||
ms = (issue.get("milestone") or {}).get("title")
|
||||
if not ms:
|
||||
return False, "沒有里程碑（backlog／沒排進 sprint 的不抓）"
|
||||
norm = _mainline_lib().norm
|
||||
if norm(ms) != norm(milestone_title):
|
||||
return False, "里程碑是「%s」，不是「%s」" % (ms, milestone_title)
|
||||
return True, "ok"
|
||||
|
||||
|
||||
def pick_order(issue):
|
||||
"""同一個里程碑裡先抓誰：p/high → 沒標 → p/low；再看開票時間（舊的先）；最後票號。
|
||||
結果要**可重現**——同一個池子問兩次要拿到同一張。"""
|
||||
labels = _label_names(issue)
|
||||
prio = 0 if "p/high" in labels else (2 if "p/low" in labels else 1)
|
||||
return (prio, issue.get("created_at") or "", issue.get("number") or 0)
|
||||
|
||||
|
||||
def _ref_of(issue, owner=None, repo=None):
|
||||
full = ((issue.get("repository") or {}).get("full_name")) or ("%s/%s" % (owner, repo))
|
||||
return "%s#%s" % (full, issue.get("number"))
|
||||
|
||||
|
||||
def load_mainline(argv):
|
||||
"""主線從哪來（依序）：--mainline <檔> → $ISEP_MAINLINE_FILE → 本機狀態檔
|
||||
（`scripts/mainline set` 寫的那個）→ $CLAUDE_PROJECT_DIR/system-dev/mainline.json（進 repo 的那份，
|
||||
雲端 clone 就讀得到，inkstone/InkStoneCo#118）。回 (dict, 來源說明)。"""
|
||||
def opt(name):
|
||||
return argv[argv.index(name) + 1] if name in argv and argv.index(name) + 1 < len(argv) else None
|
||||
ML = _mainline_lib()
|
||||
tried = []
|
||||
explicit = opt("--mainline") or os.environ.get("ISEP_MAINLINE_FILE", "").strip()
|
||||
if explicit:
|
||||
try:
|
||||
with open(explicit) as f:
|
||||
d = json.load(f)
|
||||
except Exception as e:
|
||||
die("🔴 讀不到主線檔 %s：%s\n 讀不到 ≠ 沒有主線——先把檔案路徑修對，不要因此去抓別的票。" % (explicit, e))
|
||||
if not isinstance(d, dict) or not d.get("title"):
|
||||
die("🔴 主線檔 %s 裡沒有 title 欄——那不是 `scripts/mainline set` 寫出來的檔。" % explicit)
|
||||
return d, explicit
|
||||
tried.append(ML.path())
|
||||
d = ML.load()
|
||||
if d:
|
||||
return d, ML.path()
|
||||
root = os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
|
||||
cand = os.path.join(root, "system-dev", "mainline.json")
|
||||
tried.append(cand)
|
||||
try:
|
||||
with open(cand) as f:
|
||||
d = json.load(f)
|
||||
if isinstance(d, dict) and d.get("title"):
|
||||
return d, cand
|
||||
except Exception:
|
||||
pass
|
||||
die("""🔴 現在沒有主線 ⇒ 沒有「可以抓的票」這回事。
|
||||
|
||||
找過：
|
||||
%s
|
||||
|
||||
這不是叫你去抓 backlog 或沒里程碑的票——那是總管排 sprint 的事，不是工人的。
|
||||
出路（擇一）：
|
||||
• 雲端：clone inkstone/InkStoneCo，用它 repo 裡那份 → `%s pick --mainline <clone>/system-dev/mainline.json`
|
||||
• 本機：`scripts/mainline list` 看有哪些 → `scripts/mainline set <owner/repo#id>`""" % (
|
||||
"\n".join(" · " + t for t in tried), self_path()))
|
||||
|
||||
|
||||
def _repos(owner):
|
||||
return [r["name"] for r in (api("/orgs/%s/repos?limit=100" % owner) or [])
|
||||
if isinstance(r, dict) and r.get("name")]
|
||||
|
||||
|
||||
def _open_milestones(owner, repos):
|
||||
"""每個 repo 的 open milestone。回 [(repo, milestone dict)]。"""
|
||||
out = []
|
||||
for r in repos:
|
||||
for m in (api("/repos/%s/%s/milestones?state=open&limit=100" % (owner, r)) or []):
|
||||
if isinstance(m, dict) and m.get("title"):
|
||||
out.append((r, m))
|
||||
return out
|
||||
|
||||
|
||||
def _issues_in(owner, repo, title):
|
||||
q = urllib.parse.urlencode({"state": "open", "type": "issues", "milestones": title, "limit": 100})
|
||||
rows = api("/repos/%s/%s/issues?%s" % (owner, repo, q)) or []
|
||||
for it in rows:
|
||||
if isinstance(it, dict):
|
||||
it.setdefault("repository", {"full_name": "%s/%s" % (owner, repo)})
|
||||
return [it for it in rows if isinstance(it, dict)]
|
||||
|
||||
|
||||
def pick_candidates(ms, now=None):
|
||||
"""依序回 [(組名, 里程碑標題, [可抓的票…])]：第一組是主線，之後是逾期的 open 里程碑
|
||||
（最逾期的先）。**每一組裡都已經照 pick_order 排好。**"""
|
||||
ML = _mainline_lib()
|
||||
now = now or _now()
|
||||
owner = ms.get("owner") or ORG
|
||||
repos = _repos(owner)
|
||||
opened = _open_milestones(owner, repos)
|
||||
groups = []
|
||||
|
||||
def collect(title):
|
||||
found = []
|
||||
for r, m in opened:
|
||||
if ML.norm(m.get("title")) != ML.norm(title):
|
||||
continue
|
||||
for it in _issues_in(owner, r, m.get("title")):
|
||||
ok, _ = pickable(it, title)
|
||||
if ok:
|
||||
found.append(it)
|
||||
return sorted(found, key=pick_order)
|
||||
|
||||
groups.append(("主線", ms["title"], collect(ms["title"])))
|
||||
|
||||
overdue = {}
|
||||
for r, m in opened:
|
||||
if ML.norm(m.get("title")) == ML.norm(ms["title"]):
|
||||
continue
|
||||
due = ML.due_of(m)
|
||||
if due and due < now:
|
||||
key = ML.norm(m.get("title"))
|
||||
if key not in overdue or due < overdue[key][0]:
|
||||
overdue[key] = (due, m.get("title"))
|
||||
for due, title in sorted(overdue.values()):
|
||||
groups.append(("逾期里程碑", title, collect(title)))
|
||||
return groups
|
||||
|
||||
|
||||
def cmd_pick(argv):
|
||||
"""回一張現在可以抓的票。離開碼：0 有、1 真的沒有、2 讀不到／沒主線。"""
|
||||
if argv and argv[0] in ("-h", "--help"):
|
||||
die(usage_text("pick"), 0)
|
||||
ms, src = load_mainline(argv)
|
||||
as_json = "--json" in argv
|
||||
show_all = "--all" in argv
|
||||
try:
|
||||
groups = pick_candidates(ms)
|
||||
except (urllib.error.URLError, OSError) as e:
|
||||
die("🔴 拿不到 Gitea（%s）。讀不到 ≠ 沒有可抓的票——先修連線，不要空轉也不要抓別的。" % e)
|
||||
|
||||
flat = [(g, t, it) for g, t, its in groups for it in its]
|
||||
if as_json:
|
||||
rows = [{"ref": _ref_of(it), "title": it.get("title"), "group": g, "milestone": t,
|
||||
"labels": _label_names(it), "created_at": it.get("created_at")}
|
||||
for g, t, it in flat]
|
||||
print(json.dumps(rows if show_all else (rows[0] if rows else None), ensure_ascii=False, indent=2))
|
||||
sys.exit(0 if rows else 1)
|
||||
|
||||
print("🎯 主線：「%s」（來源：%s）" % (ms["title"], src))
|
||||
for g, t, its in groups:
|
||||
if g == "主線" or its or show_all:
|
||||
print(" %s「%s」：可抓 %d 張" % (g, t, len(its)))
|
||||
print()
|
||||
if not flat:
|
||||
print("⚪ 沒有可以抓的票——主線與逾期里程碑都抓盡了。")
|
||||
print(" 出路：收工回報（這是正常結果，不是故障）。")
|
||||
print(" 🔴 不要去抓 backlog／沒里程碑／掛 Human／已有人的票——那些不在規則裡，抓了就是跳線。")
|
||||
sys.exit(1)
|
||||
|
||||
shown = flat if show_all else flat[:1]
|
||||
for g, t, it in shown:
|
||||
print(" ● %s %s" % (_ref_of(it), _label_names(it)))
|
||||
print(" %s" % (it.get("title") or "")[:80])
|
||||
print(" %s「%s」 開票 %s" % (g, t, (it.get("created_at") or "")[:10]))
|
||||
print()
|
||||
if "--claim" in argv:
|
||||
top = flat[0][2]
|
||||
cmd_claim([_ref_of(top)] + [a for a in argv if a != "--claim"])
|
||||
return
|
||||
print("📌 認領（指派＋s/doing＋【身份】留言，一個動作）：")
|
||||
print(" %s claim %s --name <你的名字>" % (self_path(), _ref_of(flat[0][2])))
|
||||
print(" 做完：分支＋證據 → `%s handback %s --to %s --label s/review --next \"…\" --evidence <URL>`"
|
||||
% (self_path(), _ref_of(flat[0][2]), HANDBACK_TO))
|
||||
|
||||
|
||||
def cmd_claim(argv):
|
||||
"""認領＝指派 ＋ s/doing ＋【身份】留言，**一個動作**。抓票規則在這裡再驗一次。"""
|
||||
if not argv or argv[0] in ("-h", "--help"):
|
||||
die(usage_text("claim"), 0 if argv else 2)
|
||||
owner, repo, num = parse_ref(argv[0])
|
||||
|
||||
def opt(name, default=None):
|
||||
return argv[argv.index(name) + 1] if name in argv and argv.index(name) + 1 < len(argv) else default
|
||||
|
||||
name = opt("--name") or "subagent"
|
||||
ident = "【身份】%s／%s/%s／%s" % (name, owner, repo, opt("--branch") or "-")
|
||||
# 身份先驗，**在打任何 API 之前**——名單以外的名字連 Gitea 都不會碰到
|
||||
check_identity(ident, "認領留言")
|
||||
|
||||
ms, _ = load_mainline(argv)
|
||||
ML = _mainline_lib()
|
||||
who = opt("--as") or (api("/user") or {}).get("login")
|
||||
if not who:
|
||||
die("🔴 拿不到本 token 的使用者名稱，也沒給 --as。")
|
||||
|
||||
issue = api("/repos/%s/%s/issues/%d" % (owner, repo, num))
|
||||
issue.setdefault("repository", {"full_name": "%s/%s" % (owner, repo)})
|
||||
ims = issue.get("milestone") or {}
|
||||
ok, why = pickable(issue, ims.get("title") or "")
|
||||
if ok:
|
||||
on_mainline = ML.norm(ims.get("title")) == ML.norm(ms["title"])
|
||||
due = ML.due_of(ims)
|
||||
if not on_mainline and not (due and due < _now()):
|
||||
ok, why = False, "里程碑「%s」既不是主線「%s」也還沒逾期" % (ims.get("title"), ms["title"])
|
||||
if not ok:
|
||||
die("""🚫 %s/%s#%d 現在不能抓：%s
|
||||
|
||||
抓票規則（inkstone/ISEP#131，全部是 Gitea 欄位）：
|
||||
s/* 恰好是 s/todo ＋ 沒有 assignee ＋ 沒有 Human／human/* ＋ 里程碑是主線（主線抓盡才抓逾期）
|
||||
出路：`%s pick` 會回一張符合規則的票；這張如果真的該你做，回票上問總管改欄位，不要硬抓。""" % (
|
||||
owner, repo, num, why, self_path()))
|
||||
|
||||
api("/repos/%s/%s/issues/%d" % (owner, repo, num), {"assignees": [who]}, method="PATCH")
|
||||
keep = [n for n in _label_names(issue) if not n.startswith("s/")] + ["s/doing"]
|
||||
_set_labels(owner, repo, num, keep)
|
||||
lines = [ident, "",
|
||||
"🙋 **認領** → `%s`" % who, "",
|
||||
"- 里程碑：「%s」%s" % (ims.get("title"), "（主線）" if ML.norm(ims.get("title")) == ML.norm(ms["title"]) else "（逾期里程碑，主線已抓盡）"),
|
||||
"- 狀態：s/todo → s/doing",
|
||||
"- 做完：分支＋證據 → `s/review` ＋ `ticket handback`（不推 main、不出貨、不改里程碑）"]
|
||||
api("/repos/%s/%s/issues/%d/comments" % (owner, repo, num), {"body": "\n".join(lines)})
|
||||
print("✅ 認領 %s/%s#%d → `%s`" % (owner, repo, num, who))
|
||||
print(" 指派：%s tag：%s" % (who, [n for n in keep if n.startswith("s/")]))
|
||||
print(" 留言第一行：%s" % ident)
|
||||
print("\n📌 做完：%s handback %s/%s#%d --to %s --label s/review --next \"…\" --evidence <URL>"
|
||||
% (self_path(), owner, repo, num, HANDBACK_TO))
|
||||
|
||||
|
||||
|
||||
CMDS = {"where": cmd_where, "say": cmd_say, "new": cmd_new, "close": cmd_close,
|
||||
"decide": cmd_decide, "subtask": cmd_subtask, "handoff": cmd_subtask,
|
||||
"handback": cmd_handback, "mine": cmd_mine, "loose": cmd_loose,
|
||||
"usage": cmd_usage}
|
||||
"usage": cmd_usage, "pick": cmd_pick, "claim": cmd_claim}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2 or sys.argv[1] not in CMDS:
|
||||
|
||||
Reference in New Issue
Block a user