diff --git a/dsh-server-docs/scripts/bash-output-guard.py b/dsh-server-docs/scripts/bash-output-guard.py index eea74e3..d0daa63 100644 --- a/dsh-server-docs/scripts/bash-output-guard.py +++ b/dsh-server-docs/scripts/bash-output-guard.py @@ -42,12 +42,14 @@ WS_FALLBACK = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__ # core=True = 几乎必然巨大、误拦率≈0 ⇒ **默认就拦** # core=False = 经常确实需要全量 ⇒ **仅 hard 模式拦**(默认放行) RULES = [ - (True, r'\bcat\s+[^|>;]*\.(log|jsonl|json|ndjson|csv|txt)\b', - '`cat` 大文本文件', '改用 `head -30 文件` / `sed -n \'1,30p\' 文件` / `wc -l 文件`'), + # cat:**按目标文件大小判定**(小文件不拦)—— 这是防日常误伤的关键,见 cat_bigfile() + (True, '__CAT_BIGFILE__', + '`cat` 大文件', '改用 `head -30 文件` / `sed -n \'1,30p\' 文件` / `wc -l 文件`'), (True, r'\bgrep\b[^|;]*-[a-zA-Z]*[rR]', '递归 grep', '改用 `grep -rn … | head -30`,或 `grep -rc …`(只要计数)'), - (True, r'\bls\b[^|;]*-[a-zA-Z]*[lR]', - '`ls -l/-R` 全量列表', '改用 `ls -la 目录 | head -20`,或 `ls 目录 | wc -l`'), + # ⚠️ 只拦 -R(递归全树);**不拦 `ls -la`**(含 l 会误伤日常操作 —— 2026-09-15 自查发现) + (True, r'\bls\b[^|;]*-[a-zA-Z]*R', + '`ls -R` 递归全树', '改用 `ls -la 目录 | head -20`,或 `ls 目录 | wc -l`'), (True, r'\bfind\s+(/[A-Za-z]|[A-Za-z]:)', '`find` 从盘符/根起全树扫', '改用 `find 具体目录 -maxdepth 3 … | head -20`'), (True, r'\bjournalctl\b(?!.*(-n\s*\d|head))', @@ -102,12 +104,45 @@ def log(root, detail): pass +CAT_BIG = 200 * 1024 # `cat` 目标文件 > 200 KB 才值得拦(小文件放行 ⇒ 防日常误伤) + + +def cat_bigfile(cmd): + """`cat <文件>` 且目标文件 > CAT_BIG ⇒ True。取不到大小(相对路径/不存在/管道源)⇒ False(放行)。""" + m = re.search(r'(?:^|[|;&]\s*)cat\s+(?:-[A-Za-z]+\s+)*([^\s|>;&]+)', cmd) + if not m: + return False + p = m.group(1).strip('"\'') + if p in ('-', '/dev/null', '/dev/stdin'): + return False + try: + return os.path.getsize(p) > CAT_BIG + except OSError: + return False + + +def segments(cmd): + """按 shell 分隔符切段(`;` `&&` `||` `|` 换行)—— 只对**每段的开头**做匹配。 + + ⚠️ 为什么不匹配整条命令文本(2026-09-15 实测教训):那样**引号里的字符串也会被拦** —— + 例如 `printf '...ls -laR...'`、测试脚本、把命令写进文档,全都会误拦(阻塞面过大)。 + 按段匹配既保住真拦(`cd x && ls -laR` 的第二段以 `ls` 开头 ⇒ 照样拦),又不误伤"只是提到"。 + """ + return [s.strip() for s in re.split(r'(?:&&|\|\||;|\n|\|)', cmd) if s.strip()] + + def reason_for(cmd, hard=False): + segs = segments(cmd) for core, pat, why, fix in RULES: if not core and not hard: # loose 条只在 hard 模式生效 continue - if re.search(pat, cmd): - return why, fix + for s in segs: + if pat == '__CAT_BIGFILE__': + if cat_bigfile(s): + return why, fix + continue + if re.match(pat, s): # ← match(段首),不是 search(全串) + return why, fix return None, None @@ -157,7 +192,7 @@ def main(): 'permissionDecision': 'deny', 'permissionDecisionReason': ( '💰 拦下:**%s** —— 这类命令的输出会**永久留在会话上下文里、每轮全量重发**' - '**永久留在会话上下文里、每轮全量重发**』(实测某会话 553 轮 × 平均 42 万 token = 2.33 亿 input,' + '(实测某会话 553 轮 × 平均 42 万 token = 2.33 亿 input,' 'output 仅占 0.35%%)。\n' '✅ 换成限流写法再发:%s\n' 'ℹ️ 若确实需要全量:**先落盘再只读关键行**(`… > /tmp/x.txt 2>&1` 然后 `sed -n \'1,40p\' /tmp/x.txt`);' diff --git a/dsh-server-docs/scripts/stop-dialog-guard.py b/dsh-server-docs/scripts/stop-dialog-guard.py index 087f5d7..987a1fa 100644 --- a/dsh-server-docs/scripts/stop-dialog-guard.py +++ b/dsh-server-docs/scripts/stop-dialog-guard.py @@ -308,6 +308,37 @@ def budget_note(tokens, ncalls, prev=None): ) +def guard_health(root): + """**自动发现并上报**:读 `bash-guard.log`,同一规则重复命中 ≥3 次 ⇒ 提示"可能误伤"。 + + 为什么要有:门禁自己不会喊疼 —— 只有"写日志"没人看。让每轮都跑的脚本顺带体检, + 机制问题才能在**下一次用户发言时**浮出来(而不是等人发现"AI 怎么老做不成事")。 + """ + try: + p = os.path.join(root, '.workbuddy', 'bash-guard.log') + rows = [l for l in io.open(p, encoding='utf-8').read().split('\n') if 'DENY|' in l][-40:] + except OSError: + return '' + if len(rows) < 3: + return '' + cnt = {} + for l in rows: + try: + k = l.split('DENY|')[1].split('|')[0].strip() + except IndexError: + continue + cnt[k] = cnt.get(k, 0) + 1 + if not cnt: + return '' + why, n = max(cnt.items(), key=lambda kv: kv[1]) + if n < 3: + return '' + return ('⚠️【门禁自检】最近 %d 次 Bash 里有 **%d 次**因「%s」被拦(同一规则重复命中)⇒ 先怀疑**误伤**,' + '不是谁的操作有问题。处置:① 换等价限流写法;② 若确认误伤 ⇒ 把 `off` 写进 ' + '`.workbuddy/bash-guard-mode`(或 env `DSH_OUTPUT_GUARD_OFF=1`),并**主动上报用户**' + '(规则该不该收窄是人的决定)。' % (len(rows), n, why)) + + def mode_of(root): try: m = io.open(os.path.join(root or '.', '.workbuddy', 'stop-guard-mode'), encoding='utf-8').read() @@ -332,15 +363,15 @@ def user_prompt_mode(payload): hit = bool(tail) and not RE_QUOTE.search(tail) and bool(RE_BAN.search(tail)) toks, ncalls, prev = session_budget(tp) note = budget_note(toks, ncalls, prev) - log(root0 or '.', 'invoked(user-prompt)|mode=%s|上轮收尾=征询句:%s|上下文=%s tok(+%s)|工具=%s 次|预算告警=%s|%s' - % (mode, hit, toks, (toks - prev) if (toks and prev) else '-', ncalls, bool(note), + gh = guard_health(root0 or '.') + log(root0 or '.', 'invoked(user-prompt)|mode=%s|上轮收尾=征询句:%s|上下文=%s tok(+%s)|工具=%s 次|预算告警=%s|门禁自检=%s|%s' + % (mode, hit, toks, (toks - prev) if (toks and prev) else '-', ncalls, bool(note), bool(gh), (tail.replace('\n', ' ')[:60] if tail else '(取不到上一轮文本)'))) - if mode == 'inject' and (hit or note): + if mode == 'inject' and (hit or note or gh): ctx = '' - if hit: - ctx += CONTEXT - if note: - ctx += ('\n\n' + note) if ctx else note + for seg in (CONTEXT if hit else '', note, gh): + if seg: + ctx += ('\n\n' + seg) if ctx else seg _emit({'hookSpecificOutput': {'hookEventName': 'UserPromptSubmit', 'additionalContext': ctx}})