dsh_shenxian/dsh-server-docs/07-scripts/skill-load-guard.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""skill-load-guard.py —— 「用户点名了方法 ⇒ 强制加载技能」的 UserPromptSubmit 钩子

为什么需要它
────────────
2026-09-16 复盘会话 `ddea70b7`（「确认guest用户数据迁移到106服务器」）：
用户 U6 明确说「D1–D6 先做哪些 **按照你的规划执行**，中间有问题**参考决策方法**」，
但该会话 265 次工具调用里 **`Skill` 计数 = 0** —— 那份方法论**从未进入上下文**。
AI 仍按常驻判据把两个**非门禁**问题（"106 旧控制面要不要停" / "D1–D6 先做哪些"）上抛给用户，
而用户随后 45 秒 / 58 秒内自己给出了答案（Q1 的答案还超出了 AI 给的选项空间）。

⇒ 与 `stop-dialog-guard.py` 同族问题：**规则写了，但没有任何机制保证它在正确的时机被取用**。
   （那个钩子处理"回复收尾时的征询句"，本钩子处理"用户点名时的方法加载"，两者互补。）

本钩子 = 覆盖"用户点名"这条面：**每次用户提交时**扫输入，命中点名关键词 ⇒
通过 `additionalContext` 注入一条**紧邻用户消息**的强制指令
（位置比静态规则文件显著得多 —— 长会话里 CODEBUDDY.md 的硬要求容易被"稀释"）。

安全设计（都不许省）
────────────────────
1. **自作用域**：只在 cwd 落在本工作区（`aliyun-dsh-server`）时生效，其他项目一律放行。
2. **绝不添乱**：任何异常 → 静默放行（exit 0）；本钩子**从不拦用户**，只在命中时**追加一段提醒**。
3. **防跑飞**：同一会话 300 秒内最多注入一次。
4. **急停双闸**（无需卸载/重启）：env `DSH_SKILL_GUARD_OFF=1`，或 `<工作区>/.workbuddy/skill-guard.disabled`。
5. **低频自证日志**：命中才写一行到 `<工作区>/.workbuddy/skill-load-guard.log` —— 用来回答"到底有没有触发"。
6. **显式 UTF-8**：读写都走 `buffer`，不依赖 `PYTHONUTF8`（`-E` 会屏蔽它 ⇒ cp936 ⇒ 含中文静默炸）。

退出码：始终 0；决策通过 stdout 的 JSON 表达。
安装（`settings.json` 的 hooks 段 · 与既有钩子同族，**并列挂在同一个 UserPromptSubmit 的 hooks 数组里**）：
  "UserPromptSubmit": [{ "hooks": [
      { "type": "command", "command": "\"<python>\" \"<此脚本>\"", "timeout": 10 },
      { "type": "command", "command": "\"<python>\" -S -E \"<stop-dialog-guard.py>\"", "timeout": 10 }
  ]}]
⚠️ hooks 是**应用启动时快照** ⇒ 装完必须**完全重启 WorkBuddy**（关窗 ≠ 退出）才加载。
"""
import io
import json
import os
import re
import sys
import time

# ── 作用域（2026-09-22 由单值扩为**多值 + env 可覆盖**）─────────────────────────
# 起因：用户 2026-09-22 拍板「B」—— 让本钩子覆盖本机全部会话区，而不是只管 aliyun-dsh-server。
# 形态刻意与 `dsh-ai1net-desktop/.workbuddy/guard/sync-scoped-guards.py` 生成的副本**同形**
#   （`_scopes()` + `_in_scope()`），这样两侧可互换、副本的机械变换规则不再需要改写本文件。
# ⛔ 不写死绝对路径：只比工作区**目录名**，换机器 / 改盘符都不受影响。
# ⚠️ env `DSH_GUARD_SCOPES`（逗号分隔）可覆盖；设为空串 ⇒ 只走默认表。
_SCOPES_DEFAULT = (
    'aliyun-dsh-server',        # 平台主工作区（本脚本所在的项目）
    'dsh-ai1net-desktop',       # 桌面壳 / 开源导出线
)
_log_rel = os.path.join('.workbuddy', 'skill-load-guard.log')


def _workdirs():
    """作用域标记元组（env `DSH_GUARD_SCOPES` 优先，逗号分隔）。"""
    raw = os.environ.get('DSH_GUARD_SCOPES')
    if raw is None:
        return _SCOPES_DEFAULT
    return tuple(s.strip() for s in raw.split(',') if s.strip())


def _in_scope(s):
    """目录名匹配即算命中（与副本同形；比绝对路径稳，不受盘符/用户名影响）。"""
    return any(x in str(s or '') for x in _workdirs())


LOG_REL = _log_rel
STATE_REL = os.path.join('.workbuddy', '.skill-load-guard.state')
DISABLE_REL = os.path.join('.workbuddy', 'skill-guard.disabled')
COOLDOWN = 300                                   # 同会话注入冷却（秒）

# 用户「点名方法 / 点名规则」的词表 —— 命中任一即注入。
# ── 分组 1：点名「决策方法」（2026-09-16 立，最稳的一条）─────────────────────
# ── 分组 2：点名「通用作业规则」（2026-09-22 用户拍板 B 后新增）─────────────────
#    目的：`agent-operating-rules`（跨工作区通用规则技能）过去**完全靠 description 匹配**，
#    是所有技能里最容易被漏的一个 —— 加三条窄词给它一条机制通道。
TRIGGERS = (
    # 分组 1 —— 决策方法族（→ 加载 `dsh-decision-method`）
    '决策方法',
    '自行决策', '自主决策', '自己决策', '自己拿主意',
    '别问我', '不要问我', '不用问我', '按你的规划', '按你的判断',
    '参考决策', '决策方法论',
    # 分组 2 —— 作业规则族（→ 加载 `agent-operating-rules`）
    '按规则来', '按规则做', '按作业规则', '遵守规则', '按规矩来',
)

# 命中词 → 该加载哪个技能（2026-09-22 加：从"只会推 dsh-decision-method"扩为按命中词分流）
_LOAD_RULES = ('按规则来', '按规则做', '按作业规则', '遵守规则', '按规矩来')


def _read_stdin():
    try:
        raw = sys.stdin.buffer.read()
    except Exception:
        return {}
    if not raw:
        return {}
    for enc in ('utf-8', 'utf-8-sig', 'gbk'):
        try:
            return json.loads(raw.decode(enc))
        except Exception:
            continue
    return {}


def _emit(obj):
    try:
        sys.stdout.buffer.write(json.dumps(obj, ensure_ascii=False).encode('utf-8'))
        sys.stdout.buffer.flush()
    except Exception:
        pass


def _workdir(obj):
    for k in ('cwd', 'project_dir', 'workspace', 'projectDir'):
        v = obj.get(k)
        if isinstance(v, str) and v:
            return v
    for k in ('CODEBUDDY_PROJECT_DIR', 'WORKSPACE_FOLDER', 'PWD'):
        v = os.environ.get(k)
        if v:
            return v
    # 兜底：本脚本位于 <工作区>/dsh-server-docs/07-scripts/
    try:
        return os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..'))
    except Exception:
        return ''


def _norm_path(p):
    """把 MSYS / Git-Bash 风格路径规范成 Windows 风格（`/e/foo` → `E:/foo`）。

    动机（2026-09-22 实测，**正在持续发生**）：**Windows 原生 python 会把
    `/e/ProgramData/x` 解释成「当前盘根 + 相对路径」= `e\\ProgramData\\x`**
    ⇒ 若当前盘是 E，就落到 `E:\\e\\ProgramData\\x`。
    实证：E 盘根长出影子目录 `E:\\e\\ProgramData\\AI技能\\<ws>\\.workbuddy\\`（355 文件 / 29 MB，仍在增长）。

    ⛔ 同源铁律：本工作区已有「**Python exe 不认 `/e/…` ⇒ 传 `E:/…`**」。
    """
    try:
        s = str(p or '')
        m = re.match(r'^/([A-Za-z])(/.*)?$', s)
        if m:
            return m.group(1).upper() + ':' + (m.group(2) or '/')
    except Exception:
        pass
    return p


def _log(workdir, line):
    try:
        p = os.path.join(_norm_path(workdir), LOG_REL)
        with io.open(p, 'a', encoding='utf-8', newline='\n') as f:
            f.write('%s %s\n' % (time.strftime('%Y-%m-%d %H:%M:%S'), line))
    except Exception:
        pass


def _cooling(workdir, sid):
    """同一会话 COOLDOWN 秒内不重复注入。"""
    try:
        p = os.path.join(workdir, STATE_REL)
        now = time.time()
        last_sid, last_ts = '', 0.0
        if os.path.exists(p):
            try:
                with io.open(p, 'r', encoding='utf-8') as f:
                    parts = f.read().strip().split('\t')
                    last_sid, last_ts = parts[0], float(parts[1] or 0)
            except Exception:
                pass
        if last_sid == sid and (now - last_ts) < COOLDOWN:
            return True
        with io.open(p, 'w', encoding='utf-8', newline='\n') as f:
            f.write('%s\t%.3f\n' % (sid, now))
    except Exception:
        pass
    return False


def main():
    if os.environ.get('DSH_SKILL_GUARD_OFF') == '1':
        return
    obj = _read_stdin()
    if not obj:
        return

    workdir = _norm_path(_workdir(obj))
    if not workdir or not _in_scope(workdir):
        return                                   # 自作用域：表内工作区之外的会话一律放行
    if os.path.exists(os.path.join(workdir, DISABLE_REL)):
        return

    prompt = ''
    for k in ('prompt', 'user_prompt', 'userPrompt', 'message', 'text'):
        v = obj.get(k)
        if isinstance(v, str) and v.strip():
            prompt = v
            break

    sid = str(obj.get('session_id') or obj.get('sessionId') or 'unknown')
    # ★ 入口即留痕（2026-09-16 加）—— 否则"没命中就没日志"，永远无法回答
    #   "它到底有没有被宿主调用"。与 stop-dialog-guard.py 同款设计；每次用户提交一行。
    _log(workdir, 'entry｜event=UserPromptSubmit｜in_scope=True｜sid=%s｜prompt_len=%d'
                 % (sid[:8], len(prompt)))
    if not prompt:
        return

    hits = [w for w in TRIGGERS if w in prompt]
    if not hits:
        return

    if _cooling(workdir, sid):
        return

    seg = re.sub(r'\s+', ' ', prompt)
    pos = min(seg.find(w) for w in hits)
    excerpt = seg[max(0, pos - 40): pos + 80].strip()

    # ── 按命中词分流（2026-09-22）：规则族与决策族给不同指令 ──────────────────
    hit_rules = [w for w in hits if w in _LOAD_RULES]
    if hit_rules:
        head = '检测到用户本轮**点名了作业规则**（命中：%s）。' % '、'.join(hit_rules)
        body = (
            '⛔ **不要凭记忆代替、也不要以"我已经知道规则"为由跳过**：\n'
            '  **先调用 Skill 工具加载 `agent-operating-rules`**（通用作业规则 · 含"什么时候自决 / 什么时候上抛 / 回复排版 / 说话方式 / 不越界 / 多棒接力"），\n'
            '  然后才开始作答。若本轮涉及功能需求判类型 / 该不该上抛，**同时加载 `dsh-feature-first`**。\n\n'
            '为什么强制：该技能是**跨工作区通用**的（不绑定任何具体项目），'
            '但它与其他技能一样**靠 description 匹配按需加载 —— 没有机制保证它在任何工作区都被取用**'
            '（它的 description 里已写「当你在任何一个工作区开始任务…」，仍可能漏）。'
            '本条注入即补上这个缺口。\n'
            '（若判定本轮确实与作业规则无关，可在作答中一句话说明后继续 —— 但**不要静默跳过加载**。）'
        )
    else:
        head = '检测到用户本轮**点名了方法**（命中：%s）。' % '、'.join(hits)
        body = (
            '⛔ **不要凭记忆代替、也不要以"我已经知道判据"为由跳过**：\n'
            '  **先调用 Skill 工具加载 `dsh-decision-method`**（决策方法论 · 含 §4.5 规则冲突裁决顺序 + 上抛前三问），\n'
            '  然后才开始作答。若本轮涉及功能需求判类型 / 该不该上抛，**同时加载 `dsh-feature-first`**。\n\n'
            '为什么强制：2026-09-16 实测（会话 ddea70b7）—— 用户点名「参考决策方法」后，'
            'AI 全程 `Skill` 调用 **0 次**，仍按旧判据把两个**非门禁**问题上抛给用户，'
            '用户 45 / 58 秒后自己给出了答案。**规则写在文件里 ≠ 会在正确的时机被取用。**\n'
            '（若判定本轮确实与决策无关，可在作答中一句话说明后继续 —— 但**不要静默跳过加载**。）'
        )

    ctx = '【技能加载闸门 · 机制层强制】%s\n' '原文片段：「…%s…」\n\n' '%s' % (head, excerpt, body)

    _emit({'hookSpecificOutput': {
        'hookEventName': 'UserPromptSubmit',
        'additionalContext': ctx,
    }})
    _log(workdir, 'HIT sid=%s hits=%s' % (sid[:8], ','.join(hits)))


if __name__ == '__main__':
    try:
        main()
    except Exception:
        pass
    sys.exit(0)