111 lines
6.0 KiB
Python
111 lines
6.0 KiB
Python
"""Continuation repairs compiled requirements without rewriting task evidence."""
|
||
import asyncio
|
||
from copy import deepcopy
|
||
from unittest.mock import AsyncMock
|
||
|
||
import pytest
|
||
|
||
from backend.prospecting import orchestrator as subject
|
||
from backend.prospecting.contracts import KNOWLEDGE_ID, ProspectingBrief
|
||
|
||
|
||
REQUEST = (
|
||
"我提供杭州装修服务，请在小红书找2位本人明确表达装修或设计服务需求的候选，优先查看评论和回复。"
|
||
"按人去重，整理需求原话、来源链接和主页入口；页面可见的小红书号能取得就记录，没有就说明。"
|
||
"最后导出名单，只做阅读和整理，不关注、不私信。"
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def saved(monkeypatch, tmp_path):
|
||
brief = ProspectingBrief(target_candidates=1, required_result_fields=["csv", "xiaohongshu_id"],
|
||
target_customer="本人明确表达装修或设计服务需求的用户", user_notes=REQUEST)
|
||
row = {"id": "resume-d38", "knowledge_id": KNOWLEDGE_ID, "status": "blocked", "current_step": "stage_results",
|
||
"inputs": {**brief.model_dump(), "goal": REQUEST, "_goal_version": 2, "_execution_policy": "conversation-v1",
|
||
"_needs_interpretation": False},
|
||
"artifacts": {"messages": [{"role": "user", "text": REQUEST},
|
||
{"role": "user", "text": "补充当前任务的信息：这次先整理多少位候选？：1"}],
|
||
"plan": {"missing": []}, "sources": [{"source_id": "s1", "source_type": "note", "content": "想找装修公司"}],
|
||
"leads": [{"source_id": "s1", "quote": "想找装修公司", "decision": "keep"}],
|
||
"operation_evidence": [{"id": "old-fact", "details": {"trace": {"run_id": "old-run"}}}],
|
||
"exports": [{"id": "old-file", "goal_version": 2}], "analysis_goal_version": 2,
|
||
"analysis_pending": False, "goal_versions": [{"version": 1}]}}
|
||
updates = []
|
||
def update(tid, **fields):
|
||
assert tid == row["id"]
|
||
updates.append(deepcopy(fields))
|
||
row.update(deepcopy(fields))
|
||
return deepcopy(row)
|
||
monkeypatch.setattr(subject.db, "get_task", lambda tid: deepcopy(row))
|
||
monkeypatch.setattr(subject.db, "update_task", update)
|
||
monkeypatch.setattr(subject, "append_progress", lambda *args, **kwargs: None)
|
||
monkeypatch.setattr(subject, "task_doc_dir", lambda tid: tmp_path)
|
||
monkeypatch.setattr(subject, "get_llm_config", lambda: {"api_key": "fixture-only"})
|
||
return row, subject.ProspectingOrchestrator(), updates
|
||
|
||
|
||
async def test_explicit_resume_repairs_fields_and_preserves_confirmed_one_person_and_original_facts(saved, monkeypatch):
|
||
row, engine, updates = saved
|
||
original = deepcopy(row)
|
||
nlu = AsyncMock(side_effect=AssertionError("continuation must not re-run intent model"))
|
||
execute = AsyncMock()
|
||
monkeypatch.setattr(subject, "interpret_message", nlu)
|
||
monkeypatch.setattr(engine, "_run", execute)
|
||
result = await engine.resume(row["id"])
|
||
job = engine.jobs[row["id"]]
|
||
await job
|
||
await asyncio.sleep(0)
|
||
assert result["status"] == "running"
|
||
assert result["inputs"]["target_candidates"] == 1
|
||
assert set(result["inputs"]["required_result_fields"]) == {"csv", "explicit_demand"}
|
||
assert result["inputs"]["_goal_version"] == 3
|
||
assert result["artifacts"]["goal_versions"][-1]["version"] == 2
|
||
assert result["artifacts"]["goal_versions"][-1]["brief"]["target_candidates"] == 1
|
||
for key in ("sources", "operation_evidence", "exports", "analysis_goal_version", "analysis_pending"):
|
||
assert result["artifacts"][key] == original["artifacts"][key]
|
||
assert result["artifacts"]["requirement_corrections"][0]["source"] == "original_user_instructions"
|
||
assert execute.await_args.args[1].target_candidates == 1
|
||
assert "explicit_demand" in execute.await_args.args[1].required_result_fields
|
||
nlu.assert_not_awaited()
|
||
before = deepcopy(row)
|
||
assert engine._reground_requirements_on_resume(row) == before
|
||
assert row["inputs"]["_goal_version"] == 3
|
||
|
||
|
||
def test_no_user_history_or_new_grounding_marker_never_rewrites_explicit_plan(saved):
|
||
row, engine, updates = saved
|
||
row["artifacts"]["messages"] = []
|
||
before = deepcopy(row)
|
||
assert engine._reground_requirements_on_resume(row) == before
|
||
assert not updates
|
||
row["artifacts"]["messages"] = [{"role": "user", "text": REQUEST}]
|
||
row["inputs"]["_requirements_grounding_version"] = subject.REQUIREMENTS_GROUNDING_VERSION
|
||
before = deepcopy(row)
|
||
assert engine._reground_requirements_on_resume(row) == before
|
||
assert not updates
|
||
|
||
|
||
async def test_review_failure_continuation_rechecks_fields_before_retrying_analysis(saved, monkeypatch):
|
||
row, engine, _ = saved
|
||
row["artifacts"]["review_error"] = "fixture analysis failure"
|
||
review = AsyncMock(side_effect=lambda tid: deepcopy(row))
|
||
monkeypatch.setattr(engine, "review_saved", review)
|
||
result = await engine.resume(row["id"])
|
||
review.assert_awaited_once_with(row["id"])
|
||
assert result["inputs"]["target_candidates"] == 1
|
||
assert result["inputs"]["_goal_version"] == 3
|
||
assert set(result["inputs"]["required_result_fields"]) == {"csv", "explicit_demand"}
|
||
assert result["artifacts"]["operation_evidence"][0]["details"]["trace"] == {"run_id": "old-run"}
|
||
|
||
|
||
def test_navigation_attempt_recovery_keeps_newer_database_receipt_and_unique_snapshot(saved, monkeypatch, tmp_path):
|
||
import json
|
||
row, engine, _ = saved
|
||
monkeypatch.setattr(subject, "task_doc_dir", lambda tid: tmp_path)
|
||
(tmp_path / "evidence.json").write_text(json.dumps({"navigation_attempts": [
|
||
{"id": "same", "status": "old"}, {"id": "snapshot-only", "status": "not_moved"}]}))
|
||
original = {"navigation_attempts": [{"id": "same", "status": "new"}, {"id": "db-only", "status": "opened"}]}
|
||
recovered = engine._recover_evidence(row["id"], original)
|
||
assert {item["id"]: item["status"] for item in recovered["navigation_attempts"]} == {
|
||
"same": "new", "snapshot-only": "not_moved", "db-only": "opened"}
|