194 lines
8.0 KiB
Python
194 lines
8.0 KiB
Python
#!/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
|
||
|
||
SCOPE = 'aliyun-dsh-server' # 只对本工作区生效
|
||
LOG_REL = os.path.join('.workbuddy', 'skill-load-guard.log')
|
||
STATE_REL = os.path.join('.workbuddy', '.skill-load-guard.state')
|
||
DISABLE_REL = os.path.join('.workbuddy', 'skill-guard.disabled')
|
||
COOLDOWN = 300 # 同会话注入冷却（秒）
|
||
|
||
# 用户「点名方法」的词表 —— 命中任一即注入。
|
||
# 「决策方法」是最稳的一条（覆盖"参考/按/用决策方法"）。
|
||
TRIGGERS = (
|
||
'决策方法',
|
||
'自行决策', '自主决策', '自己决策', '自己拿主意',
|
||
'别问我', '不要问我', '不用问我', '按你的规划', '按你的判断',
|
||
'参考决策', '决策方法论',
|
||
)
|
||
|
||
|
||
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/scripts/
|
||
try:
|
||
return os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..'))
|
||
except Exception:
|
||
return ''
|
||
|
||
|
||
def _log(workdir, line):
|
||
try:
|
||
p = os.path.join(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 = _workdir(obj)
|
||
if not workdir or SCOPE not in 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()
|
||
|
||
ctx = (
|
||
'【技能加载闸门 · 机制层强制】检测到用户本轮**点名了方法**（命中：%s）。\n'
|
||
'原文片段：「…%s…」\n\n'
|
||
'⛔ **不要凭记忆代替、也不要以"我已经知道判据"为由跳过**：\n'
|
||
' **先调用 Skill 工具加载 `dsh-decision-method`**（决策方法论 · 含 §4.5 规则冲突裁决顺序 + 上抛前三问），\n'
|
||
' 然后才开始作答。若本轮涉及功能需求判类型 / 该不该上抛，**同时加载 `dsh-feature-first`**。\n\n'
|
||
'为什么强制：2026-09-16 实测（会话 ddea70b7）—— 用户点名「参考决策方法」后，'
|
||
'AI 全程 `Skill` 调用 **0 次**，仍按旧判据把两个**非门禁**问题上抛给用户，'
|
||
'用户 45 / 58 秒后自己给出了答案。**规则写在文件里 ≠ 会在正确的时机被取用。**\n'
|
||
'（若判定本轮确实与决策无关，可在作答中一句话说明后继续 —— 但**不要静默跳过加载**。）'
|
||
) % ('、'.join(hits), excerpt)
|
||
|
||
_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)
|