"""End-to-end effects tests for project-scoped cross-conversation memory.

These tests deliberately use the real SQLite repositories and the real run
policy.  Only infrastructure outside this feature's boundary is substituted:
the completed-thread checkpointer, the extraction LLM, and the unrelated
generic/personal memory switch used by independence regressions.
"""

from __future__ import annotations

import asyncio
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest
from fastapi import FastAPI, HTTPException, Request
from langchain_core.messages import AIMessage, HumanMessage
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine

from app.gateway import analysis_project_memory as memory_runtime
from app.gateway import analysis_projects as project_policy
from app.gateway.routers.thread_runs import RunCreateRequest
from deerflow.agents.analysis_project import project_system_prompt
from deerflow.config import system_settings
from deerflow.persistence.analysis_project_memory import (
    AnalysisProjectMemoryItemRow,
    AnalysisProjectMemoryJobRow,
    AnalysisProjectMemoryStateRow,
    AnalysisProjectMemoryStore,
)
from deerflow.persistence.analysis_projects import (
    AnalysisProjectConfigRow,
    AnalysisProjectRow,
    AnalysisProjectStore,
    AnalysisProjectThreadRow,
)
from deerflow.runtime.user_context import reset_current_user, set_current_user


class _Checkpoint:
    """Small mutable checkpointer implementing the one read used by memory."""

    def __init__(self) -> None:
        self._messages: dict[str, list[object]] = {}

    def set_messages(self, thread_id: str, messages: list[object]) -> None:
        self._messages[thread_id] = messages

    async def aget_tuple(self, config: dict) -> SimpleNamespace | None:
        thread_id = str(config.get("configurable", {}).get("thread_id") or "")
        messages = self._messages.get(thread_id)
        if messages is None:
            return None
        return SimpleNamespace(
            checkpoint={"channel_values": {"messages": messages}}
        )


@dataclass
class _Harness:
    engine: AsyncEngine
    projects: AnalysisProjectStore
    memories: AnalysisProjectMemoryStore
    checkpointer: _Checkpoint
    app: FastAPI
    request: Request

    async def close(self) -> None:
        await self.engine.dispose()


async def _make_harness(tmp_path) -> _Harness:
    engine = create_async_engine(
        f"sqlite+aiosqlite:///{tmp_path / 'project-memory-integration.db'}"
    )

    @event.listens_for(engine.sync_engine, "connect")
    def _enable_foreign_keys(dbapi_connection, _connection_record):  # noqa: ANN001
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()

    # Create only the production tables exercised by this feature.  This keeps
    # the test independent from unrelated schemas while preserving all real
    # foreign keys, constraints, transactions, and repository code paths.
    async with engine.begin() as connection:
        for table in (
            AnalysisProjectConfigRow.__table__,
            AnalysisProjectRow.__table__,
            AnalysisProjectThreadRow.__table__,
            AnalysisProjectMemoryStateRow.__table__,
            AnalysisProjectMemoryItemRow.__table__,
            AnalysisProjectMemoryJobRow.__table__,
        ):
            await connection.run_sync(table.create)

    session_factory = async_sessionmaker(engine, expire_on_commit=False)
    projects = AnalysisProjectStore(session_factory)
    memories = AnalysisProjectMemoryStore(session_factory)
    checkpointer = _Checkpoint()
    app = FastAPI()
    app.state.analysis_project_store = projects
    app.state.analysis_project_memory_store = memories
    app.state.analysis_project_memory_dispatcher = None
    app.state.checkpointer = checkpointer
    # Absence of an encoder is a supported production degradation path.
    app.state.llmwiki_embedding = None
    request = Request(
        {
            "type": "http",
            "app": app,
            "headers": [],
            "method": "POST",
            "path": "/api/threads/test/runs",
            "query_string": b"",
            "root_path": "",
            "scheme": "http",
            "server": ("testserver", 80),
            "client": ("testclient", 50000),
        }
    )
    return _Harness(engine, projects, memories, checkpointer, app, request)


@contextmanager
def _as_user(user_id: str):
    token = set_current_user(SimpleNamespace(id=user_id))
    try:
        yield
    finally:
        reset_current_user(token)


def _set_master_switch(monkeypatch: pytest.MonkeyPatch, *, enabled: bool) -> None:
    """Set the unrelated generic-memory switch for independence regressions."""

    monkeypatch.setattr(
        system_settings,
        "load_system_settings",
        lambda: SimpleNamespace(memory=SimpleNamespace(enabled=enabled)),
    )


def _run_body(question: str) -> RunCreateRequest:
    return RunCreateRequest(
        input={"messages": [{"role": "user", "content": question}]},
        context={},
        config={"configurable": {}, "context": {}},
    )


@pytest.mark.asyncio
async def test_thread_a_extraction_is_recalled_by_thread_b_and_enters_prompt(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Exercise the complete durable path users depend on across conversations."""

    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
    # Project memory must remain fully functional while generic/personal
    # memory is disabled deployment-wide.
    _set_master_switch(monkeypatch, enabled=False)
    harness = await _make_harness(tmp_path)
    try:
        project = await harness.projects.create(
            "alice", mode="qa", name="火星登陆", background="制定登陆方案"
        )
        await harness.projects.bind_thread(project["id"], "alice", "thread-a")
        await harness.projects.bind_thread(project["id"], "alice", "thread-b")
        harness.checkpointer.set_messages(
            "thread-a",
            [
                HumanMessage(id="message-a-user", content="最终着陆点定在哪里？"),
                AIMessage(
                    id="message-a-assistant",
                    content="已经确认：最终着陆点为乌托邦平原。",
                ),
            ],
        )

        queued = await memory_runtime.enqueue_completed_project_run(
            app=harness.app,
            project=project,
            thread_id="thread-a",
            run_id="run-a",
        )
        assert queued is not None
        assert queued["status"] == "queued"

        [job] = await harness.memories.claim_jobs(
            lease_owner="integration-worker",
            lease_until=datetime.now(UTC) + timedelta(minutes=3),
            limit=1,
        )
        executor = memory_runtime.AnalysisProjectMemoryExecutor(harness.app)
        extraction_llm = AsyncMock(
            return_value={
                "items": [
                    {
                        "logical_key": "landing-site",
                        "kind": "decision",
                        "content": "最终着陆点为乌托邦平原",
                        "confidence": 0.98,
                        "importance": 0.95,
                        "pinned": False,
                    }
                ],
                "snapshot": {"summary": "登陆地点已经确认"},
            }
        )
        monkeypatch.setattr(executor, "_model_extraction", extraction_llm)
        await executor.execute(job, lease_owner="integration-worker")

        extraction_llm.assert_awaited_once()
        stored_items = await harness.memories.list_items(project["id"], "alice")
        stored = next(
            item for item in stored_items if item["logical_key"] == "landing-site"
        )
        assert stored["content"] == "最终着陆点为乌托邦平原"
        assert stored["source_thread_id"] == "thread-a"
        assert stored["source_message_ids"] == [
            "message-a-user",
            "message-a-assistant",
        ]
        assert isinstance(stored["metadata"]["source_order"], float)
        assert any(
            item["kind"] == "episode"
            and item["metadata"].get("raw_project_conversation") is True
            for item in stored_items
        )
        [completed_job] = await harness.memories.list_jobs(project["id"], "alice")
        assert completed_job["status"] == "completed"
        assert completed_job["result"]["fallback"] is False

        with _as_user("alice"):
            prepared, resolved = await project_policy.prepare_project_run(
                _run_body("上次确定的着陆点是什么？"),
                "thread-b",
                harness.request,
            )

        assert resolved is not None and resolved["id"] == project["id"]
        assert prepared.context["project_memory_scope"] == {
            "kind": "analysis_project",
            "project_id": project["id"],
            "owner_user_id": "alice",
            "enabled": True,
        }
        recalled = prepared.context["project_memory"]
        # Revision 1 is the immediate raw-exchange index written when the run
        # completes; the extractor merge advances the durable project state
        # once more.
        assert recalled["revision"] == 2
        assert recalled["items"][0]["content"] == "最终着陆点为乌托邦平原"
        assert recalled["items"][0]["source_thread_id"] == "thread-a"

        prompt = project_system_prompt(prepared.context)
        assert "<project_memory>" in prompt
        assert "最终着陆点为乌托邦平原" in prompt
        assert "同一项目历史对话整理出的背景数据" in prompt
    finally:
        await harness.close()


@pytest.mark.asyncio
async def test_thread_b_recalls_immediate_nickname_while_incremental_job_is_running(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """An ordinary extractor job must not hide its already indexed context."""

    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
    harness = await _make_harness(tmp_path)
    try:
        project = await harness.projects.create(
            "alice", mode="qa", name="昵称记忆测试"
        )
        await harness.projects.bind_thread(project["id"], "alice", "thread-a")
        await harness.projects.bind_thread(project["id"], "alice", "thread-b")
        harness.checkpointer.set_messages(
            "thread-a",
            [
                HumanMessage(
                    id="nickname-user",
                    content="你现在叫铁蛋，我给你起的新名字。",
                ),
                AIMessage(id="nickname-assistant", content="好的。"),
            ],
        )

        queued = await memory_runtime.enqueue_completed_project_run(
            app=harness.app,
            project=project,
            thread_id="thread-a",
            run_id="nickname-run",
        )
        assert queued is not None
        immediate_items = await harness.memories.list_items(project["id"], "alice")
        assert any(
            item["logical_key"] == "assistant.nickname"
            and "铁蛋" in item["content"]
            for item in immediate_items
        )

        [running] = await harness.memories.claim_jobs(
            lease_owner="slow-incremental-worker",
            lease_until=datetime.now(UTC) + timedelta(minutes=3),
            limit=1,
        )
        assert running["id"] == queued["id"]
        state = await harness.memories.get_state(project["id"], "alice")
        assert state is not None and state["status"] == "building"

        with _as_user("alice"):
            prepared, resolved = await project_policy.prepare_project_run(
                _run_body("我给你起的新昵称叫什么？"),
                "thread-b",
                harness.request,
            )

        assert resolved is not None and resolved["id"] == project["id"]
        recalled = prepared.context["project_memory"]
        assert recalled is not None
        assert any("铁蛋" in item["content"] for item in recalled["items"])
        assert "铁蛋" in project_system_prompt(prepared.context)
    finally:
        await harness.close()


@pytest.mark.asyncio
async def test_project_and_owner_boundaries_prevent_memory_leakage(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
    _set_master_switch(monkeypatch, enabled=True)
    harness = await _make_harness(tmp_path)
    try:
        alpha = await harness.projects.create("alice", mode="qa", name="Alpha")
        beta = await harness.projects.create("alice", mode="qa", name="Beta")
        bob_project = await harness.projects.create("bob", mode="qa", name="Bob")
        await harness.projects.bind_thread(alpha["id"], "alice", "alpha-thread")
        await harness.projects.bind_thread(beta["id"], "alice", "beta-thread")
        await harness.projects.bind_thread(bob_project["id"], "bob", "bob-thread")

        await harness.memories.upsert_item(
            alpha["id"],
            "alice",
            logical_key="alpha-secret",
            kind="fact",
            content="Alpha 项目的代号是朱雀",
            source_thread_id="alpha-source",
        )
        await harness.memories.upsert_item(
            beta["id"],
            "alice",
            logical_key="beta-secret",
            kind="fact",
            content="Beta 项目的代号是玄武",
            source_thread_id="beta-source",
        )
        await harness.memories.upsert_item(
            bob_project["id"],
            "bob",
            logical_key="bob-secret",
            kind="fact",
            content="Bob 项目的代号是白虎",
            source_thread_id="bob-source",
        )

        with _as_user("alice"):
            prepared, _ = await project_policy.prepare_project_run(
                _run_body("项目代号是什么？"), "beta-thread", harness.request
            )
        serialized = str(prepared.context["project_memory"])
        assert "Beta 项目的代号是玄武" in serialized
        assert "Alpha 项目的代号是朱雀" not in serialized
        assert "Bob 项目的代号是白虎" not in serialized

        # Repository defence-in-depth requires both project and owner; knowing
        # a project id alone cannot reveal its memory.
        wrong_owner = await harness.memories.recall(
            alpha["id"], "bob", query="朱雀"
        )
        assert wrong_owner["state"] is None
        assert wrong_owner["items"] == []

        # The run-policy boundary also rejects an owner trying to use another
        # user's bound thread before any memory can be injected.
        with _as_user("bob"), pytest.raises(HTTPException) as exc_info:
            await project_policy.prepare_project_run(
                _run_body("Alpha 项目的代号是什么？"),
                "alpha-thread",
                harness.request,
            )
        assert exc_info.value.status_code == 404
    finally:
        await harness.close()


@pytest.mark.asyncio
async def test_generic_memory_switch_off_does_not_block_project_memory(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
    _set_master_switch(monkeypatch, enabled=False)
    harness = await _make_harness(tmp_path)
    try:
        project = await harness.projects.create("alice", mode="qa", name="关闭记忆")
        await harness.projects.bind_thread(project["id"], "alice", "thread-a")
        await harness.projects.bind_thread(project["id"], "alice", "thread-b")
        await harness.memories.upsert_item(
            project["id"],
            "alice",
            logical_key="project-date",
            kind="fact",
            content="项目约定日期为2026年12月31日",
            source_thread_id="thread-a",
        )
        harness.checkpointer.set_messages(
            "thread-a",
            [
                HumanMessage(id="u", content="项目约定日期改为2026年12月31日"),
                AIMessage(id="a", content="已确认项目约定日期为2026年12月31日"),
            ],
        )

        with _as_user("alice"):
            prepared, _ = await project_policy.prepare_project_run(
                _run_body("项目约定日期是哪天？"), "thread-b", harness.request
            )
        assert prepared.context["project_memory_scope"]["enabled"] is True
        assert prepared.context["project_memory"]["items"][0]["content"] == (
            "项目约定日期为2026年12月31日"
        )
        assert "项目约定日期为2026年12月31日" in project_system_prompt(
            prepared.context
        )

        queued = await memory_runtime.enqueue_completed_project_run(
            app=harness.app,
            project=project,
            thread_id="thread-a",
            run_id="run-a",
        )
        assert queued is not None and queued["status"] == "queued"

        class CompletingExecutor:
            app = harness.app

            async def execute(self, job, *, lease_owner):
                await harness.memories.complete_job(
                    job["id"],
                    lease_owner=lease_owner,
                    result={"project_memory_independent": True},
                )

        dispatcher = memory_runtime.AnalysisProjectMemoryDispatcher(
            harness.memories,
            CompletingExecutor(),
            concurrency=1,
        )
        assert await dispatcher.dispatch_once() == 1
        tasks = list(dispatcher._running.values())
        await asyncio.gather(*tasks)
        [completed] = await harness.memories.list_jobs(project["id"], "alice")
        assert completed["id"] == queued["id"]
        assert completed["status"] == "completed"
        assert completed["result"] == {"project_memory_independent": True}
    finally:
        await harness.close()


@pytest.mark.asyncio
async def test_keyword_recall_is_effective_without_vector_encoder(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Vector unavailability must degrade to useful deterministic recall."""

    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
    _set_master_switch(monkeypatch, enabled=True)
    harness = await _make_harness(tmp_path)
    try:
        project = await harness.projects.create("alice", mode="qa", name="预算项目")
        await harness.projects.bind_thread(project["id"], "alice", "query-thread")
        await harness.memories.upsert_item(
            project["id"],
            "alice",
            logical_key="project-budget",
            kind="fact",
            content="The confirmed project budget is 500000 yuan",
            importance=0.2,
            source_thread_id="budget-thread",
        )
        await harness.memories.upsert_item(
            project["id"],
            "alice",
            logical_key="project-venue",
            kind="fact",
            content="The conference venue is Shanghai",
            importance=1.0,
            source_thread_id="venue-thread",
        )

        direct = await harness.memories.recall(
            project["id"],
            "alice",
            query="What is the confirmed project budget in yuan?",
        )
        assert direct["items"][0]["logical_key"] == "project-budget"
        assert direct["items"][0]["recall_signals"]["lexical"] > 0
        assert all(
            item["recall_signals"]["vector"] == 0 for item in direct["items"]
        )
        assert all(
            item["recall_signals"]["vector_compatible"] is False
            for item in direct["items"]
        )

        with _as_user("alice"):
            prepared, _ = await project_policy.prepare_project_run(
                _run_body("What is the confirmed project budget in yuan?"),
                "query-thread",
                harness.request,
            )
        assert prepared.context["project_memory"]["items"][0]["content"] == (
            "The confirmed project budget is 500000 yuan"
        )
        assert "The confirmed project budget is 500000 yuan" in project_system_prompt(
            prepared.context
        )
    finally:
        await harness.close()