231 lines
14 KiB
Python
231 lines
14 KiB
Python
"""Explicit web verification outranks a model's saved-text-only shortcut."""
|
||
import asyncio
|
||
from copy import deepcopy
|
||
from unittest.mock import AsyncMock
|
||
|
||
import pytest
|
||
import httpx
|
||
|
||
from backend.prospecting import dialogue, orchestrator as subject
|
||
from backend.prospecting.contracts import IntentPlan, KNOWLEDGE_ID, ProspectingBrief, TARGET_QUESTION
|
||
from backend.prospecting.prompts import browser_prompt
|
||
from backend.prospecting.request_routing import browser_recheck_requested, ground_execution_plan, recheck_without_analysis
|
||
|
||
|
||
REQUEST = ("先原生历史返回原笔记，只核对已保存子回复d352，必要时定位第一组，"
|
||
"用resolve_reply_target核实际回复对象。不新增来源，不重新搜索或分析名单；无唯一身份结束。")
|
||
|
||
|
||
@pytest.mark.parametrize('text', [REQUEST,
|
||
'不新增来源，但请回到原网页核对已保存回复的直接对象。',
|
||
'请在网页上重新核对这条已保存的回复，不再搜索。',
|
||
'先返回原笔记，只核实已保存的这条子回复。',
|
||
'用 `resolve_reply_target` 核对这条回复实际回复了谁，不整理新名单。',
|
||
'通过原生历史后退，再检查现有来源。',
|
||
'先取消离线整理，返回实际网页核对。',
|
||
'不需要再搜索，但要在浏览器里重新核对回复归属。',
|
||
'请点击该回复对象的真实主页核对身份。',
|
||
])
|
||
def test_actual_web_recheck_is_not_confused_with_additional_collection(text):
|
||
assert browser_recheck_requested(text)
|
||
|
||
|
||
@pytest.mark.parametrize('text', [
|
||
'不需要回网页，仅按已有原话复核。',
|
||
'不要返回原笔记，只整理当前已保存的原话。',
|
||
'不用resolve_reply_target，直接复核现有原话。',
|
||
'无需在网页上重新核对，仅离线按新标准分析。',
|
||
'返回原笔记不用做，按现有原话判断即可。',
|
||
'请回到原笔记核对；取消网页回查，改为离线整理。',
|
||
'“先原生历史返回原笔记，用resolve_reply_target”是引用，按现有原话复核。',
|
||
'「请在网页上核对」只是示例，实际只分析现有材料。',
|
||
"'用resolve_reply_target核对回复对象'是日志，先讨论方案。",
|
||
'> 请返回原笔记核对\n只复核已存文本。',
|
||
'```text\n用resolve_reply_target核实际回复对象\n```\n仅整理现有名单。',
|
||
'例如用resolve_reply_target核对，但本次只分析已有原话。',
|
||
'日志显示返回原笔记，只是描述旧任务。',
|
||
'先解释resolve_reply_target是什么，不运行。',
|
||
'讨论网页回查为何要返回原笔记，先不操作。',
|
||
'按新标准复核已保存原话，不新增搜索。',
|
||
'核对已有回复的父子关系表，不操作网页。',
|
||
'只按原话核对网页来源，不操作网页。',
|
||
'resolve_reply_target是干什么的，我们先聊聊。',
|
||
])
|
||
def test_offline_negated_quoted_and_explanatory_text_does_not_open_browser(text):
|
||
assert not browser_recheck_requested(text)
|
||
|
||
|
||
@pytest.mark.parametrize('text', [
|
||
'现在请对当前页面进行网页核对，只读预检，不保存来源；不调用resolve_reply_target。',
|
||
'请在网页核对现有回复，不用resolve_reply_target，不滚动、不重新导航。',
|
||
'请核对当前网页，不需要回到原笔记，不调用resolve_reply_target。',
|
||
'在当前网页核对，不打开新的网页，不调用resolve_reply_target。',
|
||
'请对当前页面进行网页核对；“不操作网页”是引用；不调用resolve_reply_target。',
|
||
])
|
||
def test_negating_one_operation_does_not_cancel_independent_page_verification(text):
|
||
assert browser_recheck_requested(text)
|
||
|
||
|
||
@pytest.mark.parametrize('text', [
|
||
'请对当前页面进行网页核对；不操作网页，仅按已有原话复核。',
|
||
'请对当前页面进行网页核对；取消回查，仅整理已有记录。',
|
||
'请对当前页面进行网页核对；不需要操作网页，仅整理已有记录。',
|
||
'请对当前页面进行网页核对；不需要回查，仅整理已有记录。',
|
||
'先调用resolve_reply_target核对；不调用resolve_reply_target。',
|
||
'网页核对不用做，不调用resolve_reply_target。',
|
||
])
|
||
def test_explicit_page_cancellation_or_only_tool_cancellation_keeps_browser_off(text):
|
||
assert not browser_recheck_requested(text)
|
||
|
||
|
||
def test_only_explicit_prohibition_during_web_recheck_skips_analysis():
|
||
assert recheck_without_analysis(REQUEST)
|
||
assert recheck_without_analysis('返回原笔记核对，不要重新分析名单。')
|
||
assert not recheck_without_analysis('返回原笔记核对，然后重新分析名单。')
|
||
assert not recheck_without_analysis('返回原笔记核对，不要重新分析名单；最后仍要分析名单。')
|
||
assert not recheck_without_analysis('仅按现有原话复核，不要重新搜索。')
|
||
assert not recheck_without_analysis('返回原笔记核对，“不要分析名单”是引用。')
|
||
assert not recheck_without_analysis('返回原笔记核对，不要重新分析名单；最后再分析名单。')
|
||
|
||
|
||
def test_grounding_overrides_wrong_model_route_without_altering_other_user_bounds():
|
||
plan = IntentPlan(brief=ProspectingBrief(intent='discover', target_candidates=1, max_comments_per_note=5),
|
||
summary='只分析现有文字', missing=[TARGET_QUESTION], reuse_existing_sources=True)
|
||
corrected = ground_execution_plan(plan, REQUEST)
|
||
assert corrected.reuse_existing_sources is False and corrected.brief.intent == 'inspect'
|
||
assert corrected.brief.target_candidates == 1 and corrected.brief.max_comments_per_note == 5
|
||
assert TARGET_QUESTION not in corrected.missing
|
||
assert plan.reuse_existing_sources is True and plan.brief.intent == 'discover'
|
||
assert ground_execution_plan(plan, '不需要回网页，仅按已有原话复核。') == plan
|
||
|
||
|
||
async def test_real_intent_parser_corrects_reuse_flag_from_mocked_provider(monkeypatch):
|
||
wrong = IntentPlan(brief=ProspectingBrief(intent='research', user_notes='复核已保存文字'),
|
||
summary='仅复核已有资料', reuse_existing_sources=True)
|
||
cfg = {'provider': 'fixture', 'model': 'fixture-model', 'api_key': 'fixture-only', 'base_url': 'https://model.invalid'}
|
||
monkeypatch.setattr(dialogue, 'get_llm_config', lambda: cfg)
|
||
client_class = httpx.AsyncClient
|
||
transport = httpx.MockTransport(lambda request: httpx.Response(200, json={
|
||
'model': cfg['model'], 'choices': [{'message': {'content': wrong.model_dump_json()}}],
|
||
'usage': {'prompt_tokens': 10, 'completion_tokens': 10, 'total_tokens': 20}}))
|
||
monkeypatch.setattr(dialogue.httpx, 'AsyncClient', lambda **kwargs: client_class(transport=transport, **kwargs))
|
||
plan = await dialogue.interpret_message(REQUEST, ProspectingBrief(intent='research'), [])
|
||
assert plan.reuse_existing_sources is False
|
||
assert plan.brief.intent == 'research'
|
||
|
||
|
||
def test_recheck_prompt_starts_from_current_page_and_actual_history_instead_of_search_home(monkeypatch):
|
||
monkeypatch.setattr(subject.settings, 'reply_relation_preflight_enabled', True)
|
||
prompt = browser_prompt(ProspectingBrief(intent='inspect', user_notes=REQUEST), [])
|
||
assert '从当前页面观察起步' in prompt and '原生历史返回原笔记' in prompt
|
||
assert 'resolve_reply_target' in prompt
|
||
assert 'https://www.xiaohongshu.com/explore\n' not in prompt
|
||
assert '先打开下面的实际入口URL' not in prompt
|
||
assert '通过真实搜索入口→' not in prompt
|
||
|
||
|
||
@pytest.fixture
|
||
def memory(monkeypatch, tmp_path):
|
||
brief = ProspectingBrief(intent='research', user_notes=REQUEST, keywords=['装修'], max_notes=1)
|
||
row = {'id': 'recheck-routing-fixture', 'knowledge_id': KNOWLEDGE_ID, 'status': 'blocked',
|
||
'current_step': 'stage_results', 'browser_profile_path': '/fixture/browser',
|
||
'inputs': {**brief.model_dump(), 'goal': REQUEST, '_goal_version': 2,
|
||
'_execution_policy': subject.EXECUTION_POLICY, '_needs_interpretation': False,
|
||
'_requirements_grounding_version': subject.REQUIREMENTS_GROUNDING_VERSION},
|
||
'artifacts': {'messages': [{'role': 'user', 'text': REQUEST}], 'plan': {'missing': [], 'reuse_existing_sources': True},
|
||
'sources': [{'source_id': 'd352', 'source_type': 'reply', 'content': '实际原话'}],
|
||
'analysis_pending': True, 'analysis_goal_version': 1, 'analysis_cache': {'old': 'unchanged'},
|
||
'decisions': [{'source_id': 'd352', 'decision': 'keep', 'quote': '实际原话'}],
|
||
'leads': [{'source_id': 'd352', 'quote': '实际原话'}]}}
|
||
monkeypatch.setattr(subject.db, 'get_task', lambda tid: deepcopy(row))
|
||
def update(tid, **fields):
|
||
assert tid == row['id']
|
||
row.update(deepcopy(fields))
|
||
return deepcopy(row)
|
||
monkeypatch.setattr(subject.db, 'update_task', update)
|
||
monkeypatch.setattr(subject, 'get_llm_config', lambda: {'api_key': 'fixture-only'})
|
||
monkeypatch.setattr(subject, 'append_progress', lambda *args, **kwargs: None)
|
||
monkeypatch.setattr(subject, 'write_summary', lambda *args, **kwargs: None)
|
||
monkeypatch.setattr(subject, 'task_doc_dir', lambda tid: tmp_path)
|
||
engine = subject.ProspectingOrchestrator()
|
||
monkeypatch.setattr(engine, '_review', lambda tid, artifacts: artifacts)
|
||
monkeypatch.setattr(engine, '_learn', lambda tid, artifacts: None)
|
||
return row, engine
|
||
|
||
|
||
async def test_prepare_rechecks_raw_user_instruction_even_if_intent_model_returns_reuse(memory, monkeypatch):
|
||
row, engine = memory
|
||
row['inputs']['_needs_interpretation'] = True
|
||
wrong = IntentPlan(brief=ProspectingBrief(intent='discover', user_notes='只分析已保存文字', target_candidates=1),
|
||
summary='仅复核现有资料', reuse_existing_sources=True)
|
||
monkeypatch.setattr(subject, 'interpret_message', AsyncMock(return_value=wrong))
|
||
execute = AsyncMock()
|
||
monkeypatch.setattr(engine, '_run', execute)
|
||
result = await engine._prepare(deepcopy(row))
|
||
await engine.jobs[row['id']]
|
||
await asyncio.sleep(0)
|
||
assert result['artifacts']['analysis_pending'] is False
|
||
assert result['artifacts']['plan']['reuse_existing_sources'] is False
|
||
assert result['inputs']['intent'] == 'inspect'
|
||
assert execute.await_count == 1
|
||
|
||
|
||
async def test_page_verification_revision_preserves_view_when_one_tool_is_forbidden(memory, monkeypatch):
|
||
row, engine = memory
|
||
interrupt = AsyncMock(return_value=deepcopy(row))
|
||
monkeypatch.setattr(engine, '_interrupt_for_revision', interrupt)
|
||
monkeypatch.setattr(engine, '_prepare', AsyncMock(side_effect=lambda current: current))
|
||
text = '现在请对当前页面进行网页核对。本轮只读预检，不保存来源，最多1篇。不滚动、不重新导航、不调用resolve_reply_target。'
|
||
await engine.message(row['id'], text)
|
||
interrupt.assert_awaited_once_with(row['id'], preserve_browser=True)
|
||
|
||
|
||
@pytest.mark.parametrize('verdict', ['success', 'failed'])
|
||
async def test_live_route_guard_and_explicit_no_analysis_preserve_old_judgments(memory, monkeypatch, verdict):
|
||
row, engine = memory
|
||
original = deepcopy(row['artifacts'])
|
||
browser = AsyncMock(return_value={'verdict': verdict, 'history': {}, 'artifacts': {}, 'stop_reason': 'agent_done' if verdict == 'success' else 'no_progress'})
|
||
monkeypatch.setattr(subject.runner, 'run_browser_step', browser)
|
||
monkeypatch.setattr(subject.runner, 'has_live_session', lambda tid: False)
|
||
analyzer = AsyncMock(side_effect=AssertionError('User forbids a new paid list analysis'))
|
||
monkeypatch.setattr(subject, 'analyze_sources', analyzer)
|
||
await engine._run_scoped(row['id'], engine._brief(row), '')
|
||
browser.assert_awaited_once()
|
||
analyzer.assert_not_awaited()
|
||
assert '从当前页面观察起步' in browser.await_args.kwargs['agent_task']
|
||
assert browser.await_args.kwargs['require_existing_page'] is True
|
||
assert browser.await_args.kwargs['reuse_managed_view'] is True
|
||
for key in ('analysis_goal_version', 'analysis_cache', 'decisions', 'leads'):
|
||
assert row['artifacts'][key] == original[key]
|
||
assert row['artifacts']['analysis_pending'] is False
|
||
assert row['artifacts']['analysis_not_requested_goal_version'] == 2
|
||
assert row['artifacts']['browser_verdict'] == verdict
|
||
assert row['status'] == ('completed' if verdict == 'success' else 'blocked')
|
||
|
||
|
||
async def test_supplied_actual_entry_allows_a_new_view_without_relying_on_missing_history(memory, monkeypatch):
|
||
row, engine = memory
|
||
url = 'https://www.xiaohongshu.com/explore/'+'a'*24
|
||
row['inputs']['subject_urls'] = [url]
|
||
browser = AsyncMock(return_value={'verdict': 'failed', 'history': {}, 'artifacts': {}, 'stop_reason': 'no_progress'})
|
||
monkeypatch.setattr(subject.runner, 'run_browser_step', browser)
|
||
monkeypatch.setattr(subject.runner, 'has_live_session', lambda tid: False)
|
||
await engine._run_scoped(row['id'], engine._brief(row), '')
|
||
assert browser.await_args.kwargs['require_existing_page'] is False
|
||
assert '用户已提供的实际入口：'+url in browser.await_args.kwargs['agent_task']
|
||
|
||
|
||
async def test_pure_offline_review_keeps_normal_reuse_route(memory, monkeypatch):
|
||
from backend.prospecting import execution
|
||
row, engine = memory
|
||
row['inputs']['goal'] = '不需要回网页，仅按已有原话复核。'
|
||
row['inputs']['user_notes'] = row['inputs']['goal']
|
||
browser = AsyncMock(side_effect=AssertionError('Offline review must not open the browser'))
|
||
monkeypatch.setattr(subject.runner, 'run_browser_step', browser)
|
||
analyzer = AsyncMock(return_value={'leads': deepcopy(row['artifacts']['leads'])})
|
||
monkeypatch.setattr(execution, 'analyze_evidence', analyzer)
|
||
await engine._run_scoped(row['id'], engine._brief(row), '')
|
||
browser.assert_not_awaited()
|
||
analyzer.assert_awaited_once()
|
||
assert 'analysis_not_requested_goal_version' not in row['artifacts']
|