#!/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 ""))