"""镜树域 M2 测试。

对应 mirror-tree PRD §14 验收场景（后端可自动化部分）：
1. 同问节三候答并存保留，后续问节绑定实际采用的候答；
2. 采用切换不动归真主脉；归真后切回旧候答正式经历不回滚；分叉归真 409（HJ-1102）；
3. 编辑分叉：旧问题/旧候答/后代仍可访问；
5. 残答封存，续生创建独立节点且不改写残答；
6. 引文快照不被来源更新替换，仅提示后继版本；
7. 移栽为新树，可追溯来源，两树引用不合流；
另覆盖三界：境界归真须 world_id、幻境归真只记录试演、入境事件留痕。
"""

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.runtime.bus import Veto
from app.shenshi_client import EchoShenshi


@pytest.fixture
async def tree_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 _activated_character(client: AsyncClient, name: str) -> str:
    embryo = (
        await client.post("/api/v1/embryo-drafts", json={"source": "original", "name": name})
    ).json()
    return (
        (await client.post(f"/api/v1/embryo-drafts/{embryo['id']}/activate")).json()["character_id"]
    )


async def _wait_run(client: AsyncClient, run_id: str) -> str:
    for _ in range(400):
        status = (await client.get(f"/api/v1/runs/{run_id}")).json()["status"]
        if status in ("completed", "failed", "interrupted", "cancelled"):
            return status
        await asyncio.sleep(0.02)
    raise AssertionError("Run 未到达终态")


async def _new_tree_with_question(client: AsyncClient, content: str, realm: str | None = None):
    character_id = await _activated_character(client, f"树主{asyncio.get_event_loop().time()}")
    tree = (
        await client.post(
            "/api/v1/trees", json={"character_id": character_id, "title": "主题树"}
        )
    ).json()
    q = (
        await client.post(
            f"/api/v1/trees/{tree['id']}/nodes",
            json={"content": content, **({"realm": realm} if realm else {})},
        )
    ).json()
    return tree["id"], q["id"], character_id


async def _answer(client: AsyncClient, question_id: str, instructions: str | None = None):
    spawned = (
        await client.post(f"/api/v1/nodes/{question_id}/answers", json={"instructions": instructions})
    ).json()
    status = await _wait_run(client, spawned["run_id"])
    return spawned["node_id"], status


async def _node_via_path(client: AsyncClient, tree_id: str, node_id: str) -> dict:
    path = (await client.get(f"/api/v1/trees/{tree_id}/path", params={"entry": node_id})).json()
    return next(n for n in path["nodes"] if n["id"] == node_id)


# ── 场景 1：多候答并存 + 绑定采用 ──────────────────────────────────────────────


async def test_three_candidates_and_binding(tree_client: AsyncClient) -> None:
    tree_id, q1, _ = await _new_tree_with_question(tree_client, "第一个问题")
    a1, s1 = await _answer(tree_client, q1, "答法一")
    a2, s2 = await _answer(tree_client, q1, "答法二")
    a3, s3 = await _answer(tree_client, q1, "答法三")
    assert (s1, s2, s3) == ("completed", "completed", "completed")

    # 三候答均保留且已固化
    for node_id in (a1, a2, a3):
        node = await _node_via_path(tree_client, tree_id, node_id)
        assert node["status"] == "solidified"

    # 采用 B 后，新问节默认挂 B（不变量 2）
    await tree_client.post(f"/api/v1/trees/{tree_id}/adopt", json={"node_id": a2})
    q2 = (
        await tree_client.post(f"/api/v1/trees/{tree_id}/nodes", json={"content": "追问"})
    ).json()
    node_q2 = await _node_via_path(tree_client, tree_id, q2["id"])
    assert node_q2["parent_node_id"] == a2


# ── 场景 2：采用/归真分离 + 分叉归真 409 ──────────────────────────────────────────────


async def test_adopt_vs_canonize_and_divergence(tree_client: AsyncClient) -> None:
    tree_id, q1, _ = await _new_tree_with_question(tree_client, "主脉问题")
    a, _ = await _answer(tree_client, q1, "答案 A")
    b, _ = await _answer(tree_client, q1, "答案 B")

    # 采用 B → 归真 B（真界）→ 切回 A：主脉不变
    await tree_client.post(f"/api/v1/trees/{tree_id}/adopt", json={"node_id": b})
    canonized = await tree_client.post(
        f"/api/v1/trees/{tree_id}/canonize",
        json={"endpoint_node_id": b, "canonical_scope": "reality"},
    )
    assert canonized.status_code == 200
    canon_id = canonized.json()["canonization_id"]

    await tree_client.post(f"/api/v1/trees/{tree_id}/adopt", json={"node_id": a})

    # 归真主脉不受采用切换影响（canonization 不可变）
    async with AsyncSessionLocal() as session:
        from sqlalchemy import select

        from app.models.mirror_tree import Canonization

        rows = (await session.execute(select(Canonization))).scalars().all()
        assert len(rows) == 1 and rows[0].id == canon_id
        assert rows[0].endpoint_node_id == b

    # 场景 8：归真与主脉分叉的另一候答 → 409 HJ-1102，附带当前主脉端点
    divergent = await tree_client.post(
        f"/api/v1/trees/{tree_id}/canonize",
        json={"endpoint_node_id": a, "canonical_scope": "reality"},
    )
    assert divergent.status_code == 409
    body = divergent.json()["error"]
    assert body["code"] == "HJ-1102"
    assert body.get("details", {}).get("current_trunk_head") == b

    # 主脉延长（B 之后继续生长再归真）允许
    await tree_client.post(f"/api/v1/trees/{tree_id}/adopt", json={"node_id": b})
    q2 = (
        await client_q(tree_client, tree_id, "延伸问题")
    ).json()
    a_next, _ = await _answer(tree_client, q2["id"])
    extension = await tree_client.post(
        f"/api/v1/trees/{tree_id}/canonize",
        json={"endpoint_node_id": a_next, "canonical_scope": "reality"},
    )
    assert extension.status_code == 200


async def client_q(client: AsyncClient, tree_id: str, content: str):
    return await client.post(f"/api/v1/trees/{tree_id}/nodes", json={"content": content})


# ── 场景 3：编辑分叉 ──────────────────────────────────────────────


async def test_edit_fork_preserves_original(tree_client: AsyncClient) -> None:
    tree_id, q1, _ = await _new_tree_with_question(tree_client, "原始问题")
    a1, _ = await _answer(tree_client, q1)

    fork = (
        await tree_client.post(
            f"/api/v1/nodes/{q1}/edit-fork", json={"new_content": "改写后的问题"}
        )
    ).json()
    assert fork["supersedes_node_id"] == q1

    original = await _node_via_path(tree_client, tree_id, q1)
    forked = await _node_via_path(tree_client, tree_id, fork["node_id"])
    assert original["content"] == "原始问题"  # 原文保留
    assert forked["content"] == "改写后的问题"
    assert forked["parent_node_id"] is None  # 与原问节同级（根）

    # 旧候答仍可访问
    old_answer = await _node_via_path(tree_client, tree_id, a1)
    assert old_answer["status"] == "solidified"


# ── 场景 5：残答封存与续生 ──────────────────────────────────────────────


async def test_residual_and_resume(tree_client: AsyncClient) -> None:
    tree_id, q1, _ = await _new_tree_with_question(tree_client, "会被中断的问题")
    # 用否决监听器制造中断：llm.pre_request 否决 → run.interrupted → 节点封存
    from app.main import app

    kernel: Kernel = app.state.kernel

    async def veto_once(node: str, context) -> None:  # noqa: ANN001
        raise Veto("中断试演", rule="test_interrupt")

    kernel.bus.on("llm.pre_request", veto_once)
    try:
        spawned = (
            await tree_client.post(f"/api/v1/nodes/{q1}/answers", json={})
        ).json()
        status = await _wait_run(tree_client, spawned["run_id"])
        assert status == "interrupted"
    finally:
        kernel.bus._listeners.pop("llm.pre_request", None)  # noqa: SLF001

    residual_id = spawned["node_id"]
    residual = await _node_via_path(tree_client, tree_id, residual_id)
    assert residual["status"] == "sealed_residual"

    # 续生：新独立节点，不改写残答
    resumed = (
        await tree_client.post(f"/api/v1/nodes/{residual_id}/resume", json={})
    ).json()
    assert await _wait_run(tree_client, resumed["run_id"]) == "completed"
    cont = await _node_via_path(tree_client, tree_id, resumed["node_id"])
    assert cont["node_type"] == "continuation"
    assert cont["parent_node_id"] == residual_id
    residual_after = await _node_via_path(tree_client, tree_id, residual_id)
    assert residual_after["status"] == "sealed_residual"  # 残答未被改写
    assert residual_after["id"] == residual_id


# ── 场景 6：引文快照与后继提示 ──────────────────────────────────────────────


async def test_citation_snapshot_and_newer_version(tree_client: AsyncClient) -> None:
    tree_id, q1, _ = await _new_tree_with_question(
        tree_client, "带引文的问题",
    )
    created = await tree_client.post(
        f"/api/v1/nodes/{q1}/citations",
        json={
            "target_type": "pdf_region",
            "locator": {"page": 3, "rect": [0.1, 0.2, 0.5, 0.6]},
            "snapshot": "第 3 页原文 V1",
            "source_id": "doc-001",
            "source_version": 1,
        },
    )
    assert created.status_code == 201

    # 来源正影为 V2（挂在别处）：旧引文不静默替换
    q2 = (await client_q(tree_client, tree_id, "另一个问题")).json()
    await tree_client.post(
        f"/api/v1/nodes/{q2['id']}/citations",
        json={
            "target_type": "pdf_region",
            "locator": {"page": 3},
            "snapshot": "第 3 页原文 V2",
            "source_id": "doc-001",
            "source_version": 2,
        },
    )

    citations = (
        await tree_client.get(f"/api/v1/nodes/{q1}/citations")
    ).json()
    assert len(citations) == 1
    assert citations[0]["snapshot"] == "第 3 页原文 V1"  # 快照保持
    assert citations[0]["newer_version_available"] is True  # 提示后继版本


# ── 场景 7：移栽 ──────────────────────────────────────────────


async def test_transplant_creates_referenced_tree(tree_client: AsyncClient) -> None:
    tree_id, q1, _ = await _new_tree_with_question(tree_client, "被移栽的问题")
    a1, _ = await _answer(tree_client, q1, "答案")
    await tree_client.post(f"/api/v1/trees/{tree_id}/adopt", json={"node_id": a1})
    q2 = (await client_q(tree_client, tree_id, "分支问题")).json()
    a2, _ = await _answer(tree_client, q2["id"], "分支答案")

    transplanted = (
        await tree_client.post(
            f"/api/v1/trees/{tree_id}/transplant",
            json={"node_id": a2, "new_title": "移栽新树"},
        )
    ).json()
    new_tree_id = transplanted["new_tree_id"]

    # 新树保留来源谱系引用；节点是全新 ID（引用关联而非物理合并）
    new_tree = (await tree_client.get(f"/api/v1/trees/{new_tree_id}")).json()
    assert new_tree["transplanted_from"] == {"source_tree_id": tree_id, "source_node_id": a2}
    new_path = (
        await tree_client.get(
            f"/api/v1/trees/{new_tree_id}/path",
            params={"entry": transplanted["working_entry_node_id"]},
        )
    ).json()
    new_ids = {n["id"] for n in new_path["nodes"]}
    assert new_ids.isdisjoint({q1, a1, q2["id"], a2})
    assert new_path["nodes"][-1]["content"] == (await _node_via_path(tree_client, tree_id, a2))["content"]

    # 原树原样保留
    old_path = (
        await tree_client.get(f"/api/v1/trees/{tree_id}/path", params={"entry": a2})
    ).json()
    assert old_path["nodes"][-1]["id"] == a2


# ── 三界作用域 ──────────────────────────────────────────────


async def test_realms_and_scopes(tree_client: AsyncClient) -> None:
    # 境界归真缺 world_id 被拒
    tree_id, q1, _ = await _new_tree_with_question(tree_client, "境界问题", realm="persistent_world")
    a1, _ = await _answer(tree_client, q1)
    missing_world = await tree_client.post(
        f"/api/v1/trees/{tree_id}/canonize",
        json={"endpoint_node_id": a1, "canonical_scope": "persistent_world"},
    )
    assert missing_world.status_code == 400

    ok = await tree_client.post(
        f"/api/v1/trees/{tree_id}/canonize",
        json={
            "endpoint_node_id": a1,
            "canonical_scope": "persistent_world",
            "world_id": "world_demo",
        },
    )
    assert ok.status_code == 200

    # 幻境试演：只记录"发生过这次试演"，与真界归真互不干扰
    tree2, q2, _ = await _new_tree_with_question(tree_client, "幻境问题", realm="simulation")
    sim_answer, _ = await _answer(tree_client, q2)
    sim = await tree_client.post(
        f"/api/v1/trees/{tree2}/canonize",
        json={"endpoint_node_id": sim_answer, "canonical_scope": "simulation"},
    )
    assert sim.status_code == 200

    # 入境事件留痕（界域切换可见）
    async with AsyncSessionLocal() as session:
        from sqlalchemy import select

        from app.models.runtime import Event

        rows = (
            await session.execute(select(Event).where(Event.type == "realm.entered"))
        ).scalars().all()
        assert any(json.loads(r.payload_json).get("tree_id") in (tree_id, tree2) for r in rows)


# ── 芽体保护 ──────────────────────────────────────────────


async def test_budding_protections(tree_client: AsyncClient) -> None:
    tree_id, q1, _ = await _new_tree_with_question(tree_client, "保护问题")
    # 芽体不可采用/归真（另答后立即操作，echo 极快完成——直接测问节与错误类型路径）
    question_adopt = await tree_client.post(
        f"/api/v1/trees/{tree_id}/adopt", json={"node_id": q1}
    )
    assert question_adopt.status_code == 400  # 问节不能作工作入口

    question_canonize = await tree_client.post(
        f"/api/v1/trees/{tree_id}/canonize",
        json={"endpoint_node_id": q1, "canonical_scope": "reality"},
    )
    assert question_canonize.status_code == 400  # 问节不能作归真端点

    # 续思不能基于问节
    bad_continue = await tree_client.post(f"/api/v1/nodes/{q1}/continue", json={})
    assert bad_continue.status_code == 400