"""AC-4 端到端：AI 对话（聊天记录）持久化（真实 pg，**零 LLM / 无网关**）。

本功能是纯 CRUD 旁路留痕（`docs/design/ai-chat-history-plan.md`）——五个生成端点逐字节
不变、只读；作者↔AI 每轮往复经 **`POST /projects/{id}/ai-messages`** 显式 append，供每章
「AI 对话」抽屉重放。故这里**不** override 任何网关（无 `_FakeAdapter`）：只打两端点、断
DB 真源。镜像 `tests/test_t6_toolbox_e2e.py` 的 `e2e_sm` engine/session 生命周期；无 pg → skip。

覆盖计划 §6 的 API 级用例：
1. 缓冲多轮线程：一批落一次「已定的 refine+clarify 交换」（作者含糊→AI 反问→作者答→AI 定稿），
   同 `thread_id`、`seq` 0..n；GET newest-first、`seq`/`role`/`thread_id`/`meta` 往复保真。
2. 作用域分区：章级线程(chapter_no=N) ∪ 项目级(chapter_no=null) 生成——GET `?chapter_no=N`
   两者都返回；GET 缺省（项目级）只返回项目级。
3. newest-first + 分页（跨多批已提交）+ `kind` 过滤。
4. append-only / read-only：append 只写 `ai_messages`（其它业务表零变化、`usage_ledger` 恒 0
   证明无网关）；GET 只读不建行。
5. 未知项目 → 404；非法 role/kind、空 messages、超上限 content/meta → 422（失败零落库）。
6. clarify-abandon 的 API 级代理：服务端无部分写路径——缓冲在客户端，端点只持久化被显式
   POST 的已定线程；从未 POST 的放弃线程在库里不留痕。
"""

from __future__ import annotations

import asyncio
import uuid
from collections.abc import AsyncIterator

import httpx
import pytest
from asgi_lifespan import LifespanManager
from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from ww_db import get_sessionmaker
from ww_db.models import (
    AiMessage,
    Chapter,
    ChapterReview,
    Job,
    Project,
    UsageLedger,
    WorldEntity,
)

# content 超上限探针 = 手稿上限 + 1（AI_MESSAGE_CONTENT_MAX = 200_000）。
_CONTENT_CAP = 200_000
# meta 序列化上界（_META_MAX_SERIALIZED = 50_000）——用超此长度的字符串塞 meta 触发 422。
_META_CAP = 50_000


@pytest.fixture
async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
    """真实 DB session 工厂；无 pg 时跳过（每测试清缓存重建 engine、结束 dispose）。"""
    get_sessionmaker.cache_clear()
    maker = get_sessionmaker()
    try:
        async with maker() as probe:
            await probe.execute(select(1))
    except Exception:
        pytest.skip("postgres not reachable")
    yield maker
    await maker.kw["bind"].dispose()
    get_sessionmaker.cache_clear()


async def _count(sm: async_sessionmaker[AsyncSession], model: type, project_uuid: uuid.UUID) -> int:
    async with sm() as s:
        return int(
            (
                await s.execute(
                    select(func.count()).select_from(model).where(model.project_id == project_uuid)  # type: ignore[attr-defined]
                )
            ).scalar_one()
        )


async def _cleanup(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
    """按 FK 顺序清理（ai_messages 虽 CASCADE，显式删更清晰）。"""
    async with sm() as cleanup:
        await cleanup.execute(delete(AiMessage).where(AiMessage.project_id == project_uuid))
        await cleanup.execute(delete(Project).where(Project.id == project_uuid))
        await cleanup.commit()


async def _create_project(client: httpx.AsyncClient, title: str) -> uuid.UUID:
    resp = await client.post("/projects", json={"title": title, "genre": "玄幻"})
    assert resp.status_code == 201, resp.text
    return uuid.UUID(resp.json()["id"])


async def test_buffered_multi_turn_thread_persists_and_lists_newest_first(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 1/4：一批落「已定的 refine+clarify 交换」→ GET newest-first，seq/role/meta 保真。"""
    from ww_api.main import create_app

    app = create_app()
    transport = httpx.ASGITransport(app=app)
    thread_id = uuid.uuid4()
    original_segment = "他站在雨里，一动不动。"
    options = [
        {"label": "保留第一人称", "value": "keep_first"},
        {"label": "改第三人称", "value": "third"},
    ]
    project_uuid: uuid.UUID | None = None
    try:
        async with LifespanManager(app):
            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
                project_uuid = await _create_project(client, "AC-4 缓冲线程作品")

                # 一批落库：作者(含糊)→AI(反问)→作者(答)→AI(定稿)；选项/version_no/segment 进 meta。
                resp = await client.post(
                    f"/projects/{project_uuid}/ai-messages",
                    json={
                        "thread_id": str(thread_id),
                        "chapter_no": 5,
                        "kind": "refine",
                        "messages": [
                            {"role": "author", "content": "帮我把这段润色得更有张力", "meta": {}},
                            {
                                "role": "ai",
                                "content": "你希望保留原文的第一人称视角吗？",
                                "meta": {
                                    "parent_kind": "refine",
                                    "options": options,
                                    "allow_free_text": True,
                                    "verification": "视角一致性",
                                },
                            },
                            {"role": "author", "content": "保留第一人称", "meta": {}},
                            {
                                "role": "ai",
                                "content": "雨丝斜织，他钉在原地，任凭寒意浸透脊背。",
                                "meta": {"version_no": 1, "segment": original_segment},
                            },
                        ],
                    },
                )
                assert resp.status_code == 201, resp.text
                appended = resp.json()["messages"]
                # 201 回显：4 条、seq 0..3（repo 赋值）、同 thread_id、同 chapter_no。
                assert [m["seq"] for m in appended] == [0, 1, 2, 3]
                assert {m["thread_id"] for m in appended} == {str(thread_id)}
                assert {m["chapter_no"] for m in appended} == {5}
                assert all(m["id"] for m in appended)  # id/created_at 已回填供乐观对账。

                # GET newest-first：同批共享 created_at → 靠 seq DESC 定序 → [3,2,1,0]。
                listed = await client.get(f"/projects/{project_uuid}/ai-messages?chapter_no=5")
                assert listed.status_code == 200
                rows = listed.json()["messages"]
                assert [r["seq"] for r in rows] == [3, 2, 1, 0]
                assert [r["role"] for r in rows] == ["ai", "author", "ai", "author"]
                assert {r["thread_id"] for r in rows} == {str(thread_id)}
                # meta 往复保真：AI 反问的选项列表 + AI 定稿的 version_no/segment。
                ai_question = next(r for r in rows if r["seq"] == 1)
                assert ai_question["meta"]["options"] == options
                assert ai_question["meta"]["allow_free_text"] is True
                ai_final = next(r for r in rows if r["seq"] == 3)
                assert ai_final["meta"]["version_no"] == 1
                # segment 存于 AI 定稿行 → 抽屉「回填选段」可重锚（计划 §3.4/§4.5）。
                assert ai_final["meta"]["segment"] == original_segment
                # 抽屉 re-accept 直读全文 content（未截断）。
                assert ai_final["content"] == "雨丝斜织，他钉在原地，任凭寒意浸透脊背。"

        # DB 真源：恰 4 行。
        assert project_uuid is not None
        assert await _count(e2e_sm, AiMessage, project_uuid) == 4
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_scope_partition_chapter_union_and_project_only(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 2/6：章级 ∪ 项目级(NULL)——GET `?chapter_no=N` 两者都返回；缺省只返回项目级。"""
    from ww_api.main import create_app

    app = create_app()
    transport = httpx.ASGITransport(app=app)
    chapter_thread = uuid.uuid4()
    project_thread = uuid.uuid4()
    project_uuid: uuid.UUID | None = None
    try:
        async with LifespanManager(app):
            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
                project_uuid = await _create_project(client, "AC-4 作用域分区作品")

                # A) 章级线程（rewrite，chapter_no=3）。
                a = await client.post(
                    f"/projects/{project_uuid}/ai-messages",
                    json={
                        "thread_id": str(chapter_thread),
                        "chapter_no": 3,
                        "kind": "rewrite",
                        "messages": [
                            {
                                "role": "author",
                                "content": "整章节奏太平",
                                "meta": {"version_no": 1},
                            },
                            {
                                "role": "ai",
                                "content": "<重写后的整章正文>",
                                "meta": {"version_no": 1},
                            },
                        ],
                    },
                )
                assert a.status_code == 201, a.text

                # B) 项目级工具箱生成（generator，chapter_no 缺省 = NULL）。
                b = await client.post(
                    f"/projects/{project_uuid}/ai-messages",
                    json={
                        "thread_id": str(project_thread),
                        "kind": "generator",
                        "tool_key": "brainstorm",
                        "messages": [
                            {
                                "role": "author",
                                "content": "想要三个突破常规的脑洞",
                                "meta": {
                                    "tool_key": "brainstorm",
                                    "input_fields": {"brief": "..."},
                                },
                            },
                            {
                                "role": "ai",
                                "content": "1. 天降系统…\n2. 转生灵宠…\n3. 末世种田…",
                                "meta": {"tool_key": "brainstorm", "output_kind": "IdeaListResult"},
                            },
                        ],
                    },
                )
                assert b.status_code == 201, b.text
                assert all(m["chapter_no"] is None for m in b.json()["messages"])

                # GET ?chapter_no=3 → 本章 ∪ 项目级 = A + B（4 行，两线程都在）。
                union = (
                    await client.get(f"/projects/{project_uuid}/ai-messages?chapter_no=3")
                ).json()["messages"]
                assert len(union) == 4
                assert {r["thread_id"] for r in union} == {
                    str(chapter_thread),
                    str(project_thread),
                }
                assert {r["chapter_no"] for r in union} == {3, None}

                # GET 缺省（无 chapter_no）→ 仅项目级 B（2 行，chapter_no 全 NULL）。
                project_only = (await client.get(f"/projects/{project_uuid}/ai-messages")).json()[
                    "messages"
                ]
                assert len(project_only) == 2
                assert {r["thread_id"] for r in project_only} == {str(project_thread)}
                assert all(r["chapter_no"] is None for r in project_only)

                # GET 另一无对话的章 → union 仍带出项目级 B（项目级现于每章抽屉，计划 §6-(6)）。
                other_chapter = (
                    await client.get(f"/projects/{project_uuid}/ai-messages?chapter_no=99")
                ).json()["messages"]
                assert {r["thread_id"] for r in other_chapter} == {str(project_thread)}
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_newest_first_pagination_and_kind_filter_across_batches(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 3：跨多批已提交 → newest-first + 分页 + `kind` 过滤。"""
    from ww_api.main import create_app

    app = create_app()
    transport = httpx.ASGITransport(app=app)
    continue_thread = uuid.uuid4()
    refine_thread = uuid.uuid4()
    project_uuid: uuid.UUID | None = None
    try:
        async with LifespanManager(app):
            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
                project_uuid = await _create_project(client, "AC-4 分页过滤作品")

                async def _post(body: dict) -> None:
                    resp = await client.post(f"/projects/{project_uuid}/ai-messages", json=body)
                    assert resp.status_code == 201, resp.text
                    # 分批提交须 created_at 单调递增 → 睡极短保跨批定序确定（同批仍靠 seq）。
                    await asyncio.sleep(0.02)

                # 批1：续写候选0（ai-only，续写无作者输入）。
                await _post(
                    {
                        "thread_id": str(continue_thread),
                        "chapter_no": 7,
                        "kind": "continue",
                        "tool_key": "continue",
                        "messages": [
                            {"role": "ai", "content": "候选一…", "meta": {"candidate_index": 0}}
                        ],
                    }
                )
                # 批2：同线程「再来一个」候选1。
                await _post(
                    {
                        "thread_id": str(continue_thread),
                        "chapter_no": 7,
                        "kind": "continue",
                        "tool_key": "continue",
                        "messages": [
                            {"role": "ai", "content": "候选二…", "meta": {"candidate_index": 1}}
                        ],
                    }
                )
                # 批3：refine 两条（最新批）。
                await _post(
                    {
                        "thread_id": str(refine_thread),
                        "chapter_no": 7,
                        "kind": "refine",
                        "messages": [
                            {"role": "author", "content": "再紧凑些", "meta": {}},
                            {
                                "role": "ai",
                                "content": "定稿…",
                                "meta": {"version_no": 1, "segment": "原段"},
                            },
                        ],
                    }
                )

                # newest-first：批3(created_at 最大) → 批2 → 批1；批内 seq DESC。
                rows = (
                    await client.get(f"/projects/{project_uuid}/ai-messages?chapter_no=7")
                ).json()["messages"]
                assert [r["kind"] for r in rows] == [
                    "refine",  # 批3 seq1 (ai 定稿)
                    "refine",  # 批3 seq0 (author)
                    "continue",  # 批2
                    "continue",  # 批1
                ]
                assert [r["role"] for r in rows] == ["ai", "author", "ai", "ai"]

                # 分页：limit=1 取最新一行；offset 前移。
                page0 = (
                    await client.get(f"/projects/{project_uuid}/ai-messages?chapter_no=7&limit=1")
                ).json()["messages"]
                assert len(page0) == 1 and page0[0]["content"] == "定稿…"
                page1 = (
                    await client.get(
                        f"/projects/{project_uuid}/ai-messages?chapter_no=7&limit=1&offset=1"
                    )
                ).json()["messages"]
                assert len(page1) == 1 and page1[0]["role"] == "author"

                # kind 过滤：只取 continue（批2、批1），candidate_index newest-first [1,0]。
                only_continue = (
                    await client.get(
                        f"/projects/{project_uuid}/ai-messages?chapter_no=7&kind=continue"
                    )
                ).json()["messages"]
                assert [r["kind"] for r in only_continue] == ["continue", "continue"]
                assert [r["meta"]["candidate_index"] for r in only_continue] == [1, 0]
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_append_writes_only_ai_messages_and_get_is_read_only(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 4：append 只写 ai_messages（其它业务表零变、usage_ledger 恒 0）；GET 只读不建行。"""
    from ww_api.main import create_app

    app = create_app()
    transport = httpx.ASGITransport(app=app)
    project_uuid: uuid.UUID | None = None
    try:
        async with LifespanManager(app):
            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
                project_uuid = await _create_project(client, "AC-4 旁路留痕作品")

                # 关键旁路表快照（append 前）。
                side_models = [UsageLedger, Chapter, ChapterReview, WorldEntity, Job]
                before = {m.__name__: await _count(e2e_sm, m, project_uuid) for m in side_models}

                resp = await client.post(
                    f"/projects/{project_uuid}/ai-messages",
                    json={
                        "thread_id": str(uuid.uuid4()),
                        "kind": "generator",
                        "tool_key": "book-title",
                        "messages": [
                            {"role": "author", "content": "起个书名", "meta": {}},
                            {"role": "ai", "content": "《九霄剑主》", "meta": {}},
                        ],
                    },
                )
                assert resp.status_code == 201, resp.text

                # 只写 ai_messages：其它业务表逐一零变化。
                after = {m.__name__: await _count(e2e_sm, m, project_uuid) for m in side_models}
                assert after == before, f"append touched a non-ai_messages table: {before}->{after}"
                # 无网关/无 LLM：usage_ledger 恒 0（纯 CRUD 证据）。
                assert await _count(e2e_sm, UsageLedger, project_uuid) == 0
                assert await _count(e2e_sm, AiMessage, project_uuid) == 2

                # GET 只读：读后 ai_messages 计数不变（不建行）。
                assert (
                    await client.get(f"/projects/{project_uuid}/ai-messages")
                ).status_code == 200
                assert await _count(e2e_sm, AiMessage, project_uuid) == 2
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_unknown_project_404_and_validation_422(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 5：未知项目 → 404；各类非法请求（role/kind/空/超上限）→ 422，且零落库。"""
    from ww_api.main import create_app

    app = create_app()
    transport = httpx.ASGITransport(app=app)
    project_uuid: uuid.UUID | None = None
    try:
        async with LifespanManager(app):
            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
                # 未知项目：POST + GET 皆 404 NOT_FOUND 信封。
                ghost = uuid.uuid4()
                miss_post = await client.post(
                    f"/projects/{ghost}/ai-messages",
                    json={
                        "thread_id": str(uuid.uuid4()),
                        "kind": "refine",
                        "messages": [{"role": "author", "content": "x", "meta": {}}],
                    },
                )
                assert miss_post.status_code == 404
                assert miss_post.json()["error"]["code"] == "NOT_FOUND"
                miss_get = await client.get(f"/projects/{ghost}/ai-messages")
                assert miss_get.status_code == 404
                assert miss_get.json()["error"]["code"] == "NOT_FOUND"

                project_uuid = await _create_project(client, "AC-4 校验作品")
                thread = str(uuid.uuid4())
                base = {"thread_id": thread, "kind": "refine"}

                # 各 422 非法体：非法 role / 非法 kind / 空 messages / 超上限 content / meta。
                invalid_bodies = [
                    {**base, "messages": [{"role": "system", "content": "x"}]},  # 非法 role
                    {
                        **base,
                        "kind": "nonsense",  # 非法 kind
                        "messages": [{"role": "author", "content": "x"}],
                    },
                    {**base, "messages": []},  # 空 messages
                    {
                        **base,
                        "messages": [{"role": "author", "content": "x" * (_CONTENT_CAP + 1)}],
                    },  # content 超上限
                    {
                        **base,
                        "messages": [
                            {"role": "author", "content": "x", "meta": {"blob": "y" * _META_CAP}}
                        ],
                    },  # meta 序列化超上限
                ]
                for body in invalid_bodies:
                    bad = await client.post(f"/projects/{project_uuid}/ai-messages", json=body)
                    assert bad.status_code == 422, (
                        f"expected 422 for {body!r}, got {bad.status_code}"
                    )

        # 失败零落库：所有 422 都没有写入任何行。
        assert project_uuid is not None
        assert await _count(e2e_sm, AiMessage, project_uuid) == 0
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_clarify_abandon_persists_only_explicit_posts(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 6：clarify-abandon 代理——服务端无部分写路径，只持久化被显式 POST 的已定线程。"""
    from ww_api.main import create_app

    app = create_app()
    transport = httpx.ASGITransport(app=app)
    concluded_thread = uuid.uuid4()
    project_uuid: uuid.UUID | None = None
    try:
        async with LifespanManager(app):
            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
                project_uuid = await _create_project(client, "AC-4 澄清放弃作品")

                # 从未 POST 的（放弃的）线程 → 库里无痕：GET 空、计数 0。
                # 缓冲是客户端的；纯放弃 = 不 POST，服务端没有任何部分写入路径。
                empty = await client.get(f"/projects/{project_uuid}/ai-messages?chapter_no=8")
                assert empty.status_code == 200
                assert empty.json()["messages"] == []
                assert await _count(e2e_sm, AiMessage, project_uuid) == 0

                # 只有「已定」线程被显式 POST 才落库。
                await client.post(
                    f"/projects/{project_uuid}/ai-messages",
                    json={
                        "thread_id": str(concluded_thread),
                        "chapter_no": 8,
                        "kind": "refine",
                        "messages": [
                            {"role": "author", "content": "润色这段", "meta": {}},
                            {
                                "role": "ai",
                                "content": "润色结果",
                                "meta": {"version_no": 1, "segment": "原段"},
                            },
                        ],
                    },
                )

                # GET 恰返回被显式 POST 的 2 行——无任何「缓冲/放弃」幻影行。
                after = (
                    await client.get(f"/projects/{project_uuid}/ai-messages?chapter_no=8")
                ).json()["messages"]
                assert len(after) == 2
                assert {r["thread_id"] for r in after} == {str(concluded_thread)}
        assert project_uuid is not None
        assert await _count(e2e_sm, AiMessage, project_uuid) == 2
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)