雲端工人會分診、會組里程碑,組了就照里程碑做(inkstone/ISEP#159)

leo 09-18:「它看到票時要能夠組成 milestone 提出版本,所以不只是看票,還要組票。」
雲端只會撿(ISEP#130),沒有任何機制分診或組里程碑;open 302 張裡 74 張沒 s/*、257 張沒掛里程碑。

- ticket triage:沒 s/* 的票 → s/backlog/s/triage/s/pending+票上一則說明(--why 必填),一個動作;
  --dup-of:說明+close/duplicate+關+在留下的那張留一則指回來(留下的那張若是 leo 的就不寫)
- ticket compose:把已分診的票組成 InkStoneCo 里程碑(開工規約 1–7 條):既有載體+前置相依、
  s/backlog → s/todo、名字不准是版本號、期限 1–60 天、不開新票;先全部讀過才寫,理由一次列齊
- 不碰:Human/human/*、指派給別人(含 Leo)——判準是標籤與 assignee 登入名
- pick --pool:還沒到期的 InkStoneCo 里程碑的前置排在散票之前;可做的都抓盡時回待分診的票,
  每列帶 next;queue-drain-guard 照 next 印(待分診是 triage,不是 claim)
- 全局票池只有一個讀法 pool_issues(),接口留給 ISEP#154(global_index 寫卡驗完後只換這一個函式)
- InkStoneCo 掛在 open 里程碑的票=載體(容器):不進票池、不算待分診

假設(寫在這裡等總管 review):
- 分診只收「沒有 s/*」的票;s/triage 仍是總管/leo 的判斷,工人只能把拿不準的丟進 s/triage
- 載體撈不到時,工人用 ticket where+ticket new 補一張 InkStoneCo User Story 當載體(開工規約第 2 條的「真缺口」)
- 期限上限 60 天(9999 與過去日期都擋)

實測:test-ticket-triage 80/80、test-ticket-pick 84/84、queue-drain-guard 28/28;
真 Gitea 代跑一趟:分診 4 張、建 InkStoneCo milestone 68(載體 InkStoneCo#136,前置 ISEP#47、arcrun-rag#199),
pick --pool 第一組即為它的前置;一次違規三格的 compose 真的零寫入擋下。
待總管定版。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-18 18:43:07 +08:00
parent 51c668f9c8
commit 2bbcda2ac3
8 changed files with 900 additions and 41 deletions
+489 -31
View File
@@ -35,6 +35,14 @@ leo 2026-08-16 三句話,本工具就是它們的機械化:
ticket claim <票> [--as <登入名>] [--name <身份>] 認領=指派+s/doing+【身份】留言,一個動作
ticket pick --claim … 上面兩件連著做
雲端工人分診+組里程碑(inkstone/ISEP#159:不只撿票,還要組票):
ticket triage --list [--all] [--json] 還沒有 s/* 的票(--all 連已分診的一起列,找重複用)
ticket triage <票> --to <s/backlog|s/triage|s/pending> --why <一句> 給狀態+票上一則說明,一個動作
ticket triage <票> --dup-of <留下的票> --why <一句> 重複:說明+close/duplicate+關+指回去
ticket compose --title … --goal … --due YYYY-MM-DD --carrier inkstone/InkStoneCo#N <前置…>
組一個 InkStoneCo 里程碑(先全部驗過才寫,不開新票)
票的寫法:`owner/repo#N`,例:`inkstone/InkStoneCo#44`
"""
import json
@@ -909,6 +917,22 @@ USAGE = {
" 認領=指派給自己 + 換成 s/doing + 第一行【身份】的留言,**一個動作**。\n"
" 抓票規則在這裡再驗一次:不是 s/todo、已有人、掛 Human、不在主線也不逾期 ⇒ 擋下(離開碼 2);\n"
" 帶 --pool(或 ISEP_DRAIN=1)時 open 票池裡的 s/todos/backlog 也認領得了",
"triage":
"用法:{bin} triage --list [--json] [--all]\n"
" {bin} triage <owner/repo#N> --to <s/backlog|s/triage|s/pending> --why \"<一句理由>\" [--name <身份>] [--sign <署名>]\n"
" {bin} triage <owner/repo#N> --dup-of <owner/repo#M> --why \"<一句理由>\" [--name <身份>] [--sign <署名>]\n"
" 分診**還沒有 s/* 的票**inkstone/ISEP#159):給它一個狀態+在票上留一則說明,一個動作;\n"
" 重複的關掉並指向留下的那張(close/duplicate,留下的那張也留一則指回來)。\n"
" --list 列出現在待分診的票;--all 連已分診的 open 票一起列(找重複用)。\n"
" 掛 Humanhuman/*、指派給別人(含 Leo)的票一律不碰;s/todo 要經 compose 排進里程碑才給。",
"compose":
"用法:{bin} compose --title \"<這一輪要讓什麼變成真的>\" --goal \"<一段話>\" --due YYYY-MM-DD\n"
" --carrier inkstone/InkStoneCo#N <前置票 owner/repo#M ...> [--name <身份>] [--sign <署名,例 cloud-worker>]\n"
" 把已分診的票組成一個里程碑(inkstone/ISEP#159CLAUDE.md 開工規約 17 條):\n"
" 里程碑建在 inkstone/InkStoneCo、掛在既有的載體票上;前置票掛成載體票的相依,s/backlog → s/todo\n"
" deliverable=一個總管在 stage 驗過的新版本(版本號由總管定)。**不開任何新票。**\n"
" 先全部驗過才寫:名字像版本號、期限不在 1–60 天內、同名里程碑已在、載體或前置掛 Human/指派別人、\n"
" 前置還沒分診或已在別的 open 里程碑裡 ⇒ 一次列出全部理由,一筆都不寫(離開碼 2)。",
"subtask":
"用法:{bin} subtask <母票 owner/repo#N> --title \"<User Story>\" -F <內文檔>\n"
" [--repo <收件 repo,預設跟母票同一個>] [--label <s/xxx,預設 s/todo>]\n"
@@ -1507,7 +1531,19 @@ def _open_issues(owner, repo):
return out
def pool_candidates(owner, me, seen=()):
def pool_issues(owner):
"""整個 org 的 open 票(不含 PR)——**全局票池的唯一讀法**,pick 的票池層與 triage 共用這一份。
🔌 接口留給 inkstone/ISEP#154:全局票池是 Arcrun 工作流 `global_index`tickets 庫),
寫卡驗完之前先用這裡直打 Gitea。要換成讀 #154 那份時**只換這一個函式**——
呼叫端拿到的形狀不變(Gitea issue dict,帶 `repository.full_name`),不另寫第三套撈法。"""
out = []
for r in _repos(owner):
out += [it for it in _open_issues(owner, r) if not it.get("pull_request")]
return out
def pool_candidates(owner, me, seen=(), issues=None):
"""第三層:整個 org 的 open 票池(inkstone/ISEP#130 c7801)。回 (可抓的票…, 池子大小, {不能抓的理由: 張數})。
leo 09-18:「我看到 Gitea 有 446 張票,要它自己領工作就是持續消化工作,它說沒事做?」
@@ -1518,19 +1554,23 @@ def pool_candidates(owner, me, seen=()):
`seen`=前兩層已經列過的票號,不重複。理由表一併回傳:**可抓 0 張時要說得出池子去哪了**。"""
found, why_count, total = [], {}, 0
for r in _repos(owner):
for it in _open_issues(owner, r):
if it.get("pull_request"):
continue
total += 1
ref = _ref_of(it)
if ref in seen:
continue
ok, why = pickable(it, None, me, POOL_STATUSES)
if ok:
found.append(it)
else:
why_count[why] = why_count.get(why, 0) + 1
for it in (pool_issues(owner) if issues is None else issues):
if it.get("pull_request"):
continue
total += 1
ref = _ref_of(it)
if ref in seen:
continue
if _is_carrier(it):
# 開工規約第 7 條:InkStoneCo 里程碑裡的票是載體(容器),做的是它的前置(inkstone/ISEP#159
why = "是里程碑「%s」的載體票(容器,做的是它的前置)" % it["milestone"].get("title")
why_count[why] = why_count.get(why, 0) + 1
continue
ok, why = pickable(it, None, me, POOL_STATUSES)
if ok:
found.append(it)
else:
why_count[why] = why_count.get(why, 0) + 1
return sorted(found, key=pick_order), total, why_count
@@ -1546,7 +1586,39 @@ def mainline_members(ms):
return _mainline_lib().collect_members(api, ms)
def pick_candidates(ms, now=None):
# 里程碑一律住在這個 repo(CLAUDE.md 開工規約第 7 條,leo 2026-09-13):
# 裡面只放載體票,別 repo 的執行票是載體票的**前置**/dependencies)。
CARRIER_REPO = "InkStoneCo"
def carrier_members(owner, m):
"""InkStoneCo 里程碑的**執行票**=載體票沿相依收下去的票,不含載體票本身(載體是容器不是任務)。
判準同主線(`hooks/lib/mainline.py::collect_members`inkstone/ISEP#133),不另寫一份。"""
refs, info = _mainline_lib().collect_members(
api, {"owner": owner, "repo": CARRIER_REPO, "id": m.get("id"), "title": m.get("title")})
roots = set(info.get("milestone") or [])
return [r for r in refs if r not in roots]
def _milestone_group(owner, r, m, me):
"""一個里程碑底下現在可抓的票。InkStoneCo 的里程碑看相依邊;別的 repo 照舊看票上的里程碑欄。"""
found = []
if r == CARRIER_REPO:
for ref in carrier_members(owner, m):
it = _fetch_issue(ref)
if it and (it.get("state") or "open") == "open" and pickable(it, None, me)[0]:
found.append(it)
return sorted(found, key=pick_order)
for it in _issues_in(owner, r, m.get("title")):
if str((it.get("milestone") or {}).get("id")) != str(m.get("id")):
continue
ok, _ = pickable(it, m.get("title"), me)
if ok:
found.append(it)
return sorted(found, key=pick_order)
def pick_candidates(ms, now=None, pool=False):
"""依序回 [(組名, 里程碑標題, [可抓的票…])]:第一組是主線,之後是逾期的 open 里程碑
(最逾期的先)。**每一組裡都已經照 pick_order 排好。**
@@ -1572,23 +1644,26 @@ def pick_candidates(ms, now=None):
groups = [("主線", ms["title"], sorted(found, key=pick_order))]
repos = _repos(owner)
overdue = []
overdue, upcoming = [], []
for r, m in _open_milestones(owner, repos):
if "%s/%s#%s" % (owner, r, m.get("id")) == anchor:
continue
due = ML.due_of(m)
if due and due < now:
overdue.append((due, r, m))
elif r == CARRIER_REPO:
upcoming.append((due, r, m))
for due, r, m in sorted(overdue, key=lambda x: x[0]):
found = []
for it in _issues_in(owner, r, m.get("title")):
if str((it.get("milestone") or {}).get("id")) != str(m.get("id")):
continue
ok, _ = pickable(it, m.get("title"), me)
if ok:
found.append(it)
groups.append(("逾期里程碑", "%s%s/%s#%s" % (m.get("title"), owner, r, m.get("id")),
sorted(found, key=pick_order)))
_milestone_group(owner, r, m, me)))
if pool:
# 第三層(inkstone/ISEP#159):還沒到期的 InkStoneCo 里程碑——雲端工人自己組的里程碑就在這裡,
# 「組了之後照里程碑往下做」靠這一層:它的前置票先於票池裡散的票。期限近的先。
# 只在接票的環境(--poolISEP_DRAIN=1)看:本機總管收工時不會因此被點名。
far = now.replace(year=now.year + 100)
for due, r, m in sorted(upcoming, key=lambda x: x[0] or far):
groups.append(("里程碑", "%s%s/%s#%s" % (m.get("title"), owner, r, m.get("id")),
_milestone_group(owner, r, m, me)))
return groups
@@ -1602,19 +1677,26 @@ def cmd_pick(argv):
pool = pool_mode(argv)
pool_total, pool_why = None, {}
try:
groups = pick_candidates(ms)
# 第三層只在前兩層抓盡(或要看全部)時才讀——它要翻整個 org,能省就省
groups = pick_candidates(ms, pool=pool)
# 票池那一層只在前面都抓盡(或要看全部)時才讀——它要翻整個 org,能省就省
if pool and (show_all or not any(its for _, _, its in groups)):
seen = {_ref_of(it) for _, _, its in groups for it in its}
found, pool_total, pool_why = pool_candidates(ms.get("owner") or ORG, _me(), seen)
me = _me()
issues = pool_issues(ms.get("owner") or ORG)
found, pool_total, pool_why = pool_candidates(ms.get("owner") or ORG, me, seen, issues)
groups.append(("票池", "open 票池(%s,不限里程碑)" % "".join(POOL_STATUSES), found))
# 最後一層(inkstone/ISEP#159):沒有可做的票時,還沒驗傷的票就是下一件事——
# 回一張待分診的票,下一步是 `triage` 不是 `claim`(佇列閘照 row 的 next 印)。
if show_all or not found:
groups.append(("待分診", "還沒有 s/* 的 open 票", triage_candidates(issues, me)))
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")}
"labels": _label_names(it), "created_at": it.get("created_at"),
"next": _next_cmd(g, _ref_of(it))}
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))
if not rows:
@@ -1623,8 +1705,8 @@ def cmd_pick(argv):
print("🎯 主線:「%s」(來源:%s" % (ms["title"], src))
for g, t, its in groups:
if g in ("主線", "票池") or its or show_all:
print(" %s「%s」:可抓 %d 張" % (g, t, len(its)))
if g in ("主線", "票池", "待分診") or its or show_all:
print(" %s「%s」:%s %d 張" % (g, t, "待分診" if g == "待分診" else "可抓", len(its)))
print()
if not flat:
_explain_empty(pool, pool_total, pool_why, sys.stdout)
@@ -1636,6 +1718,12 @@ def cmd_pick(argv):
print(" %s" % (it.get("title") or "")[:80])
print(" %s「%s」 開票 %s" % (g, t, (it.get("created_at") or "")[:10]))
print()
if flat[0][0] == "待分診":
# 分診不是認領:沒有人要「做」這張票,要的是決定它去哪
print("📌 可做的票都抓盡了,下一件事是分診(還沒驗傷的票不會自己消失):")
print(" " + _next_cmd("待分診", _ref_of(flat[0][2])))
print(" 找重複:%s triage --list --all  組里程碑:%s usage compose" % (self_path(), self_path()))
return
if "--claim" in argv:
top = flat[0][2]
cmd_claim([_ref_of(top)] + [a for a in argv if a != "--claim"])
@@ -1647,6 +1735,14 @@ def cmd_pick(argv):
% (self_path(), _ref_of(flat[0][2]), HANDBACK_TO))
def _next_cmd(group, ref):
"""這一列的下一步,**貼了就能跑的那一行**(佇列閘照這一格印,不自己猜)。"""
if group == "待分診":
return ("%s triage %s --to <s/backlog|s/triage|s/pending> --why \"<一句理由>\""
" (重複的:--dup-of <owner/repo#M>" % (self_path(), ref))
return "%s claim %s --name <你的名字>" % (self_path(), ref)
def _explain_empty(pool, total, why, out):
"""可抓 0 張時講清楚池子去哪了。**不准說成「正常、不是故障」**inkstone/ISEP#130 c7797
09-18 六趟 run 都把「可抓 0 張」寫成正常收工,而池子裡有幾百張 open 票)。"""
@@ -1729,10 +1825,372 @@ def cmd_claim(argv):
# ══════════════════════════════════════════════════════════════════════════
# inkstone/ISEP#159 — 分診+組里程碑:雲端工人不只「撿」,還要「組」
#
# leo 2026-09-18:「它看到票時要能夠組成 milestone 提出版本,所以不只是看票,還要組票。」
# 「我需要幫忙的 AI,而不是我要一直服務它的 AI。」
# 09-18 實量 inkstone orgopen 302 張,沒有 s/* 的 74 張、沒掛里程碑的 257 張。
#
# 哪張跟哪張重複、哪幾張該組成一輪——那是**判斷**,留給讀票的人(雲端工人);
# 這裡只管判斷寫回 Gitea 的那一步,讓它**一個動作、有說明、碰不到不該碰的**:
# · 判準全是欄位與唯一識別碼(s/* 標籤、assignee 登入名、票號、里程碑 id),沒有一格在猜文字
# · 不該碰的=掛 Human/human/*、指派給別人(含 Leo)——讀可以,一個字都不寫
# · compose 先全部驗過才寫第一筆;驗不過一次列出全部理由(改一格撞一次=閘在懲罰人)
# ══════════════════════════════════════════════════════════════════════════
TRIAGE_TO = ("s/backlog", "s/triage", "s/pending")
COMPOSE_MEMBER_STATUSES = ("s/backlog", "s/todo", "s/doing", "s/review", "s/stage", "s/pending")
COMPOSE_MAX_DAYS = 60
def hands_off(issue, me):
"""這張票能不能寫(留言/改標籤/關)。回 None=可以,否則回理由。**純函式。**"""
if not isinstance(issue, dict) or not issue.get("number"):
return "讀不到這張票"
if issue.get("pull_request"):
return "這是 PR,不是票"
if (issue.get("state") or "open") != "open":
return "已經關了"
labels = _label_names(issue)
if human_gated(labels):
return "掛著 %s——要 leo 親手做的,不碰" % "、".join(
n for n in labels if n == "Human" or n.startswith("human/"))
who = [a.get("login") for a in (issue.get("assignees") or []) if isinstance(a, dict)]
if not who and isinstance(issue.get("assignee"), dict):
who = [issue["assignee"].get("login")]
others = [w for w in who if w and w != me]
if others or (who and not me):
return "指派給 %s——棒子在別人手上,不碰" % "、".join(str(w) for w in who if w)
return None
def _status_of(issue):
return sorted(n for n in _label_names(issue) if n.startswith("s/"))
def triageable(issue, me):
"""待分診=還沒有任何 s/*,而且碰得了。回 (ok, 理由)。**純函式。**"""
why = hands_off(issue, me)
if why:
return False, why
st = _status_of(issue)
if st:
return False, "已經分診過(%s" % "、".join(st)
if _is_carrier(issue):
return False, "是里程碑「%s」的載體票(容器,狀態跟著它的前置走)" % issue["milestone"].get("title")
return True, "ok"
def _is_carrier(issue):
"""InkStoneCo 裡掛在 open 里程碑上的票=載體(開工規約第 7 條)。只看欄位:repo 與里程碑狀態。"""
full = ((issue.get("repository") or {}).get("full_name")) or ""
ims = issue.get("milestone") or {}
return full == "%s/%s" % (ORG, CARRIER_REPO) and bool(ims) and (ims.get("state") or "open") == "open"
def triage_candidates(issues, me):
"""票池裡現在待分診的票,舊的先(放最久的最該有人看一眼)。"""
found = [it for it in issues if triageable(it, me)[0]]
return sorted(found, key=lambda it: (it.get("created_at") or "", _ref_of(it)))
def _opt(argv, name, default=None):
return argv[argv.index(name) + 1] if name in argv and argv.index(name) + 1 < len(argv) else default
def _positional(argv, flags_with_value):
out, skip = [], False
for a in argv:
if skip:
skip = False
continue
if a in flags_with_value:
skip = True
continue
if a.startswith("--"):
continue
out.append(a)
return out
def _fetch(owner, repo, num):
it = api("/repos/%s/%s/issues/%d" % (owner, repo, num))
if isinstance(it, dict):
it.setdefault("repository", {"full_name": "%s/%s" % (owner, repo)})
return it
def _sign(argv):
"""`--sign cloud-worker` ⇒ 留言內文署名 `[cloud-worker]`。身份欄(--name)只收名單上的名字,
雲端工人不在名單上(cloud-worker.md 步驟 1),署名就寫在內文——誰跑的就署誰,不寫死。"""
v = (_opt(argv, "--sign") or "").strip()
return "[%s]" % v if v else ""
def _ident_line(name, owner, repo):
return "【身份】%s%s/%s-" % (name, owner, repo)
# ── triage ───────────────────────────────────────────────────────────────
def cmd_triage(argv):
"""分診一張還沒有 s/* 的票:給狀態+留說明(或當重複關掉並指向留下的那張),一個動作。"""
if not argv or argv[0] in ("-h", "--help"):
die(usage_text("triage"), 0 if argv else 2)
if argv[0] == "--list":
return _triage_list(argv[1:])
ref = argv[0]
owner, repo, num = parse_ref(ref)
to, dup, why = _opt(argv, "--to"), _opt(argv, "--dup-of"), (_opt(argv, "--why") or "").strip()
name = _opt(argv, "--name") or "subagent"
# ── 離線就能判的先判,一通 API 都不打 ──
problems = []
if bool(to) == bool(dup):
problems.append("--to 與 --dup-of 恰好給一個(分到哪個狀態,或跟哪張重複)")
if to and to not in TRIAGE_TO:
extra = ("——s/todo 的意思是「已排進里程碑」,走 `ticket compose` 排進去才給" if to == "s/todo" else "")
problems.append("--to 只收 %s,你給的是 %s%s" % ("".join(TRIAGE_TO), to, extra))
if not why:
problems.append("--why 要寫一句理由——票上那則說明就是它(驗收:每張被動過的票上都說得出分到哪、為什麼)")
dparts = None
if dup:
dparts = parse_ref(dup)
if dparts == (owner, repo, num):
problems.append("--dup-of 指到自己")
if problems:
die("🚫 triage %s 還沒送出,參數要補:\n%s\n\n%s" % (
ref, "\n".join(" · " + p for p in problems), usage_text("triage")))
ident = _ident_line(name, owner, repo)
check_identity(ident, "分診留言")
me = _me()
issue = _fetch(owner, repo, num)
ok, reason = triageable(issue, me)
if not ok:
die("🚫 %s 不分診:%s\n 出路:`%s triage --list` 列得出現在待分診的票;"
"已經有 s/* 的票要改狀態,走 `ticket handback``ticket decide`。" % (ref, reason, self_path()))
before_labels = _label_names(issue)
new_status = "close/duplicate" if dup else to
ids = _labels_of(owner, repo)
if new_status not in ids:
die("🚫 %s/%s 沒有標籤 %s(標籤的唯一真相源是 ISEP labels.yaml,改那裡再跑 gitea-labels-sync"
% (owner, repo, new_status))
target, target_note = None, None
if dup:
target = _fetch(*dparts)
if not isinstance(target, dict) or not target.get("number"):
die("🚫 --dup-of %s 讀不到" % dup)
if target.get("pull_request"):
die("🚫 --dup-of %s 是 PR,不是票" % dup)
if (target.get("state") or "open") != "open":
die("🚫 --dup-of %s 已經關了——跟一張關掉的票「重複」不代表這件事做完了。\n"
" 出路:這張給 --to s/triage 並在 --why 寫「疑似與 %s 重複、那張已關」,交總管看。" % (dup, dup))
target_note = hands_off(target, me)
# ── 寫:說明先落地,再改欄位(改到一半炸掉時,票上至少說得出發生了什麼)──
dref = "%s/%s#%d" % dparts if dparts else None
lines = [ident, "",
"🗂️ **分診**inkstone/ISEP#159%s" % _sign(argv), ""]
if dup:
lines += ["- 與 %s 重複 → 關掉這張,後續在那張" % dref,
"- 為什麼:%s" % why]
else:
lines += ["- 狀態:(沒有 s/*)→ `%s`" % to,
"- 為什麼:%s" % why]
api("/repos/%s/%s/issues/%d/comments" % (owner, repo, num), {"body": "\n".join(lines)})
keep = [n for n in before_labels if not n.startswith("s/")] + [new_status]
api("/repos/%s/%s/issues/%d/labels" % (owner, repo, num), {"labels": [ids[n] for n in keep]}, method="PUT")
if dup:
api("/repos/%s/%s/issues/%d" % (owner, repo, num), {"state": "closed"}, method="PATCH")
if target_note is None:
to_ident = _ident_line(name, dparts[0], dparts[1])
api("/repos/%s/%s/issues/%d/comments" % dparts,
{"body": "\n".join([to_ident, "", "🗂️ %s 與這張重複,已關掉併到這裡(%s%s" % (ref, why, _sign(argv))])})
print("✅ 分診 %s → %s" % (ref, ("重複,併入 %s(已關)" % dref) if dup else to))
print(" 票上留了說明:%s" % why)
if dup and target_note:
print(" ⚪ 沒在 %s 留指回來的留言:%s" % (dref, target_note))
def _triage_list(argv):
owner = ORG
me = _me()
issues = pool_issues(owner)
pending = triage_candidates(issues, me)
show = issues if "--all" in argv else pending
if "--json" in argv:
rows = [{"ref": _ref_of(it), "title": it.get("title"), "status": _status_of(it),
"milestone": (it.get("milestone") or {}).get("title"),
"triageable": triageable(it, me)[0], "created_at": it.get("created_at")}
for it in sorted(show, key=lambda it: _ref_of(it))]
print(json.dumps(rows, ensure_ascii=False, indent=2))
return
untouched = sum(1 for it in issues if not _status_of(it) and not triageable(it, me)[0])
print("🗂️ open 票池 %d 張;待分診 %d 張(另有 %d 張沒 s/* 但掛 Human/指派給別人,不碰)"
% (len(issues), len(pending), untouched))
for it in sorted(show, key=lambda it: _ref_of(it)):
st = "、".join(_status_of(it)) or "(沒有 s/*"
print(" %-28s %-12s %s" % (_ref_of(it), st, (it.get("title") or "")[:70]))
if pending:
print("\n📌 %s" % _next_cmd("待分診", _ref_of(pending[0])))
# ── compose ──────────────────────────────────────────────────────────────
def _version_like(title):
"""名字就是版本號(`1.4.52 雲端零件包`)——開工規約第 5 條:版本號是 deliverable,不是名字。"""
return bool(re.match(r"^\s*[vV]?\d+(\.\d+)+", title or ""))
def cmd_compose(argv):
"""把已分診的票組成一個 InkStoneCo 里程碑(載體票+前置),先全部驗過才寫。"""
if not argv or argv[0] in ("-h", "--help"):
die(usage_text("compose"), 0 if argv else 2)
from datetime import datetime, timedelta, timezone
title = (_opt(argv, "--title") or "").strip()
goal = (_opt(argv, "--goal") or "").strip()
due_s = (_opt(argv, "--due") or "").strip()
carrier_s = _opt(argv, "--carrier")
name = _opt(argv, "--name") or "subagent"
members_s = _positional(argv, ("--title", "--goal", "--due", "--carrier", "--name", "--sign"))
problems = []
if not title:
problems.append("--title 要寫:名字回答「這一輪要讓什麼變成真的」(開工規約第 5 條)")
elif _version_like(title):
problems.append("--title「%s」以版本號開頭——版本號是 deliverable 不是名字(開工規約第 5 條,leo 2026-08-24" % title)
if not goal:
problems.append("--goal 要寫一段話:這一輪做完,用戶那邊什麼會變成真的")
due = None
now = _now()
try:
due = datetime.strptime(due_s, "%Y-%m-%d").replace(hour=23, minute=59, second=59, tzinfo=timezone.utc)
except ValueError:
problems.append("--due 要是 YYYY-MM-DD(開工規約第 6 條:期限一律要填),你給的是「%s」" % due_s)
if due and not (now < due <= now + timedelta(days=COMPOSE_MAX_DAYS + 1)):
problems.append("--due %s 不在今天之後 %d 天內——過去的期限等於已逾期,太遠的期限是「沒有期限」穿了期限的衣服"
% (due_s, COMPOSE_MAX_DAYS))
carrier = None
if not carrier_s:
problems.append("--carrier 要給一張 inkstone/InkStoneCo 的既有票當載體(開工規約第 7 條;不開新票)")
else:
carrier = parse_ref(carrier_s)
if carrier[:2] != (ORG, CARRIER_REPO):
problems.append("--carrier %s 不在 %s/%s——里程碑一律建在 InkStoneCo,載體票也在那裡" % (carrier_s, ORG, CARRIER_REPO))
members = []
for m in members_s:
r = parse_ref(m)
if r == carrier:
problems.append("%s 是載體票本身,不用再列成前置" % m)
elif r not in members:
members.append(r)
if not members:
problems.append("至少列一張前置票(別 repo 的執行票)——沒有前置的里程碑交不出版本")
if problems:
die("🚫 compose 還沒送出(一筆都沒寫):\n%s\n\n%s" % (
"\n".join(" · " + p for p in problems), usage_text("compose")))
ident = _ident_line(name, ORG, CARRIER_REPO)
check_identity(ident, "組里程碑留言")
# ── 線上驗:全部讀過一次,理由收齊才決定寫不寫 ──
me = _me()
ML = _mainline_lib()
cref = "%s/%s#%d" % carrier
citem = _fetch(*carrier)
why = hands_off(citem, me)
if why:
problems.append("載體 %s%s" % (cref, why))
elif citem.get("milestone") and (citem["milestone"].get("state") or "open") == "open":
problems.append("載體 %s 已經掛在里程碑「%s」——一張載體只載一輪" % (cref, citem["milestone"].get("title")))
else:
for b in _blocks(*carrier):
bm = (b or {}).get("milestone") or {}
if (b or {}).get("state", "open") == "open" and bm and (bm.get("state") or "open") == "open":
problems.append("載體 %s 已經是 %s(里程碑「%s」)的前置——它那一輪還在跑,不能再載另一輪"
% (cref, ML.issue_ref(b) or "?", bm.get("title")))
break
existing = api("/repos/%s/%s/milestones?state=open&limit=100" % (ORG, CARRIER_REPO)) or []
for m in existing:
if isinstance(m, dict) and ML.norm(m.get("title")) == ML.norm(title):
problems.append("InkStoneCo 已經有 open 里程碑「%s」(id %s)——同名不再建一個" % (m.get("title"), m.get("id")))
rows = []
for mo, mr, mn in members:
mref = "%s/%s#%d" % (mo, mr, mn)
it = _fetch(mo, mr, mn)
why = hands_off(it, me)
if why:
problems.append("前置 %s%s" % (mref, why))
continue
st = _status_of(it)
if len(st) != 1 or st[0] not in COMPOSE_MEMBER_STATUSES:
problems.append("前置 %s 還沒分診(現在 %s)——先 `ticket triage` 決定要不要做" % (mref, st or "沒有 s/*"))
continue
ims = it.get("milestone") or {}
if ims and (ims.get("state") or "open") == "open":
problems.append("前置 %s 已經掛在里程碑「%s」" % (mref, ims.get("title")))
continue
for b in _blocks(mo, mr, mn):
bm = (b or {}).get("milestone") or {}
if (b or {}).get("state", "open") == "open" and bm and (bm.get("state") or "open") == "open":
problems.append("前置 %s 已經是 %s(里程碑「%s」)的前置——一張票同時只排一輪"
% (mref, ML.issue_ref(b) or "?", bm.get("title")))
break
else:
rows.append(((mo, mr, mn), mref, it, st[0]))
if problems:
die("🚫 compose 驗不過(一筆都沒寫):\n%s\n\n"
"出路:把上面每一格換掉(換一張載體、拿掉那張前置、先 triage),再送一次。"
"掛 Human/指派給 Leo 的票不會被組進來——那是 leo 的。" % "\n".join(" · " + p for p in problems))
# ── 寫 ──
member_refs = [r for _, r, _, _ in rows]
desc = "\n".join([
"這一輪要讓什麼變成真的:%s" % goal,
"deliverable=一個總管在 stage 驗過的新版本(版本號由總管定);前置全部關閉=100%=才出版本。",
"載體票:%s;前置:%s" % (cref, "、".join(member_refs)),
"組票:%s %sinkstone/ISEP#159CLAUDE.md 開工規約 17 條)" % (_sign(argv) or name, now.strftime("%Y-%m-%d")),
])
ms = api("/repos/%s/%s/milestones" % (ORG, CARRIER_REPO),
{"title": title, "description": desc, "due_on": due.strftime("%Y-%m-%dT%H:%M:%SZ")})
mid = (ms or {}).get("id")
if not mid:
die("🔴 建里程碑沒拿到 id(Gitea 回:%s)——後面一筆都沒寫" % str(ms)[:200])
url = "%s/%s/%s/milestone/%s" % (HOST, ORG, CARRIER_REPO, mid)
print("✅ 里程碑「%s」(%s/%s milestone %s,期限 %s" % (title, ORG, CARRIER_REPO, mid, due_s))
api("/repos/%s/%s/issues/%d" % carrier, {"milestone": mid}, method="PATCH")
print(" 載體 %s 掛上里程碑" % cref)
for (mo, mr, mn), mref, it, st in rows:
add_dependency(carrier, (mo, mr, mn))
moved = ""
if st == "s/backlog":
ids = _labels_of(mo, mr)
keep = [n for n in _label_names(it) if not n.startswith("s/")] + ["s/todo"]
if all(n in ids for n in keep):
api("/repos/%s/%s/issues/%d/labels" % (mo, mr, mn), {"labels": [ids[n] for n in keep]}, method="PUT")
moved = "s/backlog → s/todo"
api("/repos/%s/%s/issues/%d/comments" % (mo, mr, mn),
{"body": "\n".join([_ident_line(name, mo, mr), "",
"🧩 排進里程碑「%s」(載體 %s,期限 %s%s %s" % (title, cref, due_s, moved, _sign(argv)),
url])})
print(" 前置 %s%s" % (mref, moved))
api("/repos/%s/%s/issues/%d/comments" % carrier,
{"body": "\n".join([ident, "", "🧩 **組里程碑**inkstone/ISEP#159%s" % _sign(argv), "",
"- 里程碑:「%s」→ %s" % (title, url),
"- 期限:%s" % due_s,
"- 這一輪要讓什麼變成真的:%s" % goal,
"- 前置(全部關閉=100%%=才出版本):%s" % "、".join(member_refs),
"- deliverable:一個總管在 stage 驗過的新版本(版本號由總管定)"])})
print(" %s" % url)
print("\n📌 接下來照里程碑做:`%s pick --pool` 會先回這個里程碑的前置票(還沒到期的里程碑排在散票之前)" % self_path())
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, "pick": cmd_pick, "claim": cmd_claim}
"usage": cmd_usage, "pick": cmd_pick, "claim": cmd_claim,
"triage": cmd_triage, "compose": cmd_compose}
if __name__ == "__main__":
if len(sys.argv) < 2 or sys.argv[1] not in CMDS: