Claude Code
156d339308
2026-08-28 第一版漏了這一格，而它剛好就在實測時咬到： 閘判定 pass、Telegram 卻 404（實例上找不到 notify_leo 工作流）， 退路留言貼上票之後**只看得到 pass**——真正的斷點一個字都沒進票。 ⇒ 下一個人會去修閘，而斷點根本不在那裡。 把留言內文抽成 fallback_body()（獨立成一支就是為了測得到）， 測試從「看原始碼有沒有那個字串」改成**真的產一份留言出來檢查**： 斷點原因、原文、身份欄三格都要在。36 條全綠。
430 lines
19 KiB
Python
Executable File
430 lines
19 KiB
Python
Executable File
#!/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 fallback_body(text, gate_fact, why=None):
|
||
"""退路留言的內文。**獨立成一支就是為了測得到**——這段的內容才是這條退路的價值。
|
||
|
||
🔴 **為什麼發不出去**一定要在裡面，不是只寫閘的判定。
|
||
2026-08-28 第一版漏了這一格：閘明明 pass，票上卻只看得到 pass，
|
||
而真正的斷點（實例上找不到 `notify_leo` 工作流）**一個字都沒進票**
|
||
⇒ 下一個人會去修閘，而斷點根本不在那裡。
|
||
"""
|
||
return "\n".join([
|
||
IDENTITY,
|
||
"",
|
||
"🔴 **這則本來要用 Telegram 發給 leo，但這個 session 發不出去**——所以貼在這裡。",
|
||
"",
|
||
"**送不出去的原因**：%s" % (why or "（沒有記錄，這本身就是個 bug）"),
|
||
"",
|
||
gate_report(gate_fact),
|
||
"",
|
||
"---",
|
||
"",
|
||
text,
|
||
])
|
||
|
||
|
||
def comment_on_ticket(ref, text, gate_fact, why=None):
|
||
"""把那則發不出去的訊息貼回票上。回 (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|%s" % (ref, text, why or "")).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 = fallback_body(text, gate_fact, why)
|
||
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, why)
|
||
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:]))
|