Claude Code
ac2f89c610
14 個 open milestone 同時亮著，其中「Mira 現代化」同名活在 5 個 repo，
所以 SOP 說的「那個 active milestone」在現場沒有指涉對象。
- hooks/lib/mainline.py 主線的唯一存放處（一個檔放得下一條），從不打網路
- scripts/mainline show/list/set/clear/refresh/adopt/has
- hooks/mainline-focus-guard.sh 派了不在主線上的票 ⇒ 攔一次，問補收還是跳線
- hooks/lib/countdown.py ⏱ 那一行的主線改讀「被標定的」，蓋過「期限最近」的猜測
- hooks/countdown-guard.sh 同一個注入點加第二行 🎯（ISEP#63 那半不動）
- 測試 24 條（離線）＋ countdown 原有 20 條仍全綠
242 lines
9.3 KiB
Python
Executable File
242 lines
9.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""hooks/lib/countdown.py —— 算「這件事已經花了多久／還剩多久」的那一行。
|
||
|
||
這是 helper，不是閘（inkstone/ISEP#40 S7：lib/ 底下的東西不算一支手寫的閘）。
|
||
兩個呼叫者共用它，所以時間的算法只有一份：
|
||
· UserPromptSubmit（`countdown-guard.sh` 注入模式）—— 把這一行送到模型眼前
|
||
· Stop（同一支的查核模式）—— 擋下來時要把同一行原樣印出來給它抄
|
||
|
||
── 基準從哪一刻算起（inkstone/ISEP#63 要我自己判斷並說明理由）──────────
|
||
leo 的原話是「**隨時看自己拖了多久**」。「拖」指向的是**任務**，不是 session，
|
||
但「任務」在機器上沒有起點——沒有任何欄位記著「我什麼時候接手這條線」。
|
||
可機械取得、又真的對應到一個交付的起點，只有一個：**這段對話的第一則訊息**。
|
||
|
||
而那正好就是總管的交貨單位：CLAUDE.md 規則三點七「**一段對話 ＝ 一個 release**」。
|
||
⇒ 已過時間 ＝ 這個 release 已經燒掉多久。**這個數字不需要任何人維護，也不會說謊。**
|
||
|
||
resume 的 session 取「transcript 第一列的時間」而不是「hook 第一次被叫到的時間」：
|
||
接關不是重新開始，昨天燒掉的兩小時仍然算在這個交付上。兩者都拿得到時取較早的。
|
||
|
||
── 第二個數字：今天的收工線 ──────────────────────────────────────────
|
||
leo 2026-08-28 07:40：「今天我整天上課……**下午四點左右**希望已經都完成了」。
|
||
⇒ 台北 16:00。過了 16:00 之後**不立刻換算到明天**——那會讓超時消失，
|
||
而超時正是他要看的東西。過線後四小時內顯示「已超過 X」，
|
||
台北 20:00 之後才滾到隔天（那時已經是下一個工作日的倒數了）。
|
||
|
||
── 第三個數字：主線 milestone 的期限（可選）────────────────────────────
|
||
從快取檔讀，**這支從不打網路**：UserPromptSubmit 走在每一則訊息的關鍵路徑上，
|
||
在那裡打網路 ＝ 每一句話都先等一次 HTTP。快取由 SessionStart 那支負責更新
|
||
（`scripts/countdown-milestone-refresh.sh`），拿不到就整段不顯示，不編數字。
|
||
|
||
🔴 **來源有兩個，優先序寫死（inkstone/ISEP#82）**：
|
||
① **被標定的主線**（`hooks/lib/mainline.py`，`scripts/mainline set` 寫的）—— 優先
|
||
② 期限最近的那一個（`milestone-due` 快取，`countdown-milestone-refresh.sh` 寫的）
|
||
② 是**猜**出來的（「最近到期的大概就是在做的那個」），而 ISEP#82 的整件事就是
|
||
「那個 active milestone」在 14 個裡沒有指涉對象。⇒ 一旦有人真的標了主線，
|
||
就不准再用猜的那個蓋過它。兩個都沒有 ⇒ 整段不顯示。
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
TAIPEI = timezone(timedelta(hours=8))
|
||
MARKER = "⏱"
|
||
|
||
|
||
def _now() -> datetime:
|
||
"""現在。測試用 ISEP_COUNTDOWN_NOW（epoch 秒）把時鐘定住。"""
|
||
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 _parse_ts(raw) -> "datetime | None":
|
||
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:
|
||
return None
|
||
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
||
|
||
|
||
def transcript_start(path: str) -> "datetime | None":
|
||
"""transcript 第一列帶時間戳的那一列 ＝ 這段對話真正的起點。"""
|
||
if not path or not os.path.exists(path):
|
||
return None
|
||
try:
|
||
with open(path) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
row = json.loads(line)
|
||
except Exception:
|
||
continue
|
||
dt = _parse_ts(row.get("timestamp"))
|
||
if dt:
|
||
return dt
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
|
||
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 session_start(session_id: str, transcript_path: str, now: datetime) -> datetime:
|
||
"""起點 ＝ transcript 第一列與「第一次看到這個 session」兩者取較早。
|
||
|
||
第一次看到就落一個檔；transcript 沒有時間戳（或讀不到）時它就是唯一的憑據。
|
||
"""
|
||
cands = []
|
||
ts = transcript_start(transcript_path)
|
||
if ts:
|
||
cands.append(ts)
|
||
|
||
sid = "".join(c for c in (session_id or "") if c.isalnum() or c in "-_")[:80]
|
||
if sid:
|
||
p = os.path.join(state_dir(), sid + ".start")
|
||
seen = None
|
||
try:
|
||
with open(p) as f:
|
||
seen = datetime.fromtimestamp(float(f.read().strip()), tz=timezone.utc)
|
||
except Exception:
|
||
seen = None
|
||
if seen is None:
|
||
seen = min(cands) if cands else now
|
||
try:
|
||
with open(p, "w") as f:
|
||
f.write("%d" % seen.timestamp())
|
||
except Exception:
|
||
pass
|
||
cands.append(seen)
|
||
|
||
return min(cands) if cands else now
|
||
|
||
|
||
def deadline(now: datetime) -> datetime:
|
||
"""今天的收工線（台北 HH:MM）。過線後四小時內不滾，讓超時看得見。"""
|
||
hhmm = os.environ.get("ISEP_COUNTDOWN_DEADLINE", "16:00").strip() or "16:00"
|
||
try:
|
||
hh, mm = (int(x) for x in hhmm.split(":", 1))
|
||
except Exception:
|
||
hh, mm = 16, 0
|
||
local = now.astimezone(TAIPEI)
|
||
d = local.replace(hour=hh, minute=mm, second=0, microsecond=0)
|
||
if local >= d + timedelta(hours=4):
|
||
d += timedelta(days=1)
|
||
return d
|
||
|
||
|
||
def _mainline():
|
||
"""借 `hooks/lib/mainline.py` 讀「被標定的主線」。拿不到就當沒有（fail-open）。
|
||
|
||
刻意用檔案路徑載入而不是 `import mainline`：本檔自己也是被
|
||
`spec_from_file_location` 載進去的，`sys.path` 裡沒有 `hooks/lib`。
|
||
"""
|
||
try:
|
||
import importlib.util
|
||
p = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mainline.py")
|
||
spec = importlib.util.spec_from_file_location("isep_mainline", p)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def milestone() -> "tuple[str, datetime | None] | None":
|
||
"""主線 milestone 的期限。優先讀被標定的主線，其次讀「期限最近」的快取。
|
||
|
||
快取檔一行：`YYYY-MM-DD|<名稱>`。兩個都沒有就沒有（回 None，整段不顯示）。
|
||
期限可能是 None（被標定的主線沒設期限）——那時只說名字，不編一個日期。
|
||
"""
|
||
raw = os.environ.get("ISEP_MILESTONE_DUE", "").strip()
|
||
name = os.environ.get("ISEP_MILESTONE_NAME", "").strip()
|
||
if not raw:
|
||
mod = _mainline()
|
||
if mod is not None:
|
||
try:
|
||
seg = mod.countdown_segment()
|
||
except Exception:
|
||
seg = None
|
||
if seg:
|
||
return seg
|
||
if not raw:
|
||
p = os.path.join(state_dir(), "milestone-due")
|
||
try:
|
||
with open(p) as f:
|
||
raw = f.read().strip()
|
||
except Exception:
|
||
return None
|
||
if "|" in raw:
|
||
raw, name = raw.split("|", 1)
|
||
raw, name = raw.strip(), name.strip()
|
||
if not raw:
|
||
return None
|
||
dt = _parse_ts(raw) or _parse_ts(raw + "T23:59:59+08:00")
|
||
if not dt:
|
||
return None
|
||
return (name or "主線", dt)
|
||
|
||
|
||
def human(delta: timedelta) -> str:
|
||
mins = int(abs(delta).total_seconds()) // 60
|
||
h, m = divmod(mins, 60)
|
||
if h and m:
|
||
return "%d 小時 %d 分" % (h, m)
|
||
if h:
|
||
return "%d 小時" % h
|
||
return "%d 分" % m
|
||
|
||
|
||
def line(session_id: str, transcript_path: str) -> str:
|
||
now = _now()
|
||
start = session_start(session_id, transcript_path, now)
|
||
dl = deadline(now)
|
||
|
||
parts = ["%s 已過 %s" % (MARKER, human(now - start))]
|
||
left = dl - now
|
||
if left.total_seconds() >= 0:
|
||
parts.append("距收工線（台北 %s）剩 %s" % (dl.strftime("%H:%M"), human(left)))
|
||
else:
|
||
parts.append("🔴 已超過收工線（台北 %s）%s" % (dl.strftime("%H:%M"), human(left)))
|
||
|
||
ms = milestone()
|
||
if ms:
|
||
name, due = ms
|
||
if due is None:
|
||
parts.append("主線 %s（沒設期限）" % name)
|
||
else:
|
||
gap = due - now
|
||
if gap.total_seconds() >= 0:
|
||
parts.append("主線 %s 剩 %s" % (name, human(gap)))
|
||
else:
|
||
parts.append("🔴 主線 %s 已逾期 %s" % (name, human(gap)))
|
||
|
||
return "｜".join(parts)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
payload = json.load(sys.stdin)
|
||
except Exception:
|
||
payload = {}
|
||
print(line(payload.get("session_id") or "", payload.get("transcript_path") or ""))
|