"""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"}