"""Memory V2 - HindsightProvider 测试 (mock 客户端,不连真实 Hindsight)。"""

from __future__ import annotations

from deerflow.agents.memory.providers.hindsight import HindsightProvider
from deerflow.config.memory_config import HindsightMemoryConfig


class _FakeResult:
    def __init__(self, text: str, score: float):
        self.text = text
        self.score = score


class _FakeRecallResponse:
    def __init__(self, results):
        self.results = results


class _FakeReflectResponse:
    def __init__(self, text):
        self.text = text


class _FakeClient:
    """模拟 hindsight_client.Hindsight 的异步 API。"""

    def __init__(self):
        self.retained: list[dict] = []
        self.recall_by_bank: dict[str, list[_FakeResult]] = {}
        self.reflect_text = "综合结论"
        self.closed = False

    async def arecall(self, *, bank_id, query, budget, max_tokens):
        return _FakeRecallResponse(self.recall_by_bank.get(bank_id, []))

    async def aretain(self, *, bank_id, content, context=None, metadata=None, document_id=None, retain_async=True):
        self.retained.append(
            {
                "bank_id": bank_id,
                "content": content,
                "metadata": metadata,
                "document_id": document_id,
            }
        )

    async def areflect(self, *, bank_id, query, budget):
        return _FakeReflectResponse(self.reflect_text)

    async def aclose(self):
        self.closed = True


def _make_provider(client: _FakeClient | None = None, **config_kwargs) -> HindsightProvider:
    config = HindsightMemoryConfig(api_url="http://localhost:8765", **config_kwargs)
    provider = HindsightProvider(config)
    provider.initialize(user_id="u1", agent_id="agent-x", thread_id="t1", base_dir="")
    if client is not None:
        # 覆盖客户端工厂,注入 fake —— 绕过真实 hindsight_client 导入。
        provider._client_factory = lambda: client
    return provider


class TestAvailability:
    def test_unavailable_when_package_missing(self, monkeypatch):
        # 显式模拟 hindsight-client 未安装 (不依赖测试环境是否真装了)
        monkeypatch.setattr(
            "deerflow.agents.memory.providers.hindsight.hindsight_client_available",
            lambda: False,
        )
        provider = HindsightProvider(HindsightMemoryConfig(api_url="http://x"))
        assert provider.is_available() is False

    def test_unavailable_when_no_api_url(self, monkeypatch):
        monkeypatch.setattr(
            "deerflow.agents.memory.providers.hindsight.hindsight_client_available",
            lambda: True,
        )
        provider = HindsightProvider(HindsightMemoryConfig(api_url=""))
        assert provider.is_available() is False

    def test_available_when_package_and_url_present(self, monkeypatch):
        monkeypatch.setattr(
            "deerflow.agents.memory.providers.hindsight.hindsight_client_available",
            lambda: True,
        )
        provider = HindsightProvider(HindsightMemoryConfig(api_url="http://localhost:8765"))
        assert provider.is_available() is True


class TestBankResolution:
    def test_user_and_work_banks(self):
        provider = _make_provider()
        assert provider._user_bank == "deerflow-user-u1"
        assert provider._work_bank == "deerflow-work-u1-agent-x"

    def test_empty_placeholder_collapses(self):
        provider = HindsightProvider(
            HindsightMemoryConfig(
                api_url="http://x",
                work_bank_id_template="deerflow-{thread_id}-work",
            )
        )
        provider.initialize(user_id="u1", agent_id="a", thread_id="", base_dir="")
        # 空 thread_id 占位符塌缩,不留多余连字符
        assert "--" not in provider._work_bank
        assert provider._work_bank == "deerflow-work"


class TestDocumentId:
    """document_id 每次 initialize 带时间戳 —— 每轮对话落到各自独立的文档。"""

    def test_different_threads_get_different_doc(self):
        p1 = HindsightProvider(HindsightMemoryConfig(api_url="http://x"))
        p1.initialize(user_id="u", agent_id="a", thread_id="t-1", base_dir="")
        p2 = HindsightProvider(HindsightMemoryConfig(api_url="http://x"))
        p2.initialize(user_id="u", agent_id="a", thread_id="t-2", base_dir="")
        assert p1._document_id != p2._document_id

    def test_doc_id_carries_thread_prefix(self):
        p = HindsightProvider(HindsightMemoryConfig(api_url="http://x"))
        p.initialize(user_id="u", agent_id="a", thread_id="t-123", base_dir="")
        # document_id 形如 "t-123-{时间戳}"
        assert p._document_id.startswith("t-123-")

    def test_no_thread_id_uses_fallback(self):
        p = HindsightProvider(HindsightMemoryConfig(api_url="http://x"))
        p.initialize(user_id="u", agent_id="a", thread_id="", base_dir="")
        # 无 thread 时前缀塌缩成 "thread-"
        assert p._document_id.startswith("thread-")

    def test_retain_writes_thread_document_id(self):
        # sync_turn 落盘时带的 document_id 形如 "t1-{时间戳}"
        client = _FakeClient()
        provider = _make_provider(client)  # thread_id="t1"
        provider.sync_turn("用户说啥", "助手答啥")
        assert client.retained[0]["document_id"].startswith("t1-")

    def test_recreated_providers_get_distinct_documents(self):
        # 同一 thread 连续 3 轮、每轮 provider 重建 —— 各自落到独立文档
        doc_ids: set[str] = set()
        for turn in range(1, 4):
            client = _FakeClient()
            provider = _make_provider(client)  # 每轮重建,thread_id 固定 t1
            provider.sync_turn(f"u{turn}", f"a{turn}")
            doc_ids.add(client.retained[0]["document_id"])
        assert len(doc_ids) == 3


class TestPrefetchRecall:
    def test_prefetch_merges_two_banks(self):
        client = _FakeClient()
        client.recall_by_bank["deerflow-user-u1"] = [_FakeResult("用户事实", 0.9)]
        client.recall_by_bank["deerflow-work-u1-agent-x"] = [_FakeResult("工作笔记", 0.7)]
        provider = _make_provider(client)
        result = provider.prefetch("查询")
        assert "用户事实" in result
        assert "工作笔记" in result
        # 高 score 在前
        assert result.index("用户事实") < result.index("工作笔记")

    def test_prefetch_dedupes(self):
        client = _FakeClient()
        client.recall_by_bank["deerflow-user-u1"] = [_FakeResult("同一条", 0.9)]
        client.recall_by_bank["deerflow-work-u1-agent-x"] = [_FakeResult("同一条", 0.5)]
        provider = _make_provider(client)
        result = provider.prefetch("查询")
        assert result.count("同一条") == 1

    def test_prefetch_empty_query_returns_empty(self):
        provider = _make_provider(_FakeClient())
        assert provider.prefetch("") == ""

    def test_prefetch_skipped_in_tools_mode(self):
        client = _FakeClient()
        client.recall_by_bank["deerflow-user-u1"] = [_FakeResult("x", 0.9)]
        provider = _make_provider(client, memory_mode="tools")
        assert provider.prefetch("查询") == ""

    def test_prefetch_skipped_when_auto_recall_off(self):
        client = _FakeClient()
        client.recall_by_bank["deerflow-user-u1"] = [_FakeResult("x", 0.9)]
        provider = _make_provider(client, auto_recall=False)
        assert provider.prefetch("查询") == ""

    def test_prefetch_no_results_returns_empty(self):
        provider = _make_provider(_FakeClient())
        assert provider.prefetch("查询") == ""

    def test_recall_returns_list(self):
        client = _FakeClient()
        client.recall_by_bank["deerflow-user-u1"] = [_FakeResult("甲", 0.9)]
        provider = _make_provider(client)
        assert provider.recall("查询") == ["甲"]


class TestSyncTurnRetain:
    def test_sync_turn_retains_to_work_bank(self):
        client = _FakeClient()
        provider = _make_provider(client)
        provider.sync_turn("用户说啥", "助手答啥")
        assert len(client.retained) == 1
        assert client.retained[0]["bank_id"] == "deerflow-work-u1-agent-x"
        assert "用户说啥" in client.retained[0]["content"]

    def test_sync_turn_skipped_when_auto_retain_off(self):
        client = _FakeClient()
        provider = _make_provider(client, auto_retain=False)
        provider.sync_turn("u", "a")
        assert client.retained == []

    def test_retain_every_n_turns_internal_counter(self):
        # 不传 turn_index 时退回进程内计数器 (同一 provider 复用的场景)
        client = _FakeClient()
        provider = _make_provider(client, retain_every_n_turns=2)
        provider.sync_turn("u1", "a1")  # 第 1 轮,缓冲
        assert client.retained == []
        provider.sync_turn("u2", "a2")  # 第 2 轮,触发
        assert len(client.retained) == 1


class TestRetainEveryNTurnsAcrossRecreation:
    """provider 每轮重建时,retain_every_n_turns 的节流必须靠外部 turn_index。"""

    def test_internal_counter_breaks_across_recreation(self):
        # 文档化 bug 根因:每个新 provider 计数器从 0→1,1 % 3 != 0 → 永不触发
        retained_total = 0
        for turn in range(1, 7):  # 6 轮
            client = _FakeClient()
            provider = _make_provider(client, retain_every_n_turns=3)
            provider.sync_turn(f"u{turn}", f"a{turn}")  # 不传 turn_index
            retained_total += len(client.retained)
        assert retained_total == 0

    def test_explicit_turn_index_drives_retain(self):
        # 修复:turn_index 由编排层按对话历史算出并传入,节流恢复正常
        retained_total = 0
        for turn in range(1, 7):  # 6 轮,retain_every_n_turns=3
            client = _FakeClient()
            provider = _make_provider(client, retain_every_n_turns=3)
            provider.sync_turn(f"u{turn}", f"a{turn}", turn_index=turn)
            retained_total += len(client.retained)
        # 第 3、6 轮触发 → 共 2 次
        assert retained_total == 2

    def test_turn_index_every_one_retains_each_turn(self):
        # retain_every_n_turns=1 时每轮都写
        for turn in range(1, 4):
            client = _FakeClient()
            provider = _make_provider(client, retain_every_n_turns=1)
            provider.sync_turn("u", "a", turn_index=turn)
            assert len(client.retained) == 1


class TestClientLifecycle:
    """客户端必须即开即关 —— 杜绝 aiohttp 'Unclosed client session'。"""

    def test_client_closed_after_recall(self):
        client = _FakeClient()
        provider = _make_provider(client)
        provider.recall("查询")
        assert client.closed is True

    def test_client_closed_after_retain(self):
        client = _FakeClient()
        provider = _make_provider(client)
        provider.sync_turn("u", "a")
        assert client.closed is True

    def test_client_closed_after_reflect(self):
        client = _FakeClient()
        provider = _make_provider(client)
        provider.reflect("问题")
        assert client.closed is True

    def test_client_closed_even_on_failure(self):
        class _BrokenClient(_FakeClient):
            async def arecall(self, **kwargs):
                raise RuntimeError("Hindsight 不可达")

        client = _BrokenClient()
        provider = _make_provider(client)
        provider.prefetch("查询")  # 失败被吞
        assert client.closed is True

    def test_shutdown_is_noop(self):
        # 无缓存客户端,shutdown 是 no-op,不应抛异常
        provider = HindsightProvider(HindsightMemoryConfig())
        provider.shutdown()


class TestOnMemoryWrite:
    def test_user_target_goes_to_user_bank(self):
        client = _FakeClient()
        provider = _make_provider(client)
        provider.on_memory_write("add", "user", "用户偏好")
        assert client.retained[0]["bank_id"] == "deerflow-user-u1"

    def test_memory_target_goes_to_work_bank(self):
        client = _FakeClient()
        provider = _make_provider(client)
        provider.on_memory_write("add", "memory", "工作笔记")
        assert client.retained[0]["bank_id"] == "deerflow-work-u1-agent-x"

    def test_remove_action_not_mirrored(self):
        client = _FakeClient()
        provider = _make_provider(client)
        provider.on_memory_write("remove", "user", "某条")
        assert client.retained == []

    def test_empty_content_not_mirrored(self):
        client = _FakeClient()
        provider = _make_provider(client)
        provider.on_memory_write("add", "user", "   ")
        assert client.retained == []


class TestReflect:
    def test_reflect_returns_text(self):
        client = _FakeClient()
        client.reflect_text = "这是综合结论"
        provider = _make_provider(client)
        assert provider.reflect("问题") == "这是综合结论"


class TestSessionSwitch:
    def test_session_switch_resets_work_bank_and_batch(self):
        provider = _make_provider(_FakeClient())
        old_doc = provider._document_id
        provider.on_session_switch("t2")
        assert provider._thread_id == "t2"
        assert provider._turn_counter == 0
        assert provider._document_id != old_doc


class TestMisc:
    def test_name(self):
        assert HindsightProvider(HindsightMemoryConfig()).name == "hindsight"

    def test_get_tool_schemas_empty(self):
        # Hindsight 工具走 LangChain BaseTool,不经 manager
        assert HindsightProvider(HindsightMemoryConfig()).get_tool_schemas() == []

    def test_context_mode_no_system_prompt(self):
        provider = _make_provider(memory_mode="context")
        assert provider.system_prompt_block() == ""

    def test_tools_mode_has_system_prompt(self):
        provider = _make_provider(memory_mode="tools")
        assert "hindsight_recall" in provider.system_prompt_block()

    def test_recall_failure_returns_empty(self):
        class _BrokenClient(_FakeClient):
            async def arecall(self, **kwargs):
                raise RuntimeError("Hindsight 不可达")

        provider = _make_provider(_BrokenClient())
        # 召回失败不抛异常,返回空串
        assert provider.prefetch("查询") == ""