"""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']