"""息壤（M3）测试：对应 rewrite-plan M3 六条验收。

1. 未归真旁枝自动入忆率 = 0（只有归真/手动入忆产生候选）；
2. 重忆未确认写入被拒（HJ-1201 语义：sensitive→pending，仅 confirm 后 active）；
3. deny 两次退出检索；
4. 正影后旧版停止普通召回（superseded），内容不改写；
5. 删除后全链路停用（墓碑无原文、检索/注入不出现）；
6. 召回带来源可展开（source_refs 指向归真记录）；归真触发实录与提取候选。
"""

from __future__ import annotations

import asyncio
import json
from collections.abc import AsyncGenerator

import httpx
import pytest
from httpx import ASGITransport, AsyncClient

from app.db.session import AsyncSessionLocal
from app.domains.jingsheng import bind_event_store
from app.kernel import Kernel, KernelEventLog
from app.services.event_store import DbEventStore
from app.shenshi_client import EchoShenshi


@pytest.fixture
async def client(
    monkeypatch: pytest.MonkeyPatch, auth_headers: dict[str, str]
) -> AsyncGenerator[AsyncClient, None]:
    from app.main import app

    event_store = DbEventStore(AsyncSessionLocal)
    bind_event_store(event_store)
    app.state.kernel = Kernel(
        shenshi=EchoShenshi(),
        session_factory=AsyncSessionLocal,
        event_log=KernelEventLog(event_store),
        timeout_seconds=10.0,
    )
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test", headers=auth_headers) as c:
        yield c
    app.state.kernel = None


async def _tree_with_answer(client: AsyncClient, title: str) -> tuple[str, str, str, str]:
    """建树→问→答→返回 (tree_id, q, a, character_id)。"""
    embryo = (
        await client.post("/api/v1/embryo-drafts", json={"source": "original", "name": f"忆灵{title}"})
    ).json()
    character_id = (
        (await client.post(f"/api/v1/embryo-drafts/{embryo['id']}/activate")).json()["character_id"]
    )
    tree = (
        await client.post("/api/v1/trees", json={"character_id": character_id, "title": title})
    ).json()
    q = (await client.post(f"/api/v1/trees/{tree['id']}/nodes", json={"content": "你记得什么"})).json()
    spawned = (await client.post(f"/api/v1/nodes/{q['id']}/answers", json={})).json()
    for _ in range(300):
        status = (await client.get(f"/api/v1/runs/{spawned['run_id']}")).json()["status"]
        if status in ("completed", "interrupted", "failed"):
            break
        await asyncio.sleep(0.02)
    return tree["id"], q["id"], spawned["node_id"], character_id


async def _memories(client: AsyncClient, character_id: str) -> list[dict]:
    return (await client.get(f"/api/v1/characters/{character_id}/memories")).json()


async def _canonize(client: AsyncClient, tree_id: str, endpoint: str, scope: str = "reality") -> dict:
    resp = await client.post(
        f"/api/v1/trees/{tree_id}/canonize",
        json={"endpoint_node_id": endpoint, "canonical_scope": scope},
    )
    assert resp.status_code == 200, resp.text
    return resp.json()


async def _wait_candidate(client: AsyncClient, character_id: str, timeout_s: float = 5.0) -> list[dict]:
    """等待归真后的异步提取落库。"""
    for _ in range(int(timeout_s / 0.05)):
        items = await _memories(client, character_id)
        if items:
            return items
        await asyncio.sleep(0.05)
    return []


# ── 验收 1：未归真不入忆 ──────────────────────────────────────────────


async def test_uncanonized_branch_never_memorized(client: AsyncClient) -> None:
    tree_id, _q, answer_id, character_id = await _tree_with_answer(client, "旁支树")
    # 未归真：多次另答也不产生任何记忆
    for _ in range(2):
        spawned = (await client.post(f"/api/v1/nodes/{_q}/answers", json={})).json()
        for _ in range(200):
            if (await client.get(f"/api/v1/runs/{spawned['run_id']}")).json()["status"] in (
                "completed",
                "interrupted",
            ):
                break
            await asyncio.sleep(0.02)
    await asyncio.sleep(0.2)
    assert await _memories(client, character_id) == []  # 自动入忆率 = 0


# ── 验收 6+管线：归真→实录+候选；重敏 pending ──────────────────────────────────────────────


async def test_canonization_creates_factual_record_and_candidates(client: AsyncClient) -> None:
    tree_id, _q, answer_id, character_id = await _tree_with_answer(client, "归真树")
    await _canonize(client, tree_id, answer_id)

    # 实录（同步落库）
    records = (await client.get(f"/api/v1/characters/{character_id}/factual-records")).json()
    assert len(records) == 1
    assert records[0]["evidence"]["path_nodes"]

    # 候选（异步提取；EchoShenshi 无 JSON → 确定性回落单条 episodic 候选）
    items = await _wait_candidate(client, character_id)
    assert len(items) == 1
    memory = items[0]
    assert memory["tier"] == "explicit"
    assert memory["source"] == "self_inferred"
    assert memory["status"] in ("candidate", "active")
    # 召回透明：来源指向归真记录，可展开
    assert memory["source_refs"]


# ── 验收 2：重忆未确认不入 active ──────────────────────────────────────────────


async def test_sensitive_requires_confirmation(client: AsyncClient) -> None:
    from app.domains.xirang import service as xirang_service

    tree_id, _q, answer_id, character_id = await _tree_with_answer(client, "重忆树")
    await _canonize(client, tree_id, answer_id)
    await _wait_candidate(client, character_id)

    # 直接放置一条重忆候选（提取管线同路径）
    placed = await xirang_service.place_candidates(
        character_id=character_id,
        candidates=[{"content": "用户的健康隐私：对花粉过敏", "type": "persona", "sensitive": True}],
        source_refs=["canon_x"],
        source="self_inferred",
    )
    sensitive = next(m for m in placed if m.tier == "sensitive")
    assert sensitive.status == "pending"  # 未确认绝不 active

    # 检索不召回 pending
    recalled = await xirang_service.recall_for_generation(character_id, "花粉 过敏")
    assert all(m.id != sensitive.id for m in recalled)

    # confirm 后进入检索
    resp = await client.post(f"/api/v1/memories/{sensitive.id}/confirm")
    assert resp.status_code == 200 and resp.json()["status"] == "active"
    recalled = await xirang_service.recall_for_generation(character_id, "花粉 过敏")
    assert any(m.id == sensitive.id for m in recalled)


# ── 验收 3+5：deny/删除全链路停用 ──────────────────────────────────────────────


async def test_deny_and_delete_exit_everywhere(client: AsyncClient) -> None:
    from app.domains.xirang import service as xirang_service

    tree_id, _q, answer_id, character_id = await _tree_with_answer(client, "删除树")
    await _canonize(client, tree_id, answer_id)
    items = await _wait_candidate(client, character_id)
    memory_id = items[0]["id"]

    denied = await client.post(f"/api/v1/memories/{memory_id}/deny")
    assert denied.status_code == 200 and denied.json()["exited_retrieval"] is True

    # 墓碑：列表可见状态但无原文；检索不出现
    listed = await _memories(client, character_id)
    tomb = next(m for m in listed if m["id"] == memory_id)
    assert tomb["status"] == "tombstoned" and tomb["content"] == ""
    recalled = await xirang_service.recall_for_generation(character_id, "共同经历")
    assert all(m.id != memory_id for m in recalled)


# ── 验收 4：正影旧版停召、内容不改写 ──────────────────────────────────────────────


async def test_correction_supersedes_old_version(client: AsyncClient) -> None:
    from app.domains.xirang import service as xirang_service

    tree_id, _q, answer_id, character_id = await _tree_with_answer(client, "正影树")
    await _canonize(client, tree_id, answer_id)
    items = await _wait_candidate(client, character_id)
    memory_id = items[0]["id"]

    corrected = await client.post(
        f"/api/v1/memories/{memory_id}/correct", json={"new_content": "正影后的准确版本"}
    )
    assert corrected.status_code == 200
    new_id = corrected.json()["id"]
    assert corrected.json()["prev_version_id"] == memory_id

    # 旧版 superseded 且不再被召回；新版生效；旧内容保留在行内（不改写历史）
    recalled = await xirang_service.recall_for_generation(character_id, "正影")
    ids = [m.id for m in recalled]
    assert new_id in ids and memory_id not in ids
    async with AsyncSessionLocal() as session:
        from app.models.xirang import Memory

        old = await session.get(Memory, memory_id)
        assert old.status == "superseded"
        assert "共同经历" in old.content  # 原文未被改写


# ── 手动入忆：必须标注类型 ──────────────────────────────────────────────


async def test_manual_memorize_requires_label(client: AsyncClient) -> None:
    tree_id, q, answer_id, _ = await _tree_with_answer(client, "手记树")
    bad = await client.post(f"/api/v1/nodes/{q}/memorize", json={"label": "unknown"})
    assert bad.status_code == 422 or bad.status_code == 400

    ok = await client.post(f"/api/v1/nodes/{q}/memorize", json={"label": "insight"})
    assert ok.status_code == 201
    body = ok.json()
    assert body["type"] == "fact_candidate"  # 未归真内容只作事实候选并显式标注
    assert body["content"].startswith("[insight]")


# ── 注入：获准记忆进入投影 prompt ──────────────────────────────────────────────


async def test_recalled_memory_injected_into_prompt(client: AsyncClient) -> None:
    tree_id, _q, answer_id, character_id = await _tree_with_answer(client, "注入树")
    await _canonize(client, tree_id, answer_id)
    await _wait_candidate(client, character_id)

    # 新问节触发另答 → llm.requested 的 prompt 含【获准记忆】段
    await client.post(f"/api/v1/trees/{tree_id}/adopt", json={"node_id": answer_id})
    q2 = (
        await client.post(f"/api/v1/trees/{tree_id}/nodes", json={"content": "再谈谈"})
    ).json()
    spawned = (await client.post(f"/api/v1/nodes/{q2['id']}/answers", json={})).json()
    for _ in range(300):
        if (await client.get(f"/api/v1/runs/{spawned['run_id']}")).json()["status"] in (
            "completed",
            "interrupted",
        ):
            break
        await asyncio.sleep(0.02)
    events = (await client.get(f"/api/v1/runs/{spawned['run_id']}/events")).json()
    requested = next(e for e in events if e["type"] == "llm.requested")
    prompt = requested["payload"]["request"]["messages"][-1]["content"]
    assert "【获准记忆（来源可查）】" in prompt
    _ = json  # 保留 json 引用供调试断言使用


# ── M3b：世谱激活进入投影 ──────────────────────────────────────────────


async def test_lorebook_activation_injected_into_prompt(client: AsyncClient) -> None:
    tree_id, _q, answer_id, character_id = await _tree_with_answer(client, "世谱树")

    # 为人物建世谱书 + 关键词条目 + 常驻条目
    book = (
        await client.post(
            "/api/v1/lorebooks",
            json={"name": "测试世谱", "character_id": character_id},
        )
    )
    if book.status_code >= 400:  # 端点形态兼容：老 CRUD 可能要求别的字段
        book = (await client.post("/api/v1/lorebooks", json={"name": "测试世谱"}))
        assert book.status_code < 400, book.text
    book_id = book.json()["id"]

    keyed = (
        await client.post(
            f"/api/v1/lorebooks/{book_id}/entries",
            json={"keys": ["星穹图书馆"], "content": "星穹图书馆只在午后开放"},
        )
    )
    if keyed.status_code >= 400:  # 兼容批量/单条两种端点形态
        keyed = await client.post(
            f"/api/v1/lorebooks/{book_id}/entries",
            json=[{"keys": ["星穹图书馆"], "content": "星穹图书馆只在午后开放"}],
        )
    assert keyed.status_code < 400, keyed.text

    await client.post(f"/api/v1/trees/{tree_id}/adopt", json={"node_id": answer_id})
    q2 = (
        await client.post(
            f"/api/v1/trees/{tree_id}/nodes", json={"content": "带我去星穹图书馆看看"}
        )
    ).json()
    spawned = (await client.post(f"/api/v1/nodes/{q2['id']}/answers", json={})).json()
    for _ in range(300):
        if (await client.get(f"/api/v1/runs/{spawned['run_id']}")).json()["status"] in (
            "completed",
            "interrupted",
        ):
            break
        await asyncio.sleep(0.02)
    events = (await client.get(f"/api/v1/runs/{spawned['run_id']}/events")).json()
    requested = next(e for e in events if e["type"] == "llm.requested")
    prompt = requested["payload"]["request"]["messages"][-1]["content"]
    assert "【世界认知（世谱命中）】" in prompt
    assert "星穹图书馆只在午后开放" in prompt