564 lines
26 KiB
Python
564 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
def load_js_process_helpers() -> str:
|
|
patcher_path = Path(__file__).resolve().parents[1] / 'fixes' / 'patch_openclaw_weixin_hermes.py'
|
|
assert patcher_path.exists(), f'missing patcher: {patcher_path}'
|
|
spec = importlib.util.spec_from_file_location('patch_openclaw_weixin_hermes', patcher_path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module.JS_PROCESS_HELPERS
|
|
|
|
|
|
|
|
def load_patcher_module() -> object:
|
|
patcher_path = Path(__file__).resolve().parents[1] / 'fixes' / 'patch_openclaw_weixin_hermes.py'
|
|
assert patcher_path.exists(), f'missing patcher: {patcher_path}'
|
|
spec = importlib.util.spec_from_file_location('patch_openclaw_weixin_hermes', patcher_path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
|
|
def run_js_process_helpers(tmp_path: Path, body: str) -> dict:
|
|
js_process_helpers = load_js_process_helpers()
|
|
node_prelude = '\n'.join(
|
|
[
|
|
'const hermesTestFs = require("node:fs");',
|
|
'globalThis.existsSync = hermesTestFs.existsSync;',
|
|
'globalThis.readFileSync = hermesTestFs.readFileSync;',
|
|
'globalThis.writeFileSync = hermesTestFs.writeFileSync;',
|
|
'globalThis.mkdirSync = hermesTestFs.mkdirSync;',
|
|
'globalThis.appendFileSync = hermesTestFs.appendFileSync;',
|
|
'globalThis.path = require("node:path");',
|
|
'globalThis.logger = { info() {}, warn() {}, error() {} };',
|
|
]
|
|
)
|
|
node_script = tmp_path / 'hermes_image_followup_check.js'
|
|
node_script.write_text(node_prelude + '\n\n' + js_process_helpers + '\n\n' + body, encoding='utf-8')
|
|
completed = subprocess.run(['node', str(node_script)], check=True, capture_output=True, text=True)
|
|
return json.loads(completed.stdout.strip())
|
|
|
|
|
|
@pytest.mark.parametrize('is_ts', [True, False], ids=['ts', 'js'])
|
|
def test_replace_or_insert_process_helpers_is_byte_idempotent(is_ts: bool) -> None:
|
|
patcher = load_patcher_module()
|
|
marker = 'export async function processOneMessage'
|
|
source = marker + '(message) {\n return message;\n}\n'
|
|
|
|
once = patcher.replace_or_insert_process_helpers(source, is_ts=is_ts)
|
|
twice = patcher.replace_or_insert_process_helpers(once, is_ts=is_ts)
|
|
helpers = patcher.TS_PROCESS_HELPERS if is_ts else patcher.JS_PROCESS_HELPERS
|
|
|
|
assert helpers + '\n\n' + marker in once
|
|
assert twice == once
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
('is_ts', 'hook_name'),
|
|
[(True, 'TS_PROCESS_HOOK'), (False, 'JS_PROCESS_HOOK')],
|
|
ids=['ts', 'js'],
|
|
)
|
|
def test_remove_and_reinsert_process_hook_is_byte_idempotent(is_ts: bool, hook_name: str) -> None:
|
|
patcher = load_patcher_module()
|
|
hook = getattr(patcher, hook_name)
|
|
indent = ' ' if is_ts else ' '
|
|
needle = indent + 'ctx.CommandAuthorized = commandAuthorized;\n'
|
|
next_nonblank = indent + 'continueProcessing();\n'
|
|
|
|
def cycle(text: str) -> str:
|
|
without_hook = patcher.remove_existing_process_hermes_hook(text)
|
|
return without_hook.replace(needle, needle + hook + '\n', 1)
|
|
|
|
once = cycle(needle + next_nonblank)
|
|
twice = cycle(once)
|
|
|
|
assert hook + '\n' + next_nonblank in once
|
|
assert twice == once
|
|
|
|
|
|
|
|
def test_looks_like_hermes_image_followup_distinguishes_diagnostics_from_real_image_questions(tmp_path: Path) -> None:
|
|
cases = {
|
|
'diag_cache_expired': ('\u56fe\u7247\u7f13\u5b58\u5df2\u5931\u6548\uff0c\u8bf7\u91cd\u65b0\u53d1\u9001\u56fe\u7247\u3002', False),
|
|
'diag_cache_error_en': ('image cache error', False),
|
|
'diag_cache_expired_en': ('the image cache expired, please resend', False),
|
|
'diag_copying_text_shows_cache_error': ('\u4e3a\u4ec0\u4e48\u6211\u590d\u5236\u53d1\u7ed9\u4f60\u7684\u6587\u5b57\uff0c\u4f60\u90fd\u663e\u793a\u56fe\u7247\u7f13\u5b58\u5df2\u5931\u6548\uff0c\u8bf7\u91cd\u65b0\u53d1\u9001\u56fe\u7247\u3002', False),
|
|
'diag_not_image': ('\u8fd9\u4e0d\u662f\u56fe\u7247\uff0c\u662f\u6587\u5b57\u6d88\u606f', False),
|
|
'diag_text_not_image_en': ('this is a text message, not an image', False),
|
|
'diag_no_image_sent': ('\u6211\u6ca1\u6709\u53d1\u56fe\u7247', False),
|
|
'diag_system_logs': ('\u8bf7\u68c0\u67e5\u7cfb\u7edf\u65e5\u5fd7\uff0c\u4ece\u673a\u5236\u4e0a\u4fee\u590d\u9519\u8bef', False),
|
|
'image_cn_1': ('\u8bf7\u770b\u4e00\u4e0b\u8fd9\u5f20\u56fe\u7247', True),
|
|
'image_cn_2': ('\u8fd9\u56fe\u662f\u4ec0\u4e48', True),
|
|
'image_cn_3': ('\u8bc6\u522b\u56fe\u7247\u5185\u5bb9', True),
|
|
'image_en': ('look at this image', True),
|
|
}
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'const cases = '
|
|
+ json.dumps({key: text for key, (text, _) in cases.items()}, ensure_ascii=False)
|
|
+ ';\nconst results = {};\n'
|
|
+ 'for (const [key, text] of Object.entries(cases)) {\n'
|
|
+ ' results[key] = looksLikeHermesImageFollowup(text);\n'
|
|
+ '}\n'
|
|
+ 'console.log(JSON.stringify(results));\n',
|
|
)
|
|
for key, (_, expected) in cases.items():
|
|
assert results[key] is expected, f'{key}: expected {expected}, got {results[key]!r}'
|
|
|
|
|
|
|
|
def test_resolve_hermes_referenced_image_followup_requires_exact_cache(tmp_path: Path) -> None:
|
|
cache_path = tmp_path / 'hermes-last-image-cache.json'
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
'const fs = require("fs");',
|
|
f'const cachePath = {json.dumps(str(cache_path))};',
|
|
'const accountId = "account-a";',
|
|
'const senderId = "sender-b";',
|
|
'const exactQuotedId = "quoted-123";',
|
|
'const cache = {};',
|
|
'cache[hermesLastImageCacheKey(accountId, senderId)] = {',
|
|
' mediaUrl: "data:image/png;base64,iVBORw0KGgo=",',
|
|
' timestampMs: Date.now(),',
|
|
' messageId: "latest-999",',
|
|
' mediaType: "image/png",',
|
|
'};',
|
|
'cache[hermesExactImageCacheKey(accountId, senderId, exactQuotedId)] = {',
|
|
' mediaUrl: "data:image/png;base64,iVBORw0KGgo=",',
|
|
' timestampMs: Date.now(),',
|
|
' messageId: exactQuotedId,',
|
|
' mediaType: "image/png",',
|
|
'};',
|
|
'fs.writeFileSync(cachePath, JSON.stringify(cache, null, 2), "utf-8");',
|
|
'process.env.AIOS_HERMES_WEIXIN_LAST_IMAGE_CACHE = cachePath;',
|
|
'const generic = resolveHermesReferencedImageFollowup(accountId, senderId, "plain quoted text", "quoted-999");',
|
|
'const exact = resolveHermesReferencedImageFollowup(accountId, senderId, "plain quoted text", exactQuotedId);',
|
|
'const forced = resolveHermesFollowupImageContext(accountId, senderId, "plain quoted text", exactQuotedId, false, true);',
|
|
'console.log(JSON.stringify({ generic, exact, forced }));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results['generic']['imageIntent'] is False
|
|
assert results['generic']['matched'] is False
|
|
assert results['generic']['reason'] == 'referenced_image_not_confirmed'
|
|
assert results['exact']['imageIntent'] is True
|
|
assert results['exact']['matched'] is True
|
|
assert results['exact']['reason'] == 'cached_image_matched'
|
|
assert results['forced']['imageIntent'] is True
|
|
assert results['forced']['matched'] is True
|
|
|
|
|
|
|
|
def test_referenced_text_without_media_evidence_stays_out_of_image_cache(tmp_path: Path) -> None:
|
|
cache_path = tmp_path / 'hermes-last-image-cache.json'
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
'const fs = require("fs");',
|
|
f'const cachePath = {json.dumps(str(cache_path))};',
|
|
'const accountId = "account-a";',
|
|
'const senderId = "sender-b";',
|
|
'const cache = {};',
|
|
'cache[hermesLastImageCacheKey(accountId, senderId)] = {',
|
|
' mediaUrl: "data:image/png;base64,iVBORw0KGgo=",',
|
|
' timestampMs: Date.now(),',
|
|
' messageId: "latest-999",',
|
|
' mediaType: "image/png",',
|
|
'};',
|
|
'fs.writeFileSync(cachePath, JSON.stringify(cache, null, 2), "utf-8");',
|
|
'process.env.AIOS_HERMES_WEIXIN_LAST_IMAGE_CACHE = cachePath;',
|
|
'const full = { item_list: [{ ref_msg: { message_item: { type: 0, id: "quoted-999" } } }] };',
|
|
'const hermesContent = "\u8fd9\u662f\u4ec0\u4e48";',
|
|
'const hermesReferenceContextPresent = hasHermesReferenceContext(full.item_list);',
|
|
'const hermesQuotedMessageId = findHermesReferencedMessageId(full.item_list);',
|
|
'const refMediaItem = undefined;',
|
|
'const hermesReferencedMediaIsImage = Boolean(refMediaItem && hermesCurrentMessageIsImage({}, refMediaItem));',
|
|
'const hermesReferencedImageFollowupIntent = hermesReferenceContextPresent && !refMediaItem;',
|
|
'const hermesCurrentImage = false;',
|
|
'const hermesFollowup = hermesCurrentImage',
|
|
' ? { imageIntent: false, matched: false, reason: "current_image_message", cacheStrategy: "current", quotedMessageId: hermesQuotedMessageId }',
|
|
' : (hermesReferencedMediaIsImage',
|
|
' ? resolveHermesFollowupImageContext(accountId, senderId, hermesContent, hermesQuotedMessageId, false, true)',
|
|
' : (hermesReferencedImageFollowupIntent',
|
|
' ? resolveHermesReferencedImageFollowup(accountId, senderId, hermesContent, hermesQuotedMessageId)',
|
|
' : resolveHermesFollowupImageContext(accountId, senderId, hermesContent)));',
|
|
'console.log(JSON.stringify({ hermesReferenceContextPresent, hermesQuotedMessageId, hermesReferencedMediaIsImage, hermesReferencedImageFollowupIntent, hermesFollowup }));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results['hermesReferenceContextPresent'] is True
|
|
assert results['hermesQuotedMessageId'] == 'quoted-999'
|
|
assert results['hermesReferencedMediaIsImage'] is False
|
|
assert results['hermesReferencedImageFollowupIntent'] is True
|
|
assert results['hermesFollowup']['imageIntent'] is False
|
|
assert results['hermesFollowup']['matched'] is False
|
|
assert results['hermesFollowup']['reason'] == 'referenced_image_not_confirmed'
|
|
|
|
|
|
def test_explicit_quoted_text_builds_grounded_plain_text_context(tmp_path: Path) -> None:
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
'const available = typeof resolveHermesQuotedTextContext === "function";',
|
|
'const items = [{ type: 1, ref_msg: { message_item: { type: 1, text_item: { text: "被引用的正文" } } } }];',
|
|
'const result = available ? resolveHermesQuotedTextContext(items, "这是什么意思？") : null;',
|
|
'console.log(JSON.stringify({ available, result }));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results['available'] is True
|
|
assert results['result'] == {
|
|
'referenced': True,
|
|
'handled': False,
|
|
'prompt': '[被引用文字]\n被引用的正文\n[/被引用文字]\n\n[当前问题]\n这是什么意思？\n[/当前问题]',
|
|
'reason': 'quoted_text_resolved',
|
|
}
|
|
|
|
|
|
def test_visible_ref_title_builds_grounded_plain_text_context(tmp_path: Path) -> None:
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
'const items = [{ type: 1, ref_msg: { title: "客户端可见的引用正文", message_item: { type: 0, msg_id: "quoted-123" } } }];',
|
|
'const result = resolveHermesQuotedTextContext(items, "这是什么意思？");',
|
|
'console.log(JSON.stringify(result));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results == {
|
|
'referenced': True,
|
|
'handled': False,
|
|
'prompt': '[被引用文字]\n客户端可见的引用正文\n[/被引用文字]\n\n[当前问题]\n这是什么意思？\n[/当前问题]',
|
|
'reason': 'quoted_text_resolved',
|
|
}
|
|
|
|
|
|
def test_nested_generic_title_is_not_treated_as_quoted_text(tmp_path: Path) -> None:
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
'const items = [{ type: 1, ref_msg: { message_item: { type: 0, title: "非协议引用摘要" } } }];',
|
|
'const result = resolveHermesQuotedTextContext(items, "这是什么意思？");',
|
|
'console.log(JSON.stringify(result));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results == {
|
|
'referenced': True,
|
|
'handled': True,
|
|
'replyText': '无法读取被引用文字，请复制原文后再提问。',
|
|
'reason': 'quoted_text_unavailable',
|
|
}
|
|
|
|
|
|
def test_exact_message_id_cache_resolves_server_reference_without_guessing(tmp_path: Path) -> None:
|
|
cache_path = tmp_path / 'hermes-message-text-cache.json'
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
f'process.env.AIOS_HERMES_WEIXIN_TEXT_MESSAGE_CACHE = {json.dumps(str(cache_path))};',
|
|
'const accountId = "account-a";',
|
|
'const senderId = "sender-b";',
|
|
'const original = { item_list: [{ type: 1, msg_id: "7481729105240418312", text_item: { text: "客户端已经显示的原始正文" } }] };',
|
|
'rememberHermesTextMessage(accountId, senderId, original, "客户端已经显示的原始正文");',
|
|
'const quotedItems = [{ type: 1, ref_msg: { message_item: { type: 0, msg_id: "7481729105240418312" } }, text_item: { text: "这是什么意思？" } }];',
|
|
'const quotedId = findHermesReferencedMessageId(quotedItems);',
|
|
'const exact = resolveHermesCachedQuotedText(accountId, senderId, quotedId);',
|
|
'const wrong = resolveHermesCachedQuotedText(accountId, senderId, "7481729105240418313");',
|
|
'const resolved = resolveHermesQuotedTextContext(quotedItems, "这是什么意思？", exact);',
|
|
'const mode = (hermesTestFs.statSync(process.env.AIOS_HERMES_WEIXIN_TEXT_MESSAGE_CACHE).mode & 0o777).toString(8);',
|
|
'console.log(JSON.stringify({ quotedId, exact, wrong, resolved, mode }));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results['quotedId'] == '7481729105240418312'
|
|
assert results['exact'] == '客户端已经显示的原始正文'
|
|
assert results['wrong'] == ''
|
|
assert results['resolved'] == {
|
|
'referenced': True,
|
|
'handled': False,
|
|
'prompt': '[被引用文字]\n客户端已经显示的原始正文\n[/被引用文字]\n\n[当前问题]\n这是什么意思？\n[/当前问题]',
|
|
'reason': 'quoted_text_resolved',
|
|
}
|
|
assert results['mode'] == '600'
|
|
|
|
|
|
def test_expired_message_id_cache_entry_is_not_resolved(tmp_path: Path) -> None:
|
|
cache_path = tmp_path / 'hermes-message-text-cache.json'
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
f'process.env.AIOS_HERMES_WEIXIN_TEXT_MESSAGE_CACHE = {json.dumps(str(cache_path))};',
|
|
'const key = hermesTextMessageCacheKey("account-a", "sender-b", "quoted-old");',
|
|
'hermesTestFs.writeFileSync(process.env.AIOS_HERMES_WEIXIN_TEXT_MESSAGE_CACHE, JSON.stringify({ [key]: { text: "过期正文", timestampMs: Date.now() - (8 * 24 * 60 * 60 * 1000) } }), { encoding: "utf-8", mode: 0o600 });',
|
|
'const result = resolveHermesCachedQuotedText("account-a", "sender-b", "quoted-old");',
|
|
'console.log(JSON.stringify({ result }));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results['result'] == ''
|
|
|
|
|
|
def test_unsafe_numeric_message_id_is_rejected_before_cache_keying(tmp_path: Path) -> None:
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
'const exactString = extractHermesMessageId({ message_id: "7482038281766381234" });',
|
|
'const unsafeNumber = extractHermesMessageId({ message_id: 7482038281766381234 });',
|
|
'const safeNumber = extractHermesMessageId({ message_id: 12345 });',
|
|
'console.log(JSON.stringify({ exactString, unsafeNumber: unsafeNumber ?? null, safeNumber }));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results == {
|
|
'exactString': '7482038281766381234',
|
|
'unsafeNumber': None,
|
|
'safeNumber': '12345',
|
|
}
|
|
|
|
|
|
def test_exact_outbound_create_time_resolves_quoted_bot_text(tmp_path: Path) -> None:
|
|
cache_path = tmp_path / 'hermes-message-text-cache.json'
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
f'process.env.AIOS_HERMES_WEIXIN_TEXT_MESSAGE_CACHE = {json.dumps(str(cache_path))};',
|
|
'const accountId = "account-a";',
|
|
'const senderId = "sender-b";',
|
|
'const createTimeMs = 1783857900123;',
|
|
'rememberHermesOutboundTextMessage(accountId, senderId, "机器人出站正文", createTimeMs);',
|
|
'const quotedItems = [{ type: 1, ref_msg: { message_item: { type: 0, msg_id: "server-quoted", create_time_ms: createTimeMs } }, text_item: { text: "进展汇报" } }];',
|
|
'const referencedTime = findHermesReferencedCreateTimeMs(quotedItems);',
|
|
'const exact = resolveHermesCachedQuotedText(accountId, senderId, "server-quoted", referencedTime);',
|
|
'const wrong = resolveHermesCachedQuotedText(accountId, senderId, "server-quoted", createTimeMs + 1);',
|
|
'const resolved = resolveHermesQuotedTextContext(quotedItems, "进展汇报", exact);',
|
|
'console.log(JSON.stringify({ referencedTime, exact, wrong, resolved }));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results['referencedTime'] == '1783857900123'
|
|
assert results['exact'] == '机器人出站正文'
|
|
assert results['wrong'] == ''
|
|
assert results['resolved']['reason'] == 'quoted_text_resolved'
|
|
assert '机器人出站正文' in results['resolved']['prompt']
|
|
|
|
|
|
def test_outbound_create_time_collision_fails_closed(tmp_path: Path) -> None:
|
|
cache_path = tmp_path / 'hermes-message-text-cache.json'
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
f'process.env.AIOS_HERMES_WEIXIN_TEXT_MESSAGE_CACHE = {json.dumps(str(cache_path))};',
|
|
'const createTimeMs = 1783857900123;',
|
|
'rememberHermesOutboundTextMessage("account-a", "sender-b", "第一条", createTimeMs);',
|
|
'rememberHermesOutboundTextMessage("account-a", "sender-b", "第二条", createTimeMs);',
|
|
'const resolved = resolveHermesCachedQuotedText("account-a", "sender-b", undefined, createTimeMs);',
|
|
'console.log(JSON.stringify({ resolved }));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results['resolved'] == ''
|
|
|
|
|
|
def test_missing_quoted_text_fails_closed_without_model_prompt(tmp_path: Path) -> None:
|
|
results = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
'const available = typeof resolveHermesQuotedTextContext === "function";',
|
|
'const items = [{ type: 1, ref_msg: { message_item: { type: 0, msg_id: "7481729105240418312" } } }];',
|
|
'const result = available ? resolveHermesQuotedTextContext(items, "这是什么意思？") : null;',
|
|
'let modelCalls = 0;',
|
|
'if (result && !result.handled) modelCalls += 1;',
|
|
'console.log(JSON.stringify({ available, result, modelCalls }));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert results['available'] is True
|
|
assert results['result'] == {
|
|
'referenced': True,
|
|
'handled': True,
|
|
'replyText': '无法读取被引用文字，请复制原文后再提问。',
|
|
'reason': 'quoted_text_unavailable',
|
|
}
|
|
assert results['modelCalls'] == 0
|
|
|
|
|
|
@pytest.mark.parametrize('hook_name', ['TS_PROCESS_HOOK', 'JS_PROCESS_HOOK'])
|
|
def test_process_hook_stops_unavailable_quote_before_model_dispatch(hook_name: str) -> None:
|
|
patcher = load_patcher_module()
|
|
hook = getattr(patcher, hook_name)
|
|
guard_name = 'TS_QUOTED_TEXT_DISPATCH_GUARD' if hook_name.startswith('TS_') else 'JS_QUOTED_TEXT_DISPATCH_GUARD'
|
|
guard = getattr(patcher, guard_name)
|
|
|
|
assert 'rememberHermesTextMessage(deps.accountId, hermesSenderId, full, textBody)' in hook
|
|
assert 'findHermesReferencedCreateTimeMs(full.item_list)' in hook
|
|
assert 'resolveHermesCachedQuotedText(deps.accountId, hermesSenderId, hermesQuotedMessageId, hermesQuotedCreateTimeMs)' in hook
|
|
assert 'resolveHermesQuotedTextContext(full.item_list, hermesContent, hermesCachedQuotedText)' in hook
|
|
assert '!hermesQuotedTextContext.referenced && !textBody.startsWith("/")' in hook
|
|
assert 'ctx.Body = hermesQuotedTextContext.prompt' in hook
|
|
assert 'await dispatchHermesQuotedTextFailClosed(hermesQuotedTextContext, dispatcher)' in guard
|
|
assert 'markDispatchIdle();' in guard
|
|
|
|
|
|
def test_dispatch_quoted_text_fail_closed_uses_dispatcher_without_model(tmp_path: Path) -> None:
|
|
result = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
'(async () => {',
|
|
' const calls = [];',
|
|
' const dispatcher = {',
|
|
' sendFinalReply(payload) { calls.push(["final", payload]); },',
|
|
' markComplete() { calls.push(["complete"]); },',
|
|
' async waitForIdle() { calls.push(["idle"]); },',
|
|
' };',
|
|
' const context = resolveHermesQuotedTextContext([{ ref_msg: { message_item: { type: 0, msg_id: "quoted-1" } } }], "这是什么意思？");',
|
|
' let modelCalls = 0;',
|
|
' const handled = await dispatchHermesQuotedTextFailClosed(context, dispatcher);',
|
|
' if (!handled) modelCalls += 1;',
|
|
' console.log(JSON.stringify({ handled, modelCalls, calls }));',
|
|
'})();',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert result['handled'] is True
|
|
assert result['modelCalls'] == 0
|
|
assert result['calls'] == [
|
|
['final', {'text': '无法读取被引用文字，请复制原文后再提问。'}],
|
|
['complete'],
|
|
['idle'],
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
('is_ts', 'guard_name', 'needle'),
|
|
[
|
|
(True, 'TS_QUOTED_TEXT_DISPATCH_GUARD', ' logger.debug(`dispatchReplyFromConfig: starting agentId=${route.agentId ?? "(none)"}`);\n'),
|
|
(False, 'JS_QUOTED_TEXT_DISPATCH_GUARD', ' logger.debug(`dispatchReplyFromConfig: starting agentId=${route.agentId ?? "(none)"}`);\n'),
|
|
],
|
|
ids=['ts', 'js'],
|
|
)
|
|
def test_dispatch_guard_remove_and_reinsert_is_byte_idempotent(is_ts: bool, guard_name: str, needle: str) -> None:
|
|
patcher = load_patcher_module()
|
|
guard = getattr(patcher, guard_name)
|
|
|
|
def cycle(text: str) -> str:
|
|
without = patcher.remove_existing_process_hermes_hook(text)
|
|
return without.replace(needle, guard + '\n' + needle, 1)
|
|
|
|
once = cycle(needle)
|
|
twice = cycle(once)
|
|
|
|
assert guard + '\n' + needle in once
|
|
assert twice == once
|
|
|
|
|
|
def test_reference_metadata_content_is_not_treated_as_quoted_text(tmp_path: Path) -> None:
|
|
result = run_js_process_helpers(
|
|
tmp_path,
|
|
'\n'.join(
|
|
[
|
|
'const items = [{ ref_msg: { message_item: { type: 0, msg_id: "quoted-1", content: "metadata only" } } }];',
|
|
'console.log(JSON.stringify(resolveHermesQuotedTextContext(items, "这是什么意思？")));',
|
|
]
|
|
),
|
|
)
|
|
|
|
assert result['handled'] is True
|
|
assert result['reason'] == 'quoted_text_unavailable'
|
|
assert result['replyText'] == '无法读取被引用文字，请复制原文后再提问。'
|
|
|
|
|
|
def test_patch_plugin_process_only_skips_slash_and_channel(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
patcher = load_patcher_module()
|
|
calls: list[tuple[str, Path]] = []
|
|
|
|
def record(name: str, result):
|
|
def _inner(root: Path):
|
|
calls.append((name, root))
|
|
return result
|
|
|
|
return _inner
|
|
|
|
monkeypatch.setattr(patcher, 'patch_ts', record('patch_ts', False))
|
|
monkeypatch.setattr(patcher, 'patch_js', record('patch_js', False))
|
|
monkeypatch.setattr(patcher, 'patch_process_messages', record('patch_process_messages', {'src/messaging/process-message.ts': True, 'dist/src/messaging/process-message.js': False}))
|
|
monkeypatch.setattr(patcher, 'patch_channel_files', record('patch_channel_files', {'src/channel.ts': False, 'dist/src/channel.js': False}))
|
|
monkeypatch.setattr(patcher, 'verify_plugin_hook', record('verify_plugin_hook', {
|
|
'hook_status': {},
|
|
'hook_installed': False,
|
|
'process_status': {'src/messaging/process-message.ts': True, 'dist/src/messaging/process-message.js': True},
|
|
'process_hook_installed': True,
|
|
'channel_status': {},
|
|
'channel_hook_installed': False,
|
|
'dispatch_reference_found': False,
|
|
'dispatch_reference_files': [],
|
|
}))
|
|
|
|
result = patcher.patch_plugin(tmp_path, process_only=True)
|
|
|
|
assert calls == [
|
|
('patch_process_messages', tmp_path),
|
|
('verify_plugin_hook', tmp_path),
|
|
]
|
|
assert result['mode'] == 'process-only'
|
|
assert result['plugin_root'] == str(tmp_path)
|
|
assert result['slash_changes'] == {}
|
|
assert result['channel_changes'] == {}
|
|
assert result['process_changes'] == {'src/messaging/process-message.ts': True, 'dist/src/messaging/process-message.js': False}
|
|
assert result['behavior_changes'] == {'src/messaging/process-message.ts': True, 'dist/src/messaging/process-message.js': False}
|
|
assert result['verification']['process_hook_installed'] is True
|
|
|
|
|
|
|
|
def test_parse_args_supports_default_root_and_process_only_flag() -> None:
|
|
patcher = load_patcher_module()
|
|
|
|
defaults = patcher.parse_args([])
|
|
assert defaults.plugin_root == patcher.DEFAULT_PLUGIN_ROOT
|
|
assert defaults.process_only is False
|
|
|
|
process_only = patcher.parse_args(['--process-only'])
|
|
assert process_only.plugin_root == patcher.DEFAULT_PLUGIN_ROOT
|
|
assert process_only.process_only is True
|
|
|
|
custom_root = Path('/tmp/custom-plugin-root')
|
|
custom = patcher.parse_args([str(custom_root), '--process-only'])
|
|
assert custom.plugin_root == custom_root
|
|
assert custom.process_only is True
|