Leo
48befe8906
薄殼 SessionStart 跑在 bootstrap.sh 之前，InkStoneCo/ 還沒 clone、家目錄空 ⇒ 兩份都不在 ⇒ countdown.py 退到「期限最近的 milestone」⇒ 第一則 prompt 印逾期 341 小時的舊線。機制沒壞，是順序。 - hooks/lib/mainline.py：第三個來源 fetch_gitea()（本檔唯一會打網路的函式，只有 SessionStart 的 `mainline refresh` 叫它；load()/line() 照舊零網路）。優先序寫進檔頭：家目錄 ＞ repo 那份 ＞ Gitea 那份。 匿名優先，匿名被拒（401/403/404）且拿得到 token 才帶 token 再問一次；讀不到回 None，逾時有上限。 - scripts/mainline refresh：兩份都不在 ⇒ fetch_gitea 一次、只寫家目錄（不 mirror）；token 拿不到不炸、不抱怨。 - hooks/tests/mainline-gitea-fallback.test.sh：31 條，假 Gitea 有 public／private／500／garbage／notitle／hang 六種回法。 假設（票上紅線寫「一律匿名不帶 token」，其前提「Gitea 匿名讀得到」實測為假）： inkstone/InkStoneCo 在 Gitea 是 private——匿名打 raw／contents／repo API 都 404，ISEP 同端點 200。 ⇒ 照字面做這張票在真環境永遠讀不到。改成匿名優先、被拒才帶 token（同一條 SessionStart 鏈上 countdown-milestone-refresh.sh 本來就帶 GITEA_TOKEN_CLAUDE_CODE 打 Gitea；D20 管的是 GitHub）。 真跑：帶 token 2.06s 讀到 inkstone/Arcrun#48；拔 token rc 0、零檔案、「現在沒有主線」。 版本：待總管定版。 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0178ef1fGw3XeZtpN7LaZrm4
424 lines
18 KiB
Python
424 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""hooks/lib/mainline.py —— 「現在的主線是哪一個」這件事的**唯一存放處**。
|
||
|
||
這是 helper，不是閘（inkstone/ISEP#40 S7：`lib/` 底下的東西不算一支手寫的閘）。
|
||
三個呼叫者共用它，所以「主線是誰」的答案只有一份：
|
||
· `scripts/mainline` —— 人問「現在的主線是哪一個」
|
||
· `hooks/lib/countdown.py` —— 每一則回覆眼前那一行的第二段
|
||
· `hooks/mainline-focus-guard.sh` —— 派了一張不屬於主線的票就攔一次
|
||
|
||
── 為什麼要有這個檔（inkstone/ISEP#82）──────────────────────────────
|
||
2026-08-27 實查：`GET /repos/inkstone/*/milestones?state=open` 回 **14 個**，
|
||
其中「Mira 現代化」這個名字同時活在 5 個 repo，還有 5 個已經逾期。
|
||
SOP 說「有 active milestone 時只做該 milestone 的事」——但**「那個」在現場沒有指涉對象**。
|
||
|
||
leo 的原話是「**14 個里程碑同時亮著卻不知道該看哪個**」。
|
||
⇒ 要解的不是「milestone 太多」（跨 repo 同名是既有做法，v0.6.0 §4／SOP S7-4），
|
||
要解的是「**這一刻我在做哪一條**」沒有答案。
|
||
|
||
⇒ 所以這個檔的形狀就是答案的形狀：**一個檔，放得下一條**。
|
||
沒有清單、沒有優先序、沒有「前三名」——被覆蓋掉的那條就不再是主線。
|
||
問「現在的主線是哪一個」永遠只會拿到 0 或 1 個答案，不會拿到 14 個。
|
||
|
||
── 主線是一個「名字」，錨在一個具體的 milestone 上 ──────────────────
|
||
Gitea 的 milestone 不能跨 repo，而同一條線常常同時開在 5 個 repo（票上實查）。
|
||
所以：
|
||
· **錨**（anchor）＝ `owner/repo#<milestone id>`：期限、進度、目標宣告都讀它
|
||
· **成員判定**＝ 票所屬 milestone 的**標題**與主線標題相同（不分 repo）
|
||
⇒ 「同名跨 repo」照舊能用，而「現在在做哪一條」仍然只有一個答案。
|
||
|
||
── `load()`／`line()` 從不打網路 ─────────────────────────────────
|
||
它們被 `UserPromptSubmit` 用（每一則訊息的關鍵路徑）。在那裡打 HTTP ＝ 每講一句話
|
||
都先等一次網路。網路那一半全部在 `scripts/mainline`（set／refresh／adopt），
|
||
本支的讀寫只碰 state 檔。讀不到 ⇒ 回「沒有主線」，**不編一條出來**。
|
||
唯一的例外是 `fetch_gitea()`（inkstone/ISEP#140，見下方第三個來源那段）：
|
||
它會打一次 Gitea，但**只有 SessionStart 的 `scripts/mainline refresh` 叫它**，
|
||
`load()`／`line()` 永遠不叫——所以每一則訊息的路徑上仍然沒有網路。
|
||
|
||
── 三個來源，優先序寫死：家目錄 ＞ repo 那份 ＞ Gitea main 那份 ───────
|
||
① 家目錄 `~/.claude/isep-countdown/mainline.json`（這台機器上 refresh 過、有進度）
|
||
② repo 那份 `InkStoneCo/system-dev/mainline.json`（隨 repo 走，inkstone/ISEP#130）
|
||
③ Gitea 上 `inkstone/InkStoneCo` main 的同一個檔（inkstone/ISEP#140）
|
||
①② 之間誰算數看 `set_at`（見 `load()`）；③ **只在 ①② 都不存在時**被問一次，
|
||
抓到就寫進 ①，之後跟 ① 一樣對待。③ 讀的就是 ② 那個檔，只是從 Gitea 拿而不是從
|
||
工作樹拿——它不是第二套主線來源。
|
||
"""
|
||
import json
|
||
import os
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
TAIPEI = timezone(timedelta(hours=8))
|
||
MARKER = "🎯"
|
||
STATE_NAME = "mainline.json"
|
||
JUMP_LOG = "mainline-jumps.jsonl"
|
||
|
||
|
||
def state_dir() -> str:
|
||
"""與倒數共用同一個狀態目錄——兩行字是同一個注入點的兩半，狀態不該分家。"""
|
||
d = os.environ.get("ISEP_COUNTDOWN_STATE_DIR", "").strip()
|
||
if not d:
|
||
d = os.path.join(os.path.expanduser("~"), ".claude", "isep-countdown")
|
||
try:
|
||
os.makedirs(d, exist_ok=True)
|
||
except Exception:
|
||
pass
|
||
return d
|
||
|
||
|
||
def path() -> str:
|
||
return os.path.join(state_dir(), STATE_NAME)
|
||
|
||
|
||
# ── repo 裡的那一份（inkstone/ISEP#130）──────────────────────────────
|
||
#
|
||
# 主線原本只住在 `~/.claude/isep-countdown/mainline.json`——**這台機器的家目錄**。
|
||
# 雲端 session 是另一台機器：家目錄是空的，clone 下來的 repo 才是它唯一拿得到的東西。
|
||
# 2026-09-05 leo：「routine 每天早上開啟後抓不到任務」——不是 Gitea 沒票，
|
||
# 是雲端根本不知道主線是哪一條。
|
||
# ⇒ `inkstone/InkStoneCo` 的 `system-dev/mainline.json` 是**隨 repo 走的那一份**
|
||
# （PR inkstone/InkStoneCo#118 放進去的）。這裡的規矩：
|
||
# · 讀：家目錄那份沒有 ⇒ 讀 repo 那份（雲端就是這個形狀）。
|
||
# 兩份都在 ⇒ 用 `set_at` 較新的（誰後標的誰算數，不分本機雲端）。
|
||
# · 寫：`set`／`adopt`／`clear` 這種「主線是誰」的改動 ⇒ 兩份一起寫（`mirror=True`）。
|
||
# `refresh`／閘順手學到成員 ⇒ 只寫家目錄那份——否則每開一個 session
|
||
# InkStoneCo 的工作樹就髒一次，而髒的 diff 沒有人會去 commit。
|
||
# · repo 那份要 commit／開 PR 才到得了雲端；這支不碰 git。
|
||
REPO_REL = os.path.join("system-dev", STATE_NAME)
|
||
|
||
|
||
def repo_candidates():
|
||
"""repo 那一份可能在哪：明指 > 專案根 > 專案根底下的 InkStoneCo/（雲端薄殼的形狀）> cwd 同兩層。"""
|
||
out = []
|
||
forced = os.environ.get("ISEP_MAINLINE_REPO_FILE", "").strip()
|
||
if forced == "none":
|
||
return [] # 測試用：把「repo 那份」整個拿掉，外界定住
|
||
if forced:
|
||
out.append(forced)
|
||
roots = []
|
||
for r in (os.environ.get("CLAUDE_PROJECT_DIR", "").strip(), os.getcwd()):
|
||
if r:
|
||
roots += [r, os.path.join(r, "InkStoneCo")]
|
||
for r in roots:
|
||
p = os.path.join(r, REPO_REL)
|
||
if p not in out:
|
||
out.append(p)
|
||
return out
|
||
|
||
|
||
def repo_path(for_write=False):
|
||
"""讀：第一個**存在**的檔。寫：第一個 `system-dev/` 目錄存在的位置（檔可以還沒有）。
|
||
找不到 ⇒ None（雲端沒 clone InkStoneCo、或 cwd 在別的 repo 時都是正常狀態）。"""
|
||
for p in repo_candidates():
|
||
if for_write:
|
||
if os.path.isdir(os.path.dirname(p)):
|
||
return p
|
||
elif os.path.isfile(p):
|
||
return p
|
||
return None
|
||
|
||
|
||
def _valid(d):
|
||
"""一份主線檔長什麼樣才算數：dict、而且有 title。三個來源用同一把尺。"""
|
||
return d if (isinstance(d, dict) and d.get("title")) else None
|
||
|
||
|
||
def _read(p):
|
||
try:
|
||
with open(p) as f:
|
||
d = json.load(f)
|
||
except Exception:
|
||
return None
|
||
return _valid(d)
|
||
|
||
|
||
# ── Gitea main 上的那一份（inkstone/ISEP#140）────────────────────────
|
||
#
|
||
# 雲端薄殼 `youlinhsieh/inkstoneco` 的 SessionStart 跑在 `bootstrap.sh` 之前：那時
|
||
# `InkStoneCo/` 還沒 clone，家目錄又是空的 ⇒ 上面兩份都不存在 ⇒ `countdown.py` 退到
|
||
# 第二基準「期限最近的 milestone」⇒ 第一則 prompt 印出一條兩週前的舊線
|
||
# （票上實測：`🔴 主線 把管理這條線做對 已逾期 341 小時`＋`🎯 現在沒有主線`）。
|
||
# 機制沒壞，是順序。而「哪一條是主線」這個指標本來就在 Gitea 上讀得到。
|
||
#
|
||
# 規矩（跟上面「repo 裡的那一份」那段接起來，不另立一套）：
|
||
# · 只在家目錄與 repo 那份**都不存在**時問 Gitea，問一次；任一份在場就不問。
|
||
# · 只有 SessionStart 的 `scripts/mainline refresh` 會叫 `fetch_gitea()`；抓到就
|
||
# `save()` 進家目錄（不 mirror：雲端沒有 InkStoneCo/ 可寫，本機也不弄髒工作樹）。
|
||
# bootstrap 之後 repo 那份出現時，兩份 `set_at` 相同 ⇒ 家目錄那份算數（它有 refresh 過的進度）。
|
||
# · 匿名優先。匿名被拒（401／403／404）且環境拿得到 Gitea token 才帶 token 再問一次
|
||
# ——`inkstone/InkStoneCo` 在 Gitea 是 **private**（2026-09-07 實測：匿名打 raw／
|
||
# contents／repo API 都 404），沒 token 就真的讀不到。
|
||
# · 讀不到（斷網／5xx／不是 JSON／沒 title／逾時）⇒ 回 None，退回現行行為，不報錯。
|
||
# 逾時有上限（`ISEP_MAINLINE_GITEA_TIMEOUT`，預設 4 秒，最多問兩次）。
|
||
# · 測試用：`ISEP_MAINLINE_GITEA_URL=none` 把這個來源整個拿掉；`TICKET_HOST` 換主機
|
||
# （既有測試都把它指到連不上的位址，所以它們照舊離線）。
|
||
GITEA_HOST_DEFAULT = "https://git.uncle6.me"
|
||
GITEA_REPO = "inkstone/InkStoneCo"
|
||
GITEA_REF = "main"
|
||
|
||
|
||
def gitea_url() -> str:
|
||
forced = os.environ.get("ISEP_MAINLINE_GITEA_URL", "").strip()
|
||
if forced == "none":
|
||
return ""
|
||
if forced:
|
||
return forced
|
||
host = (os.environ.get("TICKET_HOST", "").strip() or GITEA_HOST_DEFAULT).rstrip("/")
|
||
return "%s/api/v1/repos/%s/raw/%s?ref=%s" % (
|
||
host, GITEA_REPO, REPO_REL.replace(os.sep, "/"), GITEA_REF)
|
||
|
||
|
||
def fetch_gitea(token="", timeout=None):
|
||
"""去 Gitea 讀 `inkstone/InkStoneCo` main 上的 `system-dev/mainline.json`。
|
||
|
||
回一份跟 `_read()` 同樣通過 `_valid()` 的 dict，或 None。**永遠不 raise。**
|
||
`token`：字串，或一個回字串的函式（只在匿名被拒時才會被叫）。
|
||
🔴 這是本檔唯一會打網路的函式，只准 SessionStart 的 `scripts/mainline refresh` 叫；
|
||
`load()`／`line()` 走的是每一則訊息的關鍵路徑，不准碰它。
|
||
"""
|
||
url = gitea_url()
|
||
if not url:
|
||
return None
|
||
if timeout is None:
|
||
try:
|
||
timeout = float(os.environ.get("ISEP_MAINLINE_GITEA_TIMEOUT", "") or 4)
|
||
except Exception:
|
||
timeout = 4.0
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
def get(headers):
|
||
req = urllib.request.Request(url, headers=headers)
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
return r.read().decode("utf-8", "replace")
|
||
|
||
try:
|
||
text = get({}) # 匿名優先（D20 的形狀：讀不帶憑證）
|
||
except urllib.error.HTTPError as e:
|
||
if e.code not in (401, 403, 404):
|
||
return None
|
||
# `token` 可以是字串，也可以是「被拒了才去拿」的函式——拿 token 可能要掃 git remote、
|
||
# 可能拿不到而想開口抱怨，匿名讀得到的時候根本不該走到那裡。
|
||
tok = token() if callable(token) else token
|
||
if not tok:
|
||
return None
|
||
try:
|
||
text = get({"Authorization": "token " + tok}) # private repo：帶 token 再問一次
|
||
except Exception:
|
||
return None
|
||
except Exception:
|
||
return None
|
||
try:
|
||
return _valid(json.loads(text))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def load():
|
||
"""讀主線。沒有／壞掉 ⇒ None（＝現在沒有主線），**不是例外**。
|
||
|
||
「沒有主線」是一個正常狀態，不是故障：ISEP#82 驗收第 4 條寫死了
|
||
「沒有任何主線時不能整組壞掉」。所以這支從不 raise。
|
||
|
||
家目錄那份與 repo 那份都在時，`set_at` 較新的算數（格式固定 `YYYY-MM-DD HH:MM`，
|
||
字串比大小就是時間比大小）；一樣新 ⇒ 家目錄那份（它才有 refresh 過的進度）。
|
||
"""
|
||
local = _read(path())
|
||
rp = repo_path()
|
||
remote = _read(rp) if rp else None
|
||
if local and remote:
|
||
return remote if (remote.get("set_at") or "") > (local.get("set_at") or "") else local
|
||
return local or remote
|
||
|
||
|
||
def save(d, mirror=False) -> bool:
|
||
"""寫家目錄那份；`mirror=True` 時同時寫 repo 那份（只有「主線是誰」變了才這樣叫）。
|
||
回 True/False 只看家目錄那份——repo 那份寫不寫得進去用 `repo_path(for_write=True)` 另外問。"""
|
||
try:
|
||
with open(path(), "w") as f:
|
||
json.dump(d, f, ensure_ascii=False, indent=2)
|
||
except Exception:
|
||
return False
|
||
if mirror:
|
||
rp = repo_path(for_write=True)
|
||
if rp:
|
||
try:
|
||
with open(rp, "w") as f:
|
||
json.dump(d, f, ensure_ascii=False, indent=2)
|
||
f.write("\n")
|
||
except Exception:
|
||
pass
|
||
return True
|
||
|
||
|
||
def clear(mirror=False) -> bool:
|
||
ok = True
|
||
try:
|
||
os.remove(path())
|
||
except FileNotFoundError:
|
||
pass
|
||
except Exception:
|
||
ok = False
|
||
if mirror:
|
||
rp = repo_path()
|
||
if rp:
|
||
try:
|
||
os.remove(rp)
|
||
except Exception:
|
||
ok = False
|
||
return ok
|
||
|
||
|
||
def parse_ts(raw):
|
||
if not isinstance(raw, str) or not raw.strip():
|
||
return None
|
||
s = raw.strip().replace("Z", "+00:00")
|
||
try:
|
||
dt = datetime.fromisoformat(s)
|
||
except Exception:
|
||
try:
|
||
dt = datetime.fromisoformat(s + "T23:59:59+08:00")
|
||
except Exception:
|
||
return None
|
||
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
||
|
||
|
||
def due_of(ms):
|
||
return parse_ts((ms or {}).get("due_on") or "")
|
||
|
||
|
||
def norm(s) -> str:
|
||
"""比對標題用。只做「看不見的差異」的正規化——大小寫與空白。
|
||
|
||
🔴 刻意不做同義詞、不做模糊比對：主線是**被指定的**，不是被猜出來的。
|
||
猜錯的方向是「把不屬於主線的票放行」，那正是這張票要擋的事。
|
||
"""
|
||
return "".join((s or "").split()).lower()
|
||
|
||
|
||
def ref_of(ms) -> str:
|
||
if not ms:
|
||
return ""
|
||
return "%s/%s#%s" % (ms.get("owner") or "?", ms.get("repo") or "?", ms.get("id") or "?")
|
||
|
||
|
||
def members(ms):
|
||
return set(ms.get("members") or []) if ms else set()
|
||
|
||
|
||
def belongs(ticket_ref, ms, ticket_milestone_title=None):
|
||
"""這張票屬不屬於主線？回 True／False／None。
|
||
|
||
None ＝ **不知道**（快取裡沒有這張票，而呼叫端也沒查到它的 milestone）。
|
||
🔴 「不知道」不等於「不屬於」——這條分界就是誤攔與否的分水嶺。
|
||
呼叫的閘拿到 None 時該放行（本 repo 心法第 2 條：誤攔比漏擋更該修）。
|
||
"""
|
||
if not ms:
|
||
return True # 沒有主線 ⇒ 沒有「不屬於主線」這回事
|
||
ref = (ticket_ref or "").strip()
|
||
if ref and ref in members(ms):
|
||
return True
|
||
if ticket_milestone_title is not None:
|
||
return norm(ticket_milestone_title) == norm(ms.get("title"))
|
||
return None
|
||
|
||
|
||
def fixture_titles():
|
||
"""測試用：把「去 Gitea 問這張票掛在哪」換成一份現成的答案。
|
||
|
||
`ISEP_MAINLINE_FIXTURE` 指向一個 JSON：`{"owner/repo#N": "milestone 標題"}`。
|
||
設了它，閘就不打網路，直接拿這份當事實。
|
||
|
||
🔴 這與 `ISEP_COUNTDOWN_NOW`（把時鐘定住）是同一個性質的東西：
|
||
**把外界定住，讓閘能離線、可重複地被測**。正式環境不會設它；
|
||
沒設 ⇒ 這支回 None，判決完全走真實的 Gitea。
|
||
"""
|
||
p = os.environ.get("ISEP_MAINLINE_FIXTURE", "").strip()
|
||
if not p:
|
||
return None
|
||
try:
|
||
with open(p) as f:
|
||
d = json.load(f)
|
||
return d if isinstance(d, dict) else None
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def record_jump(ticket_ref, ms):
|
||
"""把「派了一張不在主線上的票」記一筆。**只留痕，不判罪。**
|
||
|
||
跳線本身常常是對的（補收、真的插件事）。留痕是為了讓「這條線今天被岔開幾次」
|
||
事後數得出來——沒有紀錄的話，跳線與專注在機器上長得一模一樣。
|
||
"""
|
||
try:
|
||
with open(os.path.join(state_dir(), JUMP_LOG), "a") as f:
|
||
f.write(json.dumps({
|
||
"at": datetime.now(timezone.utc).isoformat(),
|
||
"ticket": ticket_ref,
|
||
"mainline": ref_of(ms),
|
||
"mainline_title": (ms or {}).get("title"),
|
||
}, ensure_ascii=False) + "\n")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _human(delta: timedelta) -> str:
|
||
days = int(abs(delta).total_seconds()) // 86400
|
||
if days >= 1:
|
||
return "%d 天" % days
|
||
hours = int(abs(delta).total_seconds()) // 3600
|
||
return "%d 小時" % hours
|
||
|
||
|
||
def due_phrase(ms, now=None) -> str:
|
||
"""期限那一段。沒設期限就說沒設——不准編一個日期出來。"""
|
||
dt = due_of(ms)
|
||
if not dt:
|
||
return "沒設期限"
|
||
now = now or datetime.now(timezone.utc)
|
||
day = dt.astimezone(TAIPEI).strftime("%Y-%m-%d")
|
||
gap = dt - now
|
||
if gap.total_seconds() >= 0:
|
||
return "%s（剩 %s）" % (day, _human(gap))
|
||
return "%s（🔴 已逾期 %s）" % (day, _human(gap))
|
||
|
||
|
||
def progress(ms):
|
||
o = int((ms or {}).get("open_issues") or 0)
|
||
c = int((ms or {}).get("closed_issues") or 0)
|
||
total = o + c
|
||
pct = int(round(c * 100.0 / total)) if total else 0
|
||
return c, total, pct
|
||
|
||
|
||
def line(now=None) -> str:
|
||
"""注入用的那一行（倒數那一行的第二行）。**沒有主線時也要回一句話。**
|
||
|
||
ISEP#82 驗收第 4 條：沒有主線不能整組壞掉，要講得出「現在沒有主線」。
|
||
⇒ 這支永遠回一個非空字串，永遠不 raise。
|
||
"""
|
||
ms = load()
|
||
if not ms:
|
||
return ("%s 現在沒有主線——沒有任何 milestone 被標成「現在在做的那一條」。"
|
||
"要標：`scripts/mainline set <owner/repo#milestone_id>`"
|
||
"（先看有哪些：`scripts/mainline list`）" % MARKER)
|
||
c, total, pct = progress(ms)
|
||
head = "%s 主線：%s「%s」｜%s｜%d／%d 張已關（%d%%）" % (
|
||
MARKER, ref_of(ms), ms.get("title") or "?", due_phrase(ms, now), c, total, pct)
|
||
goal = (ms.get("description") or "").strip().splitlines()
|
||
if goal:
|
||
head += "\n 目標：" + goal[0].strip()[:120]
|
||
return head
|
||
|
||
|
||
def countdown_segment(now=None):
|
||
"""給 `countdown.py` 用的 (名稱, 期限) ——讓 ⏱ 那一行說得出主線是誰。
|
||
|
||
為什麼要塞進 ⏱ 那一行：⏱ 那一行**已經有一支 Stop 閘在查它有沒有被戴上**
|
||
（inkstone/ISEP#63）。主線的名字掛在那一行上，就跟著那道已經驗過的閘一起
|
||
到 leo 眼前——不必為了同一件事再立第二道會擋人的閘。
|
||
"""
|
||
ms = load()
|
||
if not ms:
|
||
return None
|
||
return (ms.get("title") or "主線", due_of(ms))
|