#!/usr/bin/env python3 """isep-notify — 發一則 Telegram 給 leo，而且**發不出去的時候不會安靜** （inkstone/ISEP#93） ━━ 這支存在的理由（2026-08-28 實測出來的限制）━━━━━━━━━━━━━━━━━━━━━━━━ 通道本身早就有（`notify_leo`，一條 curl、不需金鑰，見頂層 wiki `agent-memory.md`）。 問題不在通道，在**閘**： · `prod-write-guard.sh` 擋「打到線上實例的寫入型請求」，而發一則 Telegram 在它眼裡跟「部署一個工作流上線」長得一模一樣（同一個 named webhook 家族路徑） · v0.10.0 已經修好——只放行 `notify_leo` 這**一個名字** · **但 hook 的註冊路徑是 session 啟動當下寫死的** ⇒ 更新 plugin 之後， 要開一個**新** session 才會載到新那份 ⇒ 所以「能不能發得出去」是**每個 session 各自不同的事實**，不是全域設定。 這支的工作就是：**先去問這個 session 的閘，再決定怎麼辦**，而不是先射再說。 🔴 **發不出去而靜默，等於沒做。** 三條出路，一條都不能省： ① 先問閘（`--gate`）：把「這個 session 載到的是哪一版、它會不會擋」變成查得到的事實 ② 閘說會擋 ⇒ **不送**（不繞路，見下），改走退路：貼回票上 ＋ 大聲印在眼前 ③ 送出去了也要驗**內層** `data.data.ok`——外層 200 不算送到（wiki 記過這個坑） ━━ 為什麼閘說會擋就不送，而不是「反正 python 打得出去」━━━━━━━━━━━━━━━ 這支用 urllib 打 HTTP，`prod-write-guard` 掛在 `Bash` 上——**它看不到這支**。 也就是說：不問閘就送，一定送得出去。**而那正是不能做的事。** leo 2026-08-11：「hook 要不讓它對正式環境做任何推送，不是某種推送」 leo（同期）：「人就會學會繞過它，那它就等於不存在」 一支「繞過閘也照送」的工具，會讓那道閘對整條 python 路徑失效。 ⇒ 本支**自己去問那道閘**，閘說不行就不行。這比 shell 那條路更嚴，不是更鬆。 ━━ 用法 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ isep-notify --gate 只印閘的判定（不送、不碰網路） isep-notify --text "…" 送一則 isep-notify --text-file <檔> 內文放檔案（長訊息用） isep-notify --text "…" --fallback-ticket inkstone/ISEP#93 送不出去時貼回那張票 isep-notify --text "…" --dry-run 全流程走一遍，只是不真的送 離開碼：0 ＝ leo 拿得到了（Telegram 送達，或退路留在票上） 1 ＝ 兩條路都沒走通（這時 stderr 一定有一段紅字，不會安靜） 環境變數（都有預設，測試才需要動）： ISEP_NOTIFY_URL notify_leo 的 trigger 網址 ISEP_NOTIFY_IDENTITY 貼票時的【身份】欄 ISEP_NOTIFY_STATE_DIR 閘判定快取的位置（預設 ~/.claude/isep-nag） ISEP_NOTIFY_GUARD 指定要問哪一支 prod-write-guard（測試用） ISEP_NOTIFY_OFFLINE=1 不打任何網路（測試用） """ import hashlib import json import os import re import subprocess import sys import time import urllib.error import urllib.request NOTIFY_URL = os.environ.get( "ISEP_NOTIFY_URL", "https://arcrun-cypher-executor.leo21c.workers.dev" "/webhooks/named/leo/notify_leo/trigger", ) IDENTITY = os.environ.get("ISEP_NOTIFY_IDENTITY", "【身份】總管／inkstone/ISEP／-") STATE_DIR = os.environ.get("ISEP_NOTIFY_STATE_DIR") or os.path.join( os.path.expanduser("~"), ".claude", "isep-nag") GITEA_HOST = os.environ.get("TICKET_HOST") or "https://git.uncle6.me" PROD_STAMP = "/tmp/.prod-write-ok" GATE_FACT = "gate-verdict.json" # ══ 一、這個 session 的閘是哪一份、哪一版 ═══════════════════════════════ # # 🔴 這一段就是工單說的「**把『這個 session 的閘是哪一版』變成它自己查得到的事實**」。 # 三個來源，可信度由高到低——**而且一定把來源講出來**， # 因為「我問到的那份」跟「這個 session 真的註冊的那份」可能不是同一個檔。 def find_guard(): """回 (guard 路徑, plugin 根目錄, 來源說明, 可信度)。找不到回 (None, ...)。""" cands = [] forced = os.environ.get("ISEP_NOTIFY_GUARD") if forced: # 測試用：直接指定要問哪一支（測「舊版會擋」時把歷史版本擺進來） return forced, os.path.dirname(os.path.dirname(forced)), "ISEP_NOTIFY_GUARD 指定", "high" root = os.environ.get("CLAUDE_PLUGIN_ROOT") if root: # hook 裡跑：這就是**這個 session 真的註冊的那一份**，沒有第二種可能 cands.append((root, "CLAUDE_PLUGIN_ROOT（這個 session 註冊的那一份）", "high")) remembered = _read_fact() if remembered and remembered.get("plugin_root"): cands.append((remembered["plugin_root"], "SessionStart 當時記下來的那一份", "high")) here = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) cands.append((here, "本腳本自己所在的那一份（不一定是 session 註冊的）", "low")) for r, why, conf in cands: g = os.path.join(r, "hooks", "prod-write-guard.sh") if os.path.isfile(g): return g, r, why, conf return None, None, "找不到任何 prod-write-guard.sh", "none" def plugin_version(root): try: with open(os.path.join(root, ".claude-plugin", "plugin.json")) as f: return json.load(f).get("version") or "未知" except Exception: return "未知" def probe_gate(url=None): """拿**這一支真的要送的請求**去問閘：你會不會擋？ 回一個 dict，永遠有 verdict： pass 閘看過了，放行 block 閘會擋（＝這個 session 載到的是舊版，或白名單被改窄了） unknown 問不出來（沒有閘、戳記在場、腳本壞了）——**不會謊稱 pass** """ url = url or NOTIFY_URL guard, root, why, conf = find_guard() fact = { "at": int(time.time()), "url": url, "guard": guard, "plugin_root": root, "source": why, "confidence": conf, "version": plugin_version(root) if root else "未知", } if not guard: fact.update(verdict="unknown", reason="這個環境找不到 ISEP 的 prod-write-guard" "（沒有閘 ⇒ 沒有東西被繞過）") return fact # 🔴 戳記在場就**不探針**：`prod-write-guard` 的戳記是**單次、用完即丟**的， # 探一次就把總管剛按下去的那一次授權燒掉了。 # （而戳記在場本來就代表「現在允許寫」⇒ 送得出去，不必問。） if os.path.exists(PROD_STAMP): fact.update(verdict="unknown", reason="/tmp/.prod-write-ok 在場：探針會把它燒掉（單次用完即丟），" "所以不問。戳記在場＝現在允許寫入") return fact payload = json.dumps({ "tool_name": "Bash", "tool_input": { "command": ("curl -s -X POST %s -H 'Content-Type: application/json' " "-d '{\"text\":\"[總管] gate probe\"}'" % url), }, }) try: p = subprocess.run(["bash", guard], input=payload, capture_output=True, text=True, timeout=30) except Exception as e: fact.update(verdict="unknown", reason="問不動那支閘：%s" % e) return fact if p.returncode == 0: fact.update(verdict="pass", reason="閘看過這一則通知，放行") elif p.returncode == 2: fact.update(verdict="block", reason="閘會擋。這個 session 載到的 ISEP 是 v%s——" "它的 prod-write-guard 還認不出 notify_leo 不是部署" % fact["version"], gate_says=(p.stderr or "").strip()[:400]) else: fact.update(verdict="unknown", reason="那支閘回了離開碼 %d（既不是放行也不是擋）" % p.returncode) return fact def _fact_path(): return os.path.join(STATE_DIR, GATE_FACT) def _read_fact(): try: with open(_fact_path()) as f: return json.load(f) except Exception: return None def remember_fact(fact): """把判定寫成檔：之後任何一支腳本（含手動跑的）都查得到，不必自己再探一次。""" try: os.makedirs(STATE_DIR, exist_ok=True) with open(_fact_path(), "w") as f: json.dump(fact, f, ensure_ascii=False, indent=2) except Exception: pass def gate_report(fact): icon = {"pass": "🟢", "block": "🔴", "unknown": "🟡"}.get(fact["verdict"], "🟡") lines = [ "%s 這個 session 的通知閘：**%s**" % (icon, fact["verdict"]), " ISEP 版本：v%s" % fact.get("version", "未知"), " 問的是哪一份：%s" % (fact.get("guard") or "（找不到）"), " 來源可信度：%s（%s）" % (fact.get("confidence"), fact.get("source")), " 理由：%s" % fact.get("reason", ""), ] if fact["verdict"] == "block": lines += [ "", " ⇒ **這個 session 發不出 Telegram**，不是網路問題，是閘的版本問題。", " ⇒ 解法（兩步，缺一不可）：", " claude plugin update isep@inkstone # 拿到 v0.10.0 以上", " **開一個新的 session** # 註冊路徑是啟動當下寫死的", " ⇒ 在那之前，本支一律走退路（貼回票上＋印在眼前），不會安靜。", ] return "\n".join(lines) # ══ 二、真的送 ═══════════════════════════════════════════════════════════ def send_telegram(text): """回 (ok, 說明)。**內層 data.data.ok 才算送到**——外層 200 不算。""" if os.environ.get("ISEP_NOTIFY_OFFLINE") == "1": return False, "ISEP_NOTIFY_OFFLINE=1：刻意不打網路" body = json.dumps({"text": text}).encode() # 🔴 一定要帶 User-Agent（2026-08-28 實撞）：不帶的話 urllib 送出的是 # `Python-urllib/3.x`，**Cloudflare 直接回 403 error code 1010**（bad user agent）。 # 症狀非常會騙人——wiki 說「一條 curl、不需任何金鑰」是對的， # curl 自己帶 UA，所以同一個網址用 curl 通、用 python 不通 # ⇒ 會被誤讀成「通道壞了」或「這個 session 沒權限」，而兩個都不是。 req = urllib.request.Request( NOTIFY_URL, data=body, method="POST", headers={"Content-Type": "application/json", "User-Agent": "isep-notify/1.0 (+inkstone/ISEP#93)"}) try: raw = urllib.request.urlopen(req, timeout=30).read().decode() except urllib.error.HTTPError as e: return False, "HTTP %s：%s" % (e.code, e.read().decode()[:200]) except Exception as e: return False, "連不上：%s" % e try: d = json.loads(raw) except Exception: return False, "回應不是 JSON：%s" % raw[:200] # 🔴 wiki 記過這個坑：外層 success=true、內層 404 也發生過 inner = ((d.get("data") or {}).get("data") or {}) if inner.get("ok") is True: return True, "送到了（message_id=%s）" % inner.get("message_id") if "找不到 workflow" in raw or "not found" in raw.lower(): # 🔴 這一種要分開講：**閘沒擋、網路也通，是那台實例上根本沒有這支工作流** # （wiki `agent-memory.md` 記過一次：2026-08-10 KV 整批換新時定義消失）。 # 講成「發不出去」會讓人去修錯的地方——修閘、修 token，而斷點不在那裡。 return False, ("通道本身斷了：實例上找不到 `notify_leo` 工作流。" "**這不是閘擋的、也不是網路問題**——要有人把它重新 push 上去" "（`mira` repo `workflows/notify-leo.yaml`）。原文：%s" % raw[:200]) return False, "外層通了但**內層沒 ok**：%s" % raw[:300] # ══ 三、退路：貼回票上 ═══════════════════════════════════════════════════ def gitea_token(): for env in ("GITEA_TOKEN_CLAUDE_CODE", "GITEA_TOKEN"): v = os.environ.get(env) if v: return v root = os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd() host = GITEA_HOST.split("//")[-1].rstrip("/") try: out = subprocess.run(["git", "-C", root, "remote", "-v"], capture_output=True, text=True, timeout=20).stdout except Exception: out = "" for line in out.splitlines(): if host in line: m = re.search(r"//[^:/]+:([^@]+)@", line) if m: return m.group(1) return None def comment_on_ticket(ref, text, gate_fact): """把那則發不出去的訊息貼回票上。回 (ok, 說明)。 🔴 **同一則不重複貼**：內容雜湊存在狀態目錄，24 小時內一樣就跳過。 票被洗版的下場跟發不出去一樣——leo 會學會不看它。 """ m = re.match(r"^([\w.-]+)/([\w.-]+)#(\d+)$", (ref or "").strip()) if not m: return False, "票的寫法是 owner/repo#N，你給的是：%s" % ref owner, repo, num = m.group(1), m.group(2), m.group(3) key = hashlib.sha256(("%s|%s" % (ref, text)).encode()).hexdigest()[:16] seen = os.path.join(STATE_DIR, "commented-%s" % key) if os.path.exists(seen) and time.time() - os.path.getmtime(seen) < 24 * 3600: return True, "24 小時內已經貼過一模一樣的內容，跳過（不洗版）" if os.environ.get("ISEP_NOTIFY_OFFLINE") == "1": return False, "ISEP_NOTIFY_OFFLINE=1：刻意不打網路" tok = gitea_token() if not tok: return False, "拿不到 Gitea token（GITEA_TOKEN_CLAUDE_CODE 沒設，remote 也沒帶憑證）" body = "\n".join([ IDENTITY, "", "🔴 **這則本來要用 Telegram 發給 leo，但這個 session 發不出去**——所以貼在這裡。", "", gate_report(gate_fact), "", "---", "", text, ]) url = "%s/api/v1/repos/%s/%s/issues/%s/comments" % (GITEA_HOST, owner, repo, num) req = urllib.request.Request( url, data=json.dumps({"body": body}).encode(), method="POST", headers={"Authorization": "token %s" % tok, "Content-Type": "application/json"}) try: r = json.load(urllib.request.urlopen(req, timeout=40)) except urllib.error.HTTPError as e: return False, "Gitea %s：%s" % (e.code, e.read().decode()[:200]) except Exception as e: return False, "連不上 Gitea：%s" % e try: os.makedirs(STATE_DIR, exist_ok=True) open(seen, "w").close() except Exception: pass return True, "已貼到 %s（%s）" % (ref, r.get("html_url") or "") # ══ 四、CLI ══════════════════════════════════════════════════════════════ def notify(text, fallback_ticket=None, dry_run=False): """回 (離開碼, 給人看的報告)。**任何一條路都會留下一段話，不會安靜。**""" fact = probe_gate() remember_fact(fact) out = [gate_report(fact), ""] if dry_run: out.append("🧪 --dry-run：到此為止，什麼都沒送。要送的內容是：") out.append("") out.append(text) return 0, "\n".join(out) if fact["verdict"] == "block": out.append("⛔ 閘說會擋 ⇒ **本支不繞路**，直接走退路。") sent, why = False, "沒有嘗試（閘會擋，繞過去等於拆了那道閘）" else: sent, why = send_telegram(text) out.append(("✅ Telegram：%s" if sent else "❌ Telegram：%s") % why) if sent: return 0, "\n".join(out) # ── 退路 ────────────────────────────────────────────────────────── if fallback_ticket: ok, detail = comment_on_ticket(fallback_ticket, text, fact) out.append(("✅ 退路（貼回票上）：%s" if ok else "❌ 退路（貼回票上）：%s") % detail) else: ok, detail = False, "沒有給 --fallback-ticket，沒有票可以貼" out.append("⚠️ 退路（貼回票上）：%s" % detail) out += [ "", "🔴 **leo 的手機上沒有出現這則。** 原文照抄在下面，" "看到這段的人有義務把它講出去：", "", text, ] return (0 if ok else 1), "\n".join(out) def main(argv): text, ticket, dry, want_json = None, None, False, False mode = "notify" i = 0 while i < len(argv): a = argv[i] if a == "--gate": mode = "gate" elif a == "--text": i += 1; text = argv[i] elif a == "--text-file": i += 1 with open(argv[i]) as f: text = f.read() elif a == "--fallback-ticket": i += 1; ticket = argv[i] elif a == "--dry-run": dry = True elif a == "--json": want_json = True elif a in ("-h", "--help"): print(__doc__) return 0 else: print("不認得的參數：%s（--help 看用法）" % a, file=sys.stderr) return 2 i += 1 if mode == "gate": fact = probe_gate() remember_fact(fact) print(json.dumps(fact, ensure_ascii=False, indent=2) if want_json else gate_report(fact)) return 0 if not text or not text.strip(): print("🔴 沒有內容可以發（--text 或 --text-file）", file=sys.stderr) return 2 if not text.lstrip().startswith("["): text = "[總管] " + text.lstrip() # 署名鐵律：text 開頭必署名 code, report = notify(text, ticket, dry) print(report, file=(sys.stdout if code == 0 else sys.stderr)) return code if __name__ == "__main__": sys.exit(main(sys.argv[1:]))