第 26 章 | Stop hook——通知集成

第 26 章：Stop hook——通知集成

学习目标

让 main Claude 完成一段输出后主动通知你（Telegram / Slack / Email）——你能离开电脑也不丢进度。

Stop hook 触发时机

sequenceDiagram
    participant U as 用户
    participant CC as Claude Code
    participant MC as main Claude
    participant Hook as Stop hook
    participant TG as Telegram

    U->>MC: /dev
    MC->>MC: dispatch 各 agent
    MC->>U: 一段输出（含 ⚠️ marker）
    Note over MC: turn 结束
    CC->>Hook: 触发 Stop hook
(传 transcript_path)
    Hook->>Hook: 读 transcript
找 ⚠️ 行
    Hook->>TG: POST /sendMessage
    TG-->>U: 📱 推送

入参 JSON

{
  "session_id": "...",
  "transcript_path": "/path/to/transcript.jsonl",
  "stop_hook_active": true
}

→ 关键是 transcript_path——可以读完整对话历史。

Marker 模式

我们的 dev.md 故意输出有固定前缀的行：

⚠️ Group 7: escalating to developer-deep ...
✓ READY TO ARCHIVE — run /opsx:archive
✗ Group 3 stopped — manual decision required

→ Hook 用正则 grep 这些行，只发重要事件，不发普通进度。

notify-escalation.py 完整实例

#!/usr/bin/env python3
import json, os, re, sys, urllib.request
from pathlib import Path

PROJECT_ROOT = Path(os.environ.get("CLAUDE_PROJECT_DIR")).resolve()

NOTIFY_PATTERNS = [
    re.compile(r"^⚠️.*$", re.MULTILINE),
    re.compile(r"^✓ READY TO ARCHIVE.*$", re.MULTILINE),
    re.compile(r"^✗ Group \d+ stopped.*$", re.MULTILINE),
]

SENT_LOG = PROJECT_ROOT / ".claude" / ".notify-sent"
CONFIG_FILE = PROJECT_ROOT / ".claude" / "telegram-notify.json"


def load_config():
    # 优先 env
    token = os.environ.get("TELEGRAM_BOT_TOKEN")
    chat_id = os.environ.get("TELEGRAM_CHAT_ID")
    if token and chat_id:
        return token, chat_id
    # 退到 config 文件
    if CONFIG_FILE.exists():
        data = json.loads(CONFIG_FILE.read_text())
        return data.get("bot_token"), str(data.get("chat_id"))
    return None


def extract_markers(transcript_path):
    found = []
    for line in Path(transcript_path).open():
        try:
            msg = json.loads(line)
        except: continue
        content = msg.get("message", {}).get("content")
        if isinstance(content, list):
            text = "\n".join(b.get("text","") for b in content
                             if isinstance(b, dict) and b.get("type") == "text")
        else:
            continue
        for pat in NOTIFY_PATTERNS:
            for m in pat.findall(text):
                found.append(m.strip())
    return found


def send_telegram(token, chat_id, text):
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    payload = json.dumps({"chat_id": chat_id, "text": text}).encode()
    req = urllib.request.Request(url, data=payload,
                                  headers={"Content-Type": "application/json"})
    try:
        urllib.request.urlopen(req, timeout=5)
        return True
    except Exception as e:
        print(f"Telegram failed: {e}", file=sys.stderr)
        return False


data = json.load(sys.stdin)
config = load_config()
if not config:
    sys.exit(0)  # 没配 = 静默退出

markers = extract_markers(data.get("transcript_path", ""))
already = set(SENT_LOG.read_text().splitlines()) if SENT_LOG.exists() else set()
new = [m for m in markers if m not in already]
if not new:
    sys.exit(0)

if send_telegram(config[0], config[1], "\n".join(new[-5:])):
    with SENT_LOG.open("a") as f:
        for m in new:
            f.write(m + "\n")

sys.exit(0)

去重很重要

不去重:  每次 turn 都把所有历史 marker 重发
         你的手机一直振
         
去重 (.notify-sent):
  保存已发的 marker 行
  下次只发新增的

完整通知链：

sequenceDiagram
    participant MC as main Claude
    participant CC as Claude Code
    participant Hook as notify-escalation.py
    participant Trans as transcript.jsonl
    participant Sent as .notify-sent
    participant TG as Telegram API
    participant You as 你

    MC->>CC: 输出 "⚠️ Group 7: STUCK..."
    Note over CC: turn 结束触发 Stop hook
    CC->>Hook: stdin: {transcript_path: ...}
    Hook->>Trans: 读 JSONL
    Hook->>Hook: 正则匹配 ⚠️/✓/✗ 行
得到 all_markers
    Hook->>Sent: 读已发送列表
    Hook->>Hook: new = all_markers - sent
    alt 有新事件
        Hook->>TG: POST /sendMessage
{chat_id, text: new}
        TG-->>You: 📱 推送
        Hook->>Sent: 追加 new 行
        Hook-->>CC: exit 0
    else 无新事件
        Hook-->>CC: exit 0 (静默)
    end
    Note over CC: 主流程不受影响

多通道扩展

# Slack webhook
def send_slack(webhook_url, text):
    urllib.request.urlopen(...)

# Email
import smtplib
def send_email(to, subject, body):
    ...

# Discord
def send_discord(webhook_url, text):
    ...

→ marker 提取逻辑不变，只换发送函数。

安全考虑

✅ token 放 settings.local.json（gitignore）
✅ token 放 env 变量（不入文件）
❌ token 写死在脚本里（push 即泄露）
❌ chat_id 在团队共享文件（隐私）

Stop hook 失败怎么办

hook exit 非 0 → Claude Code 显示警告但**不阻塞主流程**
                  你的 turn 该完成的还是完成了
                  只是没收到通知

→ Stop hook 失败不会破坏开发。最坏情况是你错过通知。

反模式

❌ marker 设得太宽（每行都触发通知）
   → 通知轰炸

❌ 不去重
   → 同一事件反复推

❌ token 写在脚本
   → 一上 git 就泄露

❌ hook 不能 5 秒内完成
   → 拖累每次 Stop

❌ hook 自己改源代码 / spec
   → 越界，会破坏状态机

你现在能做什么

• 写 Stop hook 接通 Telegram / Slack / 任意 HTTP 通道
• 用 marker 模式只发重要事件
• 实现去重避免轰炸