"""Regression coverage for idempotent Deep Research job submission."""

import json
from datetime import UTC, datetime
from types import SimpleNamespace

import pytest

from app.gateway import deep_research_report_rewrite as report_rewrite
from app.gateway.routers import deep_research


class _SessionStore:
    def __init__(self) -> None:
        self.updates: list[dict] = []

    async def get(self, session_id: str, *, user_id: str):  # noqa: ARG002
        return {
            "id": session_id,
            "query": "测试主题",
            "title": "测试主题",
            "config_snapshot": {"mode": "basic"},
        }

    async def count_active_by_user(self, *, user_id: str) -> int:  # noqa: ARG002
        return 0

    async def update(self, session_id: str, *, user_id: str, **fields):  # noqa: ARG002
        self.updates.append(fields)
        return {"id": session_id, **fields}


class _JobStore:
    async def get_active_for_session(self, session_id: str, *, user_id: str):  # noqa: ARG002
        return None

    async def try_create_or_get_active(self, **kwargs):  # noqa: ANN003
        # Simulate a retry after an earlier request committed the job but did
        # not finish responding to the browser.
        return {"id": "drj_existing", "status": "queued"}, False


class _Dispatcher:
    def __init__(self) -> None:
        self.nudges = 0

    def nudge(self) -> None:
        self.nudges += 1


def _request(*, sessions, jobs, dispatcher, sources=None):
    return SimpleNamespace(
        app=SimpleNamespace(
            state=SimpleNamespace(
                deep_research_session_store=sessions,
                deep_research_job_store=jobs,
                deep_research_dispatcher=dispatcher,
                deep_research_source_store=sources,
            )
        )
    )


@pytest.mark.asyncio
async def test_retry_of_existing_queued_job_nudges_dispatcher(monkeypatch):
    """An idempotent retry must recover a queued-but-not-yet-dispatched job."""

    async def fake_require_user(request):  # noqa: ANN001
        return "u1"

    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
    sessions = _SessionStore()
    dispatcher = _Dispatcher()
    request = _request(sessions=sessions, jobs=_JobStore(), dispatcher=dispatcher)

    response = await deep_research.start_job(
        "drs_1",
        deep_research.StartJobRequest(entry="legacy"),
        request,
    )

    assert response == {
        "jobId": "drj_existing",
        "sessionId": "drs_1",
        "status": "queued",
        "reused": True,
    }
    assert dispatcher.nudges == 1
    assert sessions.updates == [{"status": "running", "active_job_id": "drj_existing"}]


@pytest.mark.asyncio
async def test_queued_job_still_starts_when_session_projection_temporarily_fails(monkeypatch):
    """A committed job is more important than its non-authoritative UI projection."""

    async def fake_require_user(request):  # noqa: ANN001
        return "u1"

    class _ProjectionFailureSessionStore(_SessionStore):
        async def update(self, session_id: str, *, user_id: str, **fields):  # noqa: ARG002
            raise RuntimeError("temporary database readback failure")

    class _NewJobStore(_JobStore):
        async def try_create_or_get_active(self, **kwargs):  # noqa: ANN003
            return {"id": "drj_new", "status": "queued"}, True

    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
    dispatcher = _Dispatcher()
    response = await deep_research.start_job(
        "drs_1",
        deep_research.StartJobRequest(entry="legacy"),
        _request(
            sessions=_ProjectionFailureSessionStore(),
            jobs=_NewJobStore(),
            dispatcher=dispatcher,
        ),
    )

    assert response["jobId"] == "drj_new"
    assert response["reused"] is False
    assert dispatcher.nudges == 1


@pytest.mark.asyncio
async def test_chat_report_card_can_enable_images_after_collection(monkeypatch):
    """The write-time card override must win over the collection-time default."""

    async def fake_require_user(request):  # noqa: ANN001
        return "u1"

    class _ChatSessionStore(_SessionStore):
        async def get(self, session_id: str, *, user_id: str):  # noqa: ARG002
            return {
                "id": session_id,
                "query": "测试主题",
                "title": "测试主题",
                "runtime_thread_id": "collector-thread",
                "config_snapshot": {
                    "mode": "basic",
                    "collection_mode": "chat",
                    "generate_images": False,
                    "max_generated_images": 2,
                },
            }

    class _CapturingJobStore(_JobStore):
        def __init__(self) -> None:
            self.created: dict | None = None

        async def try_create_or_get_active(self, **kwargs):  # noqa: ANN003
            self.created = kwargs
            return {"id": "drj_images", "status": "queued"}, True

    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
    sessions = _ChatSessionStore()
    jobs = _CapturingJobStore()
    response = await deep_research.start_job(
        "drs_1",
        deep_research.StartJobRequest(
            entry="chat",
            config={"generate_images": True, "max_generated_images": 3},
        ),
        _request(sessions=sessions, jobs=jobs, dispatcher=_Dispatcher()),
    )

    assert response["jobId"] == "drj_images"
    assert jobs.created is not None
    snapshot = jobs.created["input_snapshot"]["config"]
    assert snapshot["generate_images"] is True
    assert snapshot["max_generated_images"] == 3
    # The session snapshot is persisted too, so a reload/replay keeps the
    # write-time choice instead of reverting to the creation-time false.
    assert any(
        update.get("config_snapshot", {}).get("generate_images") is True
        for update in sessions.updates
    )


@pytest.mark.asyncio
async def test_report_variant_uses_normal_job_with_selected_sources(monkeypatch):
    """Regeneration must queue the original report engine, never a rewrite job."""

    async def fake_require_user(request):  # noqa: ANN001
        return "u1"

    class _VariantSessionStore(_SessionStore):
        async def get(self, session_id: str, *, user_id: str):  # noqa: ARG002
            return {
                "id": session_id,
                "query": "新能源汽车市场",
                "title": "新能源汽车市场 · 新结构报告",
                "status": "draft",
                "runtime_thread_id": "dr_variant_thread",
                "config_snapshot": {
                    "mode": "basic",
                    "collection_mode": "chat",
                    "report_variant": True,
                    "parent_session_id": "drs_parent",
                },
            }

    class _SelectedSources:
        async def count_by_session(self, *args, **kwargs):  # noqa: ANN002, ANN003
            return 12

    class _CapturingJobStore(_JobStore):
        def __init__(self) -> None:
            self.created: dict | None = None

        async def try_create_or_get_active(self, **kwargs):  # noqa: ANN003
            self.created = kwargs
            return {"id": "drj_regenerate", "status": "queued"}, True

    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
    sessions = _VariantSessionStore()
    jobs = _CapturingJobStore()
    response = await deep_research.start_job(
        "drs_variant",
        deep_research.StartJobRequest(
            entry="regenerate",
            config={
                "custom_outline": "# 新结构\n\n## 第四章",
                "report_instruction": "统一表达并保留可核验事实。",
            },
        ),
        _request(
            sessions=sessions,
            jobs=jobs,
            dispatcher=_Dispatcher(),
            sources=_SelectedSources(),
        ),
    )

    assert response["jobId"] == "drj_regenerate"
    snapshot = jobs.created["input_snapshot"]
    assert snapshot["entry"] == "regenerate"
    assert snapshot["config"]["custom_outline"].endswith("## 第四章")
    assert snapshot["config"]["report_instruction"] == "统一表达并保留可核验事实。"
    assert "report_variant" not in snapshot["config"]
    assert "parent_session_id" not in snapshot["config"]
    assert any(
        update.get("config_snapshot", {}).get("report_variant") is True
        for update in sessions.updates
    )
    assert any(
        update.get("config_snapshot", {}).get("parent_session_id") == "drs_parent"
        for update in sessions.updates
    )


@pytest.mark.asyncio
async def test_stream_stays_open_when_durable_replay_has_a_transient_failure(monkeypatch):
    """A stream DB blip must yield a recoverable warning, not an ASGI error group."""

    async def fake_require_user(request):  # noqa: ANN001
        return "u1"

    class _StreamJobStore:
        async def get(self, job_id: str, *, user_id: str):  # noqa: ARG002
            return {"id": job_id, "session_id": "drs_1"}

        async def get_unscoped(self, job_id: str):  # noqa: ARG002
            return None

    class _UnavailableEventStore:
        async def list_after(self, *args, **kwargs):  # noqa: ANN002, ANN003
            raise RuntimeError("temporary replica failure")

    async def is_connected() -> bool:
        return False

    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
    request = SimpleNamespace(
        app=SimpleNamespace(
            state=SimpleNamespace(
                deep_research_job_store=_StreamJobStore(),
                deep_research_event_store=_UnavailableEventStore(),
            )
        ),
        is_disconnected=is_connected,
    )

    response = await deep_research.stream_job("drj_1", request)
    chunk = await anext(response.body_iterator)
    await response.body_iterator.aclose()

    assert "durable_replay_temporarily_unavailable" in chunk


@pytest.mark.asyncio
async def test_full_report_rewrite_persists_selected_report_structure(monkeypatch):
    """A selected structure must survive durable replay and constrain the worker."""

    async def fake_require_user(request):  # noqa: ANN001
        return "u1"

    class _CompletedSessionStore(_SessionStore):
        async def get(self, session_id: str, *, user_id: str):  # noqa: ARG002
            return {
                "id": session_id,
                "query": "测试主题",
                "title": "测试主题",
                "status": "completed",
                "report_markdown": "# 原报告\n\n原有内容。",
                "config_snapshot": {"mode": "basic"},
            }

    class _SourceStore:
        async def list_by_session(self, *args, **kwargs):  # noqa: ANN002, ANN003
            # SQL-backed historical sources return native datetimes, whereas
            # the job snapshot is persisted as a JSON column.
            return [{"id": "src_1", "created_at": datetime(2026, 8, 21, tzinfo=UTC)}]

    class _CapturingJobStore(_JobStore):
        def __init__(self) -> None:
            self.created: dict | None = None

        async def try_create_or_get_active(self, **kwargs):  # noqa: ANN003
            # Match the repository's JSON persistence boundary. This used to
            # raise TypeError for historical source rows with datetime fields.
            json.dumps(kwargs["input_snapshot"], ensure_ascii=False)
            self.created = kwargs
            return {"id": "drj_rewrite", "status": "queued"}, True

    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
    jobs = _CapturingJobStore()
    outline = "# 执行摘要\n\n# 一、核心发现\n\n# 二、行动建议"
    response = await deep_research.start_full_report_rewrite_job(
        "drs_1",
        deep_research.FullReportRewriteRequest(
            instruction="压缩重复段落，保持事实准确。",
            reportOutline=outline,
            operation="generate",
        ),
        _request(
            sessions=_CompletedSessionStore(),
            jobs=jobs,
            dispatcher=_Dispatcher(),
            sources=_SourceStore(),
        ),
    )

    assert response["jobId"] == "drj_rewrite"
    assert jobs.created is not None
    snapshot = jobs.created["input_snapshot"]
    assert snapshot["operation"] == "generate"
    assert snapshot["report_outline"] == outline
    assert "【报告结构要求】" in snapshot["instruction"]
    assert "# 一、核心发现" in snapshot["instruction"]
    assert snapshot["sources"][0]["created_at"] == "2026-08-21T00:00:00+00:00"


@pytest.mark.asyncio
async def test_regeneration_forks_report_and_selected_sources(monkeypatch):
    """Choosing another structure must never mutate the completed parent report."""

    async def fake_require_user(request):  # noqa: ANN001
        return "u1"

    class _VariantSessionStore:
        def __init__(self) -> None:
            self.parent = {
                "id": "drs_parent",
                "user_id": "u1",
                "title": "新能源研究",
                "query": "新能源市场",
                "mode": "basic",
                "status": "completed",
                "report_markdown": (
                    "# 原报告\n\n"
                    "已保存引用 [来源：src_parent]；"
                    "流式引用 [[source:src_parent]]；"
                    "已取消的资料 [[source:src_unselected]]。"
                ),
                "config_snapshot": {"mode": "basic", "custom_outline": "# 原结构"},
                "source_count": 1,
            }
            self.created: dict | None = None
            self.updated: dict | None = None
            self.deleted: list[str] = []

        async def get(self, session_id: str, *, user_id: str):  # noqa: ARG002
            return self.parent if session_id == "drs_parent" else None

        async def create(self, **fields):  # noqa: ANN003
            self.created = fields
            return {**fields, "report_markdown": None, "source_count": 0}

        async def update(self, session_id: str, *, user_id: str, **fields):  # noqa: ARG002
            self.updated = {"session_id": session_id, **fields}
            return {
                "id": session_id,
                "user_id": "u1",
                "title": self.created["title"] if self.created else "",
                "query": "新能源市场",
                "mode": "basic",
                "status": self.created["status"] if self.created else "draft",
                "config_snapshot": self.created["config_snapshot"] if self.created else {},
                **fields,
            }

        async def delete(self, session_id: str, *, user_id: str):  # noqa: ARG002
            self.deleted.append(session_id)
            return True

    class _VariantSourceStore:
        async def clone_selected_to_session(self, source_session_id: str, target_session_id: str, *, user_id: str):  # noqa: ARG002
            assert source_session_id == "drs_parent"
            assert target_session_id.startswith("drs_")
            return {"src_parent": "src_variant"}

    monkeypatch.setattr(deep_research, "_require_user", fake_require_user)
    async def fake_ensure_runtime_thread(request, session, user_id):  # noqa: ANN001, ARG001
        await sessions.update(session["id"], user_id=user_id, runtime_thread_id="dr_variant_thread")

    sessions = _VariantSessionStore()
    monkeypatch.setattr(deep_research, "_ensure_chat_runtime_thread", fake_ensure_runtime_thread)
    response = await deep_research.create_report_variant(
        "drs_parent",
        deep_research.CreateReportVariantRequest(reportOutline="# 新结构\n\n## 结论"),
        _request(
            sessions=sessions,
            jobs=_JobStore(),
            dispatcher=_Dispatcher(),
            sources=_VariantSourceStore(),
        ),
    )

    assert response.id != "drs_parent"
    assert response.title == "新能源研究 · 新结构报告"
    assert response.report_markdown is None
    assert response.status == "draft"
    assert response.config == {
        "mode": "basic",
        "custom_outline": "# 新结构\n\n## 结论",
        "report_variant": True,
        "parent_session_id": "drs_parent",
    }
    assert sessions.parent["report_markdown"] == (
        "# 原报告\n\n"
        "已保存引用 [来源：src_parent]；"
        "流式引用 [[source:src_parent]]；"
        "已取消的资料 [[source:src_unselected]]。"
    )
    assert sessions.updated is not None
    assert sessions.updated["source_count"] == 1
    assert sessions.deleted == []


def test_normalise_citations_accepts_display_ready_and_legacy_markers() -> None:
    source_ids = {"drs_src_1234567890abcdef1234567890abcdef"}

    result, cited = report_rewrite._normalise_citations(  # noqa: SLF001
        "甲[来源：drs_src_1234567890abcdef1234567890abcdef]；"
        "乙[[source:drs_src_1234567890abcdef1234567890abcdef]]；"
        "丙[来源: drs_src_1234567890abcdef1234567890abcdef]。",
        source_ids,
    )

    assert result == (
        "甲[来源：drs_src_1234567890abcdef1234567890abcdef]；"
        "乙[来源：drs_src_1234567890abcdef1234567890abcdef]；"
        "丙[来源：drs_src_1234567890abcdef1234567890abcdef]。"
    )
    assert cited == ["drs_src_1234567890abcdef1234567890abcdef"]


def test_normalise_citations_repairs_unique_truncated_selected_id() -> None:
    full_id = "drs_src_ac653c4a6f364e59a30b044e98d32e2f"

    result, cited = report_rewrite._normalise_citations(  # noqa: SLF001
        "事实[来源：drs_src_ac653c4a6f364e59a30b044e98d32e]。",
        {full_id},
    )

    assert result == f"事实[来源：{full_id}]。"
    assert cited == [full_id]


def test_normalise_citations_still_rejects_unknown_source() -> None:
    with pytest.raises(ValueError, match="未选资料来源"):
        report_rewrite._normalise_citations(  # noqa: SLF001
            "事实[来源：drs_src_ffffffffffffffffffffffffffffffff]。",
            {"drs_src_ac653c4a6f364e59a30b044e98d32e2f"},
        )


def test_normalise_citations_removes_provider_truncated_marker_without_discarding_report() -> None:
    result, cited = report_rewrite._normalise_citations(  # noqa: SLF001
        "# 参考资料\n\n已完成的正文。\n[来源：drs_src_ac653c4a6f364e59a30b044e98d32",
        {"drs_src_ac653c4a6f364e59a30b044e98d32e2f"},
    )

    assert result == "# 参考资料\n\n已完成的正文。"
    assert cited == []


@pytest.mark.asyncio
async def test_generate_operation_uses_selected_sources_not_inherited_report(monkeypatch) -> None:
    captured: dict = {}

    class _CompletionBackend:
        def __init__(self, config) -> None:  # noqa: ANN001
            captured["config"] = config

        async def stream_complete(self, **kwargs):  # noqa: ANN003
            captured.update(kwargs)
            return SimpleNamespace(
                text=(
                    "# 新报告\n\n事实[来源：drs_src_1234567890abcdef1234567890abcdef]。\n\n"
                    "<!-- DEERFLOW_REPORT_COMPLETE -->"
                ),
                usage={},
            )

    monkeypatch.setattr(report_rewrite, "DeerFlowCompletionBackend", _CompletionBackend)

    async def _ignore(_delta: str) -> None:
        return None

    result, cited, _usage = await report_rewrite.stream_full_report_rewrite(
        session={"config_snapshot": {"mode": "basic"}, "query": "新能源汽车"},
        report="# 不应进入模型的旧报告\n\n旧结构和旧措辞。",
        sources=[
            {
                "id": "drs_src_1234567890abcdef1234567890abcdef",
                "title": "已选资料",
                "raw_content": "可核验事实",
            }
        ],
        instruction="按新结构生成",
        style="formal_analysis",
        model_name=None,
        on_delta=_ignore,
        operation="generate",
    )

    prompt = "\n".join(str(message["content"]) for message in captured["messages"])
    assert "不应进入模型的旧报告" not in prompt
    assert "# 已选来源" in prompt
    assert captured["operation"] == "full_report_generate"
    assert captured["max_tokens"] == 8_192
    assert result.endswith("[来源：drs_src_1234567890abcdef1234567890abcdef]。")
    assert cited == ["drs_src_1234567890abcdef1234567890abcdef"]


@pytest.mark.asyncio
async def test_report_generation_continues_when_first_segment_has_no_completion_marker(monkeypatch) -> None:
    calls: list[str] = []

    class _CompletionBackend:
        def __init__(self, _config) -> None:  # noqa: ANN001
            pass

        async def stream_complete(self, **kwargs):  # noqa: ANN003
            calls.append(str(kwargs["operation"]))
            if len(calls) == 1:
                return SimpleNamespace(text="# 新报告\n\n第一部分。", usage={"completion_tokens": 100})
            return SimpleNamespace(
                text="## 第二部分\n\n续写完成。\n\n<!-- DEERFLOW_REPORT_COMPLETE -->",
                usage={"completion_tokens": 50},
            )

    monkeypatch.setattr(report_rewrite, "DeerFlowCompletionBackend", _CompletionBackend)

    async def _ignore(_delta: str) -> None:
        return None

    result, cited, usage = await report_rewrite.stream_full_report_rewrite(
        session={"config_snapshot": {"mode": "basic"}, "query": "新能源汽车"},
        report="# 旧报告",
        sources=[],
        instruction="按新结构生成",
        style="formal_analysis",
        model_name=None,
        on_delta=_ignore,
        operation="generate",
    )

    assert calls == ["full_report_generate", "full_report_generate_continue"]
    assert result == "# 新报告\n\n第一部分。\n\n## 第二部分\n\n续写完成。"
    assert cited == []
    assert usage["completion_tokens"] == 150