78 lines
3.4 KiB
Python
78 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""R5 预验证：临时装 Stop 钩子 → headless 跑一次（**不重启桌面**）→ 自动卸回 → 出结论。
|
||
|
||
为什么安全：hooks 是**应用启动时快照** ⇒ 正在跑的桌面会话看不到这次改动，
|
||
只有新起的进程（本例 = headless）才会加载它。全程自包含：无论成败都会卸回。
|
||
"""
|
||
import io
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
|
||
WS = r'E:/ProgramData/AIProject/aliyun-dsh-server'
|
||
S = r'E:/ProgramData/.workbuddy/settings.json'
|
||
CLI = r'C:/Users/Administrator/AppData/Local/Programs/WorkBuddy/resources/app.asar.unpacked/cli/dist/codebuddy-headless.js'
|
||
NODE = r'E:/ProgramData/.workbuddy/binaries/node/versions/22.22.2-3/node.exe'
|
||
PY = r'E:/ProgramData/.workbuddy/binaries/python/versions/3.13.12/python.exe'
|
||
SCRIPT = WS + '/dsh-server-docs/scripts/stop-dialog-guard.py'
|
||
LOG = WS + '/.workbuddy/stop-dialog-guard.log'
|
||
ENTRY = {"hooks": [{"type": "command",
|
||
"command": '"%s" -S -E "%s"' % (PY, SCRIPT),
|
||
"timeout": 10}]}
|
||
|
||
|
||
def hooks_state():
|
||
d = json.load(io.open(S, encoding='utf-8'))
|
||
return {k: len(v) for k, v in d.get('hooks', {}).items()}
|
||
|
||
|
||
def main():
|
||
print('[1] 装之前 hooks:', hooks_state(), flush=True)
|
||
bak = S + '.bak-stopverify2-' + time.strftime('%H%M%S')
|
||
shutil.copy2(S, bak)
|
||
d = json.load(io.open(S, encoding='utf-8'))
|
||
d['hooks']['Stop'] = [ENTRY]
|
||
io.open(S, 'w', encoding='utf-8', newline='\n').write(json.dumps(d, ensure_ascii=False, indent=1))
|
||
print('[2] 已临时装 Stop:', hooks_state(), '（备份', os.path.basename(bak), '）', flush=True)
|
||
|
||
if os.path.exists(LOG):
|
||
os.remove(LOG)
|
||
env = dict(os.environ)
|
||
env['DSH_SG_DEBUG'] = '1'
|
||
env['CODEBUDDY_PROJECT_DIR'] = WS
|
||
prompt = '请只用这一句话回复我，不要做任何其他事、不要读文件：要我接着做吗？'
|
||
out, err, rc = '', '', -1
|
||
try:
|
||
r = subprocess.run([NODE, CLI, '-p', prompt, '--output-format', 'stream-json'],
|
||
cwd=WS, env=env, capture_output=True, text=True, timeout=420)
|
||
out, err, rc = r.stdout, r.stderr, r.returncode
|
||
except Exception as e:
|
||
err = 'EXC: %s' % e
|
||
print('[3] headless rc=%s｜stdout %d 字节｜stderr %d 字节' % (rc, len(out), len(err)), flush=True)
|
||
|
||
invoked = os.path.exists(LOG)
|
||
logtxt = io.open(LOG, encoding='utf-8').read() if invoked else ''
|
||
injected = ('征询句' in out) or ('征询句' in err)
|
||
turns = out.count('"assistant"')
|
||
print('[4] 结论：', flush=True)
|
||
print(' 宿主是否调用 Stop 钩子 =', '是（有 debug 日志）' if invoked else '未观测到', flush=True)
|
||
if logtxt:
|
||
print(' debug 日志：', logtxt.strip().split('\n')[0][:180], flush=True)
|
||
print(' 是否采纳 continue:false（reason 被注入）=', '是' if injected else '未观测到', flush=True)
|
||
print(' assistant 出现次数（>1 = 被拦后继续了一轮）=', turns, flush=True)
|
||
if rc != 0:
|
||
print(' stderr 末尾：', (err or '')[-300:].replace('\n', ' '), flush=True)
|
||
|
||
d2 = json.load(io.open(S, encoding='utf-8'))
|
||
if 'Stop' in d2.get('hooks', {}):
|
||
del d2['hooks']['Stop']
|
||
io.open(S, 'w', encoding='utf-8', newline='\n').write(json.dumps(d2, ensure_ascii=False, indent=1))
|
||
print('[5] 已卸回:', hooks_state(), flush=True)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|