dsh_shenxian/dsh-server-docs/scripts/stop-dialog-guard.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""stop-dialog-guard.py —— 「禁止用征询句收尾」的 Stop 钩子（WorkBuddy / CodeBuddy）

为什么需要它
────────────
2026-09-15 实测：本工作区日志里 `tool=AskUserQuestion` 调用数 = 09-12: 43 / 09-13: 3 / **09-14: 0 / 09-15: 0**
⇒ 既有「提问闸门」（PreToolUse + matcher ^AskUserQuestion$）**拦的是几乎不走的工具面**，
而真实的上抛（"要我接着做吗 / 请确认 / 说一声即可"）发生在**正文里** —— 没有任何机制覆盖。

本钩子 = 覆盖那条面：**每次回复结束时**读 transcript 的**最后一条 assistant 文本**，
只扫**收尾段**（最后两行有效内容）里的征询句式；命中 → 返回 `{"continue": false, "reason": …}`
让 Agent **继续一轮并自我纠正**（把该自己做的事做掉，或改写成「需要你拍板」一节）。

安全设计（都不许省）
────────────────────
1. **自作用域**：只在 `transcript_path` 落在本工作区（`aliyun-dsh-server`）时生效，其他项目一律放行。
2. **防死循环**：输入里的 `stop_hook_active == true` 时**不再阻拦**（官方语义：本次停止已由 stop hook 触发过）。
3. **绝不添乱**：任何异常 → 静默放行（exit 0）。判定只在**收尾段**做，避免正文引用规则时误伤。
4. **性能**：只读转录**末尾 256 KB**（实测整库最大转录 31.9 MB、全文读 14 MB ≈ 832 ms ⇒ 不可接受），只看 stdin + 该文件。
5. **防跑飞**：同一会话 600 秒内最多拦**一次**。
6. **急停双闸**（无需卸载/重启）：env `DSH_STOP_GUARD_OFF=1`，或新建 `<工作区>/.workbuddy/stop-guard.disabled`。
7. **低频自证日志**：命中才写一行（`<工作区>/.workbuddy/stop-dialog-guard.log`），用来回答"到底有没有触发"。

退出码：始终 0；决策通过 stdout 的 JSON 表达。
安装（settings.json 的 hooks 段 · 见档案 73 / 99）：
  "Stop": [{ "hooks": [{ "type": "command",
      "command": "\"<python>\" \"<此脚本>\"", "timeout": 10 }] }]
⚠️ hooks 是**应用启动时快照** ⇒ 装完必须**完全重启 WorkBuddy**；桌面版无 /hooks 面板，等效。
⛔ **安装命令不要给本脚本加 `-E`（或任何会屏蔽 PYTHONUTF8 的 flag）**：本机环境本就设了
   `PYTHONUTF8=1` / `PYTHONIOENCODING=utf-8`，而 `-E` 会把它们**全部忽略** ⇒ stdin 回退 **cp936** ⇒
   含中文的 payload 解码即炸。本脚本现已改为走 `buffer` 显式 UTF-8（读写都加固），但**不要靠加固兜底**，
   装的时候也别再引入新雷。（2026-09-15 实测：`-S -E` 曾让本钩子"看起来从未被调用"整整一天。）
"""
import io
import json
import os
import re
import sys
import time

SCOPE = 'aliyun-dsh-server'          # 只对本工作区生效
LOG_REL = os.path.join('.workbuddy', 'stop-dialog-guard.log')
# 兜底工作区：用于"每次调用必留痕"（万一宿主没给 cwd、也没设 CODEBUDDY_PROJECT_DIR）
# 本脚本位于 <工作区>/dsh-server-docs/scripts/ ⇒ 往上三级即工作区
WS_FALLBACK = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# 只扫「收尾段」：出现这些就是"把该自己做的事甩回给用户"
PATTERNS = [
    r'要我(再|接着|继续|现在)?[^。！？\n]{0,20}吗',
    r'要不要我[^。！？\n]{0,20}',
    r'是否要我[^。！？\n]{0,20}',
    r'需要我[^。！？\n]{0,20}吗',
    r'请确认[^。！？\n]{0,16}',
    r'要不要(继续|现在做|我来)[^。！？\n]{0,20}',
    r'是否(继续|需要我)[^。！？\n]{0,20}',
    r'说一声即可',
    r'你看(怎么办|怎么弄|要不要)',
    r'你(决定|拍板)一下',
]
RE_BAN = re.compile('|'.join(PATTERNS))

REASON = (
    '⛔ 收尾句是**征询句**，但按本平台规则（`CODEBUDDY.md §1`「回话前自检」+ `dsh-feature-first §5.3` 铁律 3）'
    '先重判三问：① 命中**真门禁**吗（不可逆破坏性操作 / 边界外六类）？没命中 → **删掉这句，自己做完，改成陈述句**（"我接着做 X"）；'
    '② 是不是在把已经定下来的事再问一遍？是 → 删；③ 这件事用户有客观可判的优劣吗？没有 → 才允许问，且**一轮只问这一句**，'
    '并写进 `dsh-feature-first §5.1` 结论骨架的「**需要你拍板**」一节 —— 该节必须是**整条回复的最后一节**、'
    '且**逐条编号**（有序段落）（2026-09-15 用户明令：「放在最后，别隐藏在回复内容中间」「按照有序段落展示」），'
    '用**陈述句**列"各候选的**优点 / 缺点** + 我的倾向"，不要用征询句。'
    '⚠️ 上抛前先过**取舍筛** —— 某个候选**只有优点 / 只有缺点** ⇒ **自己拍掉、不要问**；'
    '且候选**竖排成段**（A / B / C 各占一行），⛔ 不横排、不做成表格的列（2026-09-15 用户明令）。'
)


TAIL_BYTES = 262144
MAX_BYTES = 4194304      # 扩窗上限 4 MB（防"巨行"时无限读）          # 只读末尾 256 KB（实测：整库最大转录 31.9 MB；全文读 14 MB = 832 ms/轮，不可接受）


def transcribe_last_assistant(path):
    """返回最后一条 assistant 文本（**从尾部向后分块读**；读不到返回 ''）。

    ⚠️ 为什么不是"一次读末尾 256 KB"：一条 assistant 记录可能本身就 > 256 KB
    （长回复 / 被回显的工具输出），此时尾窗会切在 JSON 行中间 ⇒ `json.loads` 失败 ⇒ **静默漏判**。
    做法：从尾部按 TAIL_BYTES 递增扩窗（上限 MAX_BYTES），**直到至少解析出一条 assistant 记录**。
    常见情形（小消息）只花一次 256 KB 读，成本可忽略。
    """
    try:
        size = os.path.getsize(path)
    except OSError:
        return ''
    with io.open(path, 'rb') as f:
        window = TAIL_BYTES
        while True:
            start = max(0, size - window)
            f.seek(start)
            raw = f.read().decode('utf-8', 'replace')
            lines = raw.split('\n')
            if start > 0:
                lines = lines[1:]                    # 丢弃被截断的首行
            for line in reversed(lines):
                line = line.strip()
                if not line:
                    continue
                try:
                    rec = json.loads(line)
                except ValueError:
                    continue
                if rec.get('type') != 'message' or rec.get('role') != 'assistant':
                    continue
                chunks = [c['text'] for c in (rec.get('content') or [])
                          if isinstance(c, dict) and isinstance(c.get('text'), str)]
                chunks += [c for c in (rec.get('content') or []) if isinstance(c, str)]
                if chunks:
                    return '\n'.join(chunks)
            if start == 0 or window >= MAX_BYTES:
                # 放行，但**留痕**（A18：静默失败是负债）——可能是一条 >MAX_BYTES 的巨型记录
                try:
                    os.environ.setdefault('_DSH_SG_MISS', '1')
                    r0 = os.environ.get('CODEBUDDY_PROJECT_DIR') or os.environ.get('DSH_WORKSPACE') or ''
                    if r0:
                        log(r0, '未能解析（窗口 %d 字节仍无 assistant 记录）' % window)
                except Exception:
                    pass
                return ''
            window = min(window * 4, MAX_BYTES)


def tail_lines(text, n=2):
    out = [l.strip() for l in text.strip().split('\n') if l.strip()]
    return '\n'.join(out[-n:])


# 转述/引用豁免：收尾行里带引号或"引用/规则/写着/禁"等词 ⇒ 是在复述规则，不是在问用户
RE_QUOTE = re.compile(r'[「」“”"\']|引用|规则|写着|禁')


RATE_WINDOW = 600            # 秒；同一会话两次「阻止停止」的最小间隔


def _rate_limited(root, sid, peek=False):
    """同一会话 RATE_WINDOW 秒内已拦过 ⇒ 本次直接放行（防连续多轮被拦）。"""
    if not root or not sid:
        return False
    p = os.path.join(root, '.workbuddy', 'cache', 'stop-guard-fires.json')
    try:
        d = json.loads(io.open(p, encoding='utf-8').read()) if os.path.exists(p) else {}
    except Exception:
        d = {}
    now = time.time()
    if now - float(d.get(sid, 0) or 0) < RATE_WINDOW:
        return True
    d = {k: v for k, v in d.items() if now - float(v or 0) < 86400}   # 只留 1 天
    d[sid] = now
    try:
        os.makedirs(os.path.dirname(p), exist_ok=True)
        io.open(p, 'w', encoding='utf-8', newline='\n').write(json.dumps(d))
    except Exception:
        pass
    return False


def log(root, detail):
    try:
        p = os.path.join(root, LOG_REL)
        os.makedirs(os.path.dirname(p), exist_ok=True)
        with io.open(p, 'a', encoding='utf-8') as f:
            f.write('%s\t%s\n' % (time.strftime('%Y-%m-%d %H:%M:%S'), detail))
        lines = io.open(p, encoding='utf-8').read().split('\n')      # 上限 300 行，超出截半（防膨胀）
        if len(lines) > 300:
            io.open(p, 'w', encoding='utf-8', newline='\n').write('\n'.join(lines[-150:]))
    except Exception:
        pass


def _entry_log(payload, raw_len):
    """⚠️ **每次被调用必留痕**（含"payload 解析失败 / 未进作用域 / 被急停"三种静默情形）。

    2026-09-15 教训：原实现只在**通过全部守卫之后**才写日志 ⇒ 日志缺失时**无法区分**
    「宿主根本没调用」与「调用了但被静默 return」—— 而这两者的处置**完全相反**
    （前者要卸载、后者要放宽作用域判据）。凡"要判有没有被调用"的探针，必须**入口即留痕**。
    """
    try:
        p = payload if isinstance(payload, dict) else {}
        tp = str(p.get('transcript_path') or '')
        root = (os.environ.get('CODEBUDDY_PROJECT_DIR') or os.environ.get('DSH_WORKSPACE')
                or p.get('cwd') or WS_FALLBACK)
        log(str(root), 'entry｜event=%s｜cwd=%s｜in_scope=%s｜tp=%s｜keys=%s｜stdin_len=%s'
            % (p.get('hook_event_name') or '(parse-fail)', p.get('cwd') or '-',
               SCOPE in tp, (tp[-80:] if tp else '-'),
               (','.join(sorted(p.keys()))[:120] or '-'), raw_len))
    except Exception:
        pass


def _read_stdin_text():
    """**显式按 UTF-8 读 stdin** —— 不要用 `sys.stdin.read()`。

    ⚠️ 2026-09-15 实测定位：本脚本的安装形态是 `python -S -E <脚本>`，而 **`-E` 会忽略
    `PYTHONUTF8=1` / `PYTHONIOENCODING=utf-8`** ⇒ `sys.stdin.encoding` 回退成 **cp936**；
    钩子 payload 里**必然含中文**（用户的提示词）⇒ 文本模式读取抛
    `UnicodeDecodeError: 'gbk' codec can't decode byte 0x80` ⇒ **钩子静默不生效、日志为空**，
    表象却是"宿主好像没调用钩子"（实为本地炸在解码上，白排查一轮）。
    读 `buffer` 即与 flag / locale 完全无关。
    """
    try:
        return sys.stdin.buffer.read().decode('utf-8', 'replace')
    except Exception:
        try:
            return sys.stdin.read()
        except Exception:
            return ''


def _emit(obj):
    """**显式按 UTF-8 写 stdout**（同理：cp936 下 `ensure_ascii=False` 的中文 / `⛔` 会 UnicodeEncodeError）。"""
    data = json.dumps(obj, ensure_ascii=False).encode('utf-8')
    try:
        sys.stdout.buffer.write(data)
        sys.stdout.buffer.flush()
    except Exception:                 # 极端兜底：退回文本写（可能丢非 GBK 字符，但不至于静默不输出）
        try:
            sys.stdout.write(data.decode('utf-8', 'replace'))
            sys.stdout.flush()
        except Exception:
            pass


# ─────────────────────────────────────────────────────────────
# 第二方案：`UserPromptSubmit`（2026-09-15 加）
# 背景：本版 WorkBuddy **不调用 `Stop` 钩子**（实测：留痕已开、探针句已验证会命中、日志仍空）⇒ 改用
#       `UserPromptSubmit`（输入同样带 `transcript_path`，且能通过 `additionalContext` 注入上下文）。
#   两级模式（**改一个文本文件即可切换，无需重启** —— 脚本内容每次调用现读）：
#     probe  ：只写日志（零风险、可判定"有没有被调用"）
#     inject ：若**上一轮回复的收尾是征询句** ⇒ 注入一段上下文，让下一轮自我纠正
#   模式文件：<工作区>/.workbuddy/stop-guard-mode （内容含 "inject" 即切到 inject，否则 probe）
CONTEXT = (
    '⛔ 【上一轮收尾自检】你上一条回复的**最后一行是征询句**（"要我…吗 / 要不要我 / 请确认 / 说一声即可"类），'
    '这属于本平台**被禁的形态**（`CODEBUDDY.md §1`「回话前自检」）。本轮的处置：'
    '① 若那件事本来就该你自己拍 —— **直接做完**，用陈述句交代；'
    '② 若确实命中真门禁（不可逆破坏性操作 / 边界外六类）—— 写进 `dsh-feature-first §5.1` 结论骨架的'
    '「**需要你拍板**」一节，该节必须是**整条回复的最后一节**、**逐条编号**，且**每个候选写明优点 / 缺点**、**候选竖排成段**（A / B / C 各占一行，⛔ 不横排、不做成表格的列）（陈述句，不要用征询句）；'
    '⚠️ 若某候选**只有优点或只有缺点** ⇒ **那不该问**，自己拍掉；'
    '③ 顺带按红线 **R11** 复核：这个改动有没有让项目某一维度**净变差**。'
)


def mode_of(root):
    try:
        m = io.open(os.path.join(root or '.', '.workbuddy', 'stop-guard-mode'), encoding='utf-8').read()
    except OSError:
        m = ''
    return 'inject' if 'inject' in m else 'probe'


def user_prompt_mode(payload):
    """UserPromptSubmit：probe=只记日志；inject=命中则注入上下文（不阻断提示词）。"""
    tp = str(payload.get('transcript_path') or '')
    if SCOPE not in tp:
        return
    if os.environ.get('DSH_STOP_GUARD_OFF'):
        return
    root0 = os.environ.get('CODEBUDDY_PROJECT_DIR') or os.environ.get('DSH_WORKSPACE') or ''
    if root0 and os.path.exists(os.path.join(root0, '.workbuddy', 'stop-guard.disabled')):
        return
    mode = mode_of(root0)
    text = transcribe_last_assistant(tp)
    tail = tail_lines(text, 1) if text else ''
    hit = bool(tail) and not RE_QUOTE.search(tail) and bool(RE_BAN.search(tail))
    log(root0 or '.', 'invoked(user-prompt)｜mode=%s｜上轮收尾=征询句:%s｜%s'
        % (mode, hit, (tail.replace('\n', ' ')[:60] if tail else '(取不到上一轮文本)')))
    if hit and mode == 'inject':
        _emit({'hookSpecificOutput': {'hookEventName': 'UserPromptSubmit',
                                      'additionalContext': CONTEXT}})


def main():
    raw = _read_stdin_text()          # ⚠️ 必须走 buffer：`-E` 下 sys.stdin 是 cp936（见 _read_stdin_text 注释）
    payload = None
    if raw.strip():
        try:
            payload = json.loads(raw)
        except ValueError:
            payload = None
    _entry_log(payload, len(raw))     # ⚠️ 先留痕，再判作用域（否则"没被调用"与"静默失配"分不开）
    if payload is None:
        return
    if (payload.get('hook_event_name') or '') == 'UserPromptSubmit':   # 第二方案分派
        return user_prompt_mode(payload)
    tp = str(payload.get('transcript_path') or '')
    if SCOPE not in tp:                       # 作用域外 → 放行
        return
    if os.environ.get('DSH_STOP_GUARD_OFF'):  # 急停（环境变量）→ 放行
        return
    root0 = os.environ.get('CODEBUDDY_PROJECT_DIR') or os.environ.get('DSH_WORKSPACE') or ''
    if root0 and os.path.exists(os.path.join(root0, '.workbuddy', 'stop-guard.disabled')):
        return                                # 急停（闸刀文件）→ 放行
    if payload.get('stop_hook_active'):       # 防死循环 → 放行
        return
    text = transcribe_last_assistant(tp)
    if not text:
        return
    sid = str(payload.get('session_id') or '')
    if os.environ.get('DSH_SG_DEBUG'):        # 调试：每次调用都留痕（用于验证宿主是否真的调用本钩子）
        log(root0 or '.', 'invoked｜scope=%s｜tail_active=%s' % (SCOPE in tp, bool(payload.get('stop_hook_active'))))
    if _rate_limited(root0, sid, peek=True):   # 只查不记账
        return
    if os.environ.get('DSH_SG_LOG_ALL', '1') != '0':   # ★本工作区内**每次调用都留痕**（可判定"有没有被调用"）
        log(root0 or '.', 'invoked｜tail_active=%s' % bool(payload.get('stop_hook_active')))
    tail = tail_lines(text, 1)          # 只看**最后一行**：命中面越窄，误报越少
    if RE_QUOTE.search(tail):            # 复述/引用规则 → 不是收尾提问
        return
    m = RE_BAN.search(tail)
    if not m:
        return
    root = (os.environ.get('CODEBUDDY_PROJECT_DIR') or os.environ.get('DSH_WORKSPACE') or '')
    _rate_limited(root0, sid)                  # 命中才记账（同一会话 10 分钟最多拦 1 次）
    log(root if os.path.isdir(root) else '.', 'stop-dialog-guard 命中：%s ｜ 收尾：%s'
        % (m.group(0), tail.replace('\n', ' ')[:80]))
    _emit({'continue': False, 'reason': REASON})


if __name__ == '__main__':
    try:
        main()
    except Exception:
        # ⚠️ 钩子绝不能因自身故障干扰会话 ⇒ 仍放行，但**必须留痕**（A18：静默失败是负债；
        #    2026-09-15 实证：本文件的 `except: pass` 曾把 `NameError: out is not defined` 藏住半小时）
        try:
            import traceback
            _r = os.environ.get('CODEBUDDY_PROJECT_DIR') or os.environ.get('DSH_WORKSPACE') or '.'
            log(_r, 'EXCEPTION｜%s' % traceback.format_exc().strip().split('\n')[-1][:120])
        except Exception:
            pass
    sys.exit(0)