"""Failure-ordering regression coverage for final Deep Research report writes."""

import pytest

from app.gateway import deep_research_job_executor as executor_module
from app.gateway.deep_research_job_executor import DeepResearchJobExecutor
from deerflow.agents.deep_research.events import DeepResearchEvent
from deerflow.agents.deep_research.types import DeepResearchResult


class _JobStore:
    def __init__(self, session_store) -> None:
        self._session_store = session_store
        self.progress_calls: list[dict] = []

    async def update_progress(self, job_id: str, **fields):  # noqa: ARG002
        assert self._session_store.report_updates, "report must be persisted before job completion"
        self.progress_calls.append(fields)
        return {"id": job_id, **fields}


class _SessionStore:
    def __init__(self, *, available: bool = True) -> None:
        self.available = available
        self.report_updates: list[dict] = []

    async def update(self, session_id: str, **fields):  # noqa: ARG002
        if not self.available:
            raise RuntimeError("temporary session database failure")
        self.report_updates.append(fields)
        return {"id": session_id, **fields}


@pytest.mark.asyncio
async def test_report_is_persisted_before_job_is_marked_completed():
    sessions = _SessionStore()
    jobs = _JobStore(sessions)
    runner = DeepResearchJobExecutor(jobs, sessions, event_store=None)

    await runner._mark_completed(
        "drj_1",
        "drs_1",
        "worker_1",
        DeepResearchResult(report_markdown="# 报告", source_ids=["src_1"]),
        user_id="user_1",
        query="课题",
    )

    assert sessions.report_updates[0]["report_markdown"] == "# 报告"
    assert jobs.progress_calls[0]["status"] == "completed"


@pytest.mark.asyncio
async def test_report_projection_failure_never_marks_job_completed(monkeypatch):
    monkeypatch.setattr(executor_module, "_REPORT_PROJECTION_RETRY_ATTEMPTS", 2)
    monkeypatch.setattr(executor_module, "_REPORT_PROJECTION_RETRY_DELAY_SECONDS", 0)
    sessions = _SessionStore(available=False)
    jobs = _JobStore(sessions)
    runner = DeepResearchJobExecutor(jobs, sessions, event_store=None)

    with pytest.raises(RuntimeError, match="研究报告未能保存"):
        await runner._mark_completed(
            "drj_1",
            "drs_1",
            "worker_1",
            DeepResearchResult(report_markdown="# 报告"),
            user_id="user_1",
            query="课题",
        )

    assert jobs.progress_calls == []


def test_terminal_timeline_events_are_valid_durable_event_types():
    assert DeepResearchEvent(type="job_completed", phase="done").type == "job_completed"
    assert DeepResearchEvent(type="job_failed", phase="done").type == "job_failed"


def test_zero_source_fallback_is_a_valid_completed_report():
    from app.gateway.deep_research_job_executor import _terminal_report_validation_error
    from deerflow.agents.deep_research.runners.basic import _build_report_fallback

    report = _build_report_fallback("没有命中的课题", "", 0)
    assert _terminal_report_validation_error(report) is None
    assert "没有命中的课题" in report


def test_fixed_outline_fallback_retains_every_selected_heading():
    from deerflow.agents.deep_research.runners.basic import (
        _build_report_fallback,
        _outline_adherence_error,
    )

    outline = """# 报告标题

# 一、现状判断
## 1.1 事实基线
## 1.2 核心矛盾
# 二、实施方案
## 2.1 重点任务
# 参考资料"""
    report = _build_report_fallback(
        "园区升级方案",
        "已确认材料 A\n已确认材料 B",
        2,
        custom_outline=outline,
        structure_mode="fixed",
    )

    assert "# 园区升级方案" in report
    for heading in (
        "# 一、现状判断",
        "## 1.1 事实基线",
        "## 1.2 核心矛盾",
        "# 二、实施方案",
        "## 2.1 重点任务",
        "# 参考资料",
    ):
        assert heading in report
    assert (
        _outline_adherence_error(
            report,
            outline,
            structure_mode="fixed",
        )
        is None
    )
    assert "# 一、现状判断" in (
        _outline_adherence_error(
            report.replace("# 一、现状判断", "## 一、现状判断"),
            outline,
            structure_mode="fixed",
        )
        or ""
    )


def test_adaptive_outline_fallback_keeps_core_chapters_without_forcing_subsections():
    from deerflow.agents.deep_research.runners.basic import _build_report_fallback

    outline = """# 报告标题
## 一、事实基线
### 1.1 可选细节
## 二、行动建议
### 2.1 可选措施"""
    report = _build_report_fallback(
        "产业研究",
        "",
        0,
        custom_outline=outline,
        structure_mode="adaptive",
    )

    assert "# 产业研究" in report
    assert "## 一、事实基线" in report
    assert "## 二、行动建议" in report
    assert "### 1.1 可选细节" not in report
    assert "### 2.1 可选措施" not in report


def test_adaptive_outline_matching_ignores_generic_sections_and_chapter_prefixes():
    from deerflow.agents.deep_research.runners.basic import _outline_adherence_error

    outline = """# 报告标题
## 第一章 引言
## 第二章 市场态势
## 第三章 结论
## 参考资料"""
    generic_report = """# 通用报告
## 引言
## 结论
## 参考资料"""

    error = _outline_adherence_error(
        generic_report,
        outline,
        structure_mode="adaptive",
    )

    assert error is not None
    assert "第二章 市场态势" in error
    assert (
        _outline_adherence_error(
            "# 市场研究\n\n## 二、市场态势\n\n已按材料展开。",
            outline,
            structure_mode="adaptive",
        )
        is None
    )
    assert (
        _outline_adherence_error(
            "# 简报\n\n## 引言\n\n正文。",
            "# 报告标题\n\n## 第一章 引言\n\n## 第二章 结论",
            structure_mode="adaptive",
        )
        is None
    )
    named_generic_outline = "# 市场态势报告\n\n## 第一章 引言\n\n## 第二章 结论"
    assert (
        _outline_adherence_error(
            "# 市场态势报告\n\n## 完全不同的通用正文",
            named_generic_outline,
            structure_mode="adaptive",
        )
        is not None
    )
    assert (
        _outline_adherence_error(
            "# 改写后的题名\n\n## 引言\n\n正文。",
            named_generic_outline,
            structure_mode="adaptive",
        )
        is None
    )


class _ReportEvents:
    def __init__(self) -> None:
        self.events: list[tuple[str, dict]] = []

    async def emit(self, event_type: str, *, phase: str = "initializing", payload=None):
        self.events.append((event_type, {"phase": phase, "payload": payload or {}}))
        return None

    async def publish_live(self, event_type: str, *, phase: str = "initializing", payload=None):
        self.events.append((event_type, {"phase": phase, "payload": payload or {}}))


class _IgnoringOutlineLLM:
    def __init__(self) -> None:
        self.calls: list[list[dict[str, str]]] = []

    async def stream_complete(self, *, messages, on_delta, **kwargs):  # noqa: ARG002
        from deerflow.agents.deep_research.types import CompletionResult

        self.calls.append(messages)
        text = "# 通用报告\n\n## 摘要\n\n这是未按所选模板生成的普通报告正文。"
        await on_delta(text)
        return CompletionResult(text=text)

    async def complete_prose(self, *, messages, **kwargs):  # noqa: ARG002
        from deerflow.agents.deep_research.types import CompletionResult

        self.calls.append(messages)
        return CompletionResult(text="# 通用报告\n\n## 结论\n\n重试仍然忽略固定大纲。")


class _PartialAdaptiveOutlineLLM(_IgnoringOutlineLLM):
    async def stream_complete(self, *, messages, on_delta, **kwargs):  # noqa: ARG002
        from deerflow.agents.deep_research.types import CompletionResult

        self.calls.append(messages)
        text = "# 年度增长方案\n\n## 一、目标\n\n保留了一个模板核心章节，其余内容按材料弹性组织。"
        await on_delta(text)
        return CompletionResult(text=text)

    async def complete_prose(self, *, messages, **kwargs):  # noqa: ARG002
        raise AssertionError("adaptive outline with core overlap must not retry")


@pytest.mark.asyncio
async def test_fixed_outline_mismatch_uses_one_retry_then_template_shaped_final_report():
    from deerflow.agents.deep_research.config import DeepResearchConfig
    from deerflow.agents.deep_research.runners.basic import BasicRunner
    from deerflow.agents.deep_research.types import DeepResearchRequest

    outline = "# 报告标题\n\n# 一、目标\n\n# 二、路径\n\n# 三、验收"
    request = DeepResearchRequest(
        session_id="session-outline",
        user_id="alice",
        query="年度增长方案",
        config=DeepResearchConfig(
            custom_outline=outline,
            structure_mode="fixed",
        ),
    )
    llm = _IgnoringOutlineLLM()
    events = _ReportEvents()

    report = await BasicRunner()._generate_report(
        llm,
        request,
        "已确认项目目标与约束",
        events,
        source_count=1,
    )

    assert len(llm.calls) == 2
    assert "固定大纲结构校验失败" in llm.calls[1][-1]["content"]
    assert report.index("# 一、目标") < report.index("# 二、路径") < report.index("# 三、验收")
    assert any(
        event_type == "warning"
        and event["payload"].get("code") == "report_template_fallback"
        for event_type, event in events.events
    )


@pytest.mark.asyncio
async def test_adaptive_outline_does_not_retry_a_material_driven_layout():
    from deerflow.agents.deep_research.config import DeepResearchConfig
    from deerflow.agents.deep_research.runners.basic import BasicRunner
    from deerflow.agents.deep_research.types import DeepResearchRequest

    request = DeepResearchRequest(
        session_id="session-adaptive-outline",
        user_id="alice",
        query="年度增长方案",
        config=DeepResearchConfig(
            custom_outline="# 报告标题\n\n## 一、目标\n\n## 二、路径",
            structure_mode="adaptive",
        ),
    )
    llm = _PartialAdaptiveOutlineLLM()

    report = await BasicRunner()._generate_report(
        llm,
        request,
        "已确认项目目标与约束",
        _ReportEvents(),
        source_count=1,
    )

    assert len(llm.calls) == 1
    assert "## 一、目标" in report


@pytest.mark.asyncio
async def test_adaptive_outline_with_zero_core_overlap_retries_then_keeps_core_shape():
    from deerflow.agents.deep_research.config import DeepResearchConfig
    from deerflow.agents.deep_research.runners.basic import BasicRunner
    from deerflow.agents.deep_research.types import DeepResearchRequest

    request = DeepResearchRequest(
        session_id="session-adaptive-outline-missed",
        user_id="alice",
        query="年度增长方案",
        config=DeepResearchConfig(
            custom_outline="# 报告标题\n\n## 一、目标\n\n### 1.1 指标\n\n## 二、路径\n\n### 2.1 动作",
            structure_mode="adaptive",
        ),
    )
    llm = _IgnoringOutlineLLM()

    report = await BasicRunner()._generate_report(
        llm,
        request,
        "已确认项目目标与约束",
        _ReportEvents(),
        source_count=1,
    )

    assert len(llm.calls) == 2
    assert "弹性大纲结构校验失败" in llm.calls[1][-1]["content"]
    assert "## 一、目标" in report
    assert "## 二、路径" in report
    assert "### 1.1 指标" not in report


def test_source_table_reload_does_not_become_session_error():
    from app.gateway.deep_research_job_executor import _session_diagnostic_error

    assert _session_diagnostic_error(
        [
            {
                "code": "client_materials_reloaded_before_write",
                "stage": "collecting",
                "detail": "写作前再次从素材表读到 15 条",
            }
        ]
    ) is None