"""M2 端到端：写 → 审(一致性) → 裁决 → 验收(事务) → 摘要入库（真实 DB，零 token）。

证明 M2 闭环（不变量 #3/#4/#5）：
  `POST /projects` → `POST .../draft`(SSE) → `PUT .../draft`(自动保存)
  → `POST .../review`(SSE，产 ≥1 冲突) → `GET .../reviews`(留痕)
  → `POST .../accept`(**残缺裁决 → 409 CONFLICT_UNRESOLVED，未写库**)
  → `POST .../accept`(**全量裁决 + 终稿 → 200**)
且 DB 为唯一真源：`chapters`(草稿+accepted 多版本) / `chapter_digests`(从**终稿**提炼) /
`chapter_reviews`(裁决留痕) / `usage_ledger`(写/审/digest 三档位记账各落库)。

确定性 & 零成本（同 M1）：真实 `Gateway` + 假适配器（吐固定 `Delta`/`parsed`，绝不联网）
+ 真实 `SqlAlchemyLedgerSink`（用量记账闭环也被走通）。无 pg → skip。

M2 多档位 + 结构化输出：假适配器据 `req.output_schema` 分支——
- writer/stream → 固定 token；
- analyst/run（`ContinuityReview`）→ 故意产 ≥1 `Conflict`（测裁决 + gate）；
- light/run（`ChapterDigestFacts`）→ facts 内容可追溯到**终稿**（验证 digest 从终稿非草稿，#4）。

坑（见 memory/gotchas）：
- `LifespanManager` 触发 lifespan → `seed_stub_user`（owner_id FK 依赖它）。
- `get_sessionmaker` engine 绑定首个 loop → 每 DB 测试清缓存重建、结束 dispose。
- 网关 ledger / 验收-side repos 只 flush；端点/验收事务 commit → usage_ledger/留痕方落库。
- ASGITransport `raise_app_exceptions=True`：gate 409 是 `AppError` → 正常返回信封（不上抛）。
"""

from __future__ import annotations

import json
import uuid
from collections.abc import AsyncIterator, Callable
from typing import Annotated

import httpx
import pytest
from asgi_lifespan import LifespanManager
from fastapi import Depends
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from ww_agents import Conflict, ContinuityReview
from ww_api.services.digest_extraction import ChapterDigestFacts
from ww_db import get_session, get_sessionmaker
from ww_db.models import (
    Chapter,
    ChapterDigest,
    ChapterReview,
    Project,
    UsageLedger,
)
from ww_llm_gateway import (
    Gateway,
    SqlAlchemyLedgerSink,
    resolve_route,
)
from ww_llm_gateway.adapters.base import (
    Capabilities,
    ProviderResult,
    ProviderUsage,
    StreamChunk,
)
from ww_llm_gateway.types import LlmRequest

# 三档位（writer/analyst/light）默认都路由到 deepseek（config.tier_defaults）。
_PROVIDER = "deepseek"

# 确定性流式写章 token。
_TOKENS = ["第", "一", "章", "：", "试炼之始。"]

# 续审故意产的固定冲突（≥1 → 触发裁决 + gate）。
_CONFLICT = Conflict(
    type="设定违例",
    where="第3段：主角施展禁术",
    refs=["世界观·禁术条目", "第1章"],
    suggestion="改为消耗灵力的普通法术，符合前文设定",
)

# 终稿里可被 digest 追溯的标记串（验证 digest 从终稿提炼，不变量 #4）。
_FINAL_MARKER = "终稿独有事件：主角在第十章觉醒血脉"

# 各档位假用量（喂记账；证明三档位用量闭环各自落库）。
_USAGE = {
    "writer": ProviderUsage(input_tokens=11, output_tokens=5),
    "analyst": ProviderUsage(input_tokens=23, output_tokens=7),
    "light": ProviderUsage(input_tokens=17, output_tokens=9),
}


class _FakeMultiTierAdapter:
    """实现 `ProviderAdapter` Protocol：按调用形态/`output_schema` 分支，绝不联网。

    - `stream()`：吐固定写章 token（writer/draft）。
    - `complete()`：据 `req.output_schema` 返回对应 `parsed`：
      `ContinuityReview`（续审，产固定冲突）/ `ChapterDigestFacts`（digest，追溯终稿）。
    每路均带 `ProviderUsage` → 网关据此各落一条 usage_ledger。
    """

    provider = _PROVIDER

    def __init__(self, *, conflicts: list[Conflict] | None = None) -> None:
        # 可配冲突数（主用例 ≥1；零冲突直通用例传 []）。
        self._conflicts = [_CONFLICT] if conflicts is None else conflicts

    def capabilities(self) -> Capabilities:
        return Capabilities(structured_output=True)

    async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
        if req.output_schema is ContinuityReview:
            review = ContinuityReview(conflicts=list(self._conflicts))
            return ProviderResult(
                text=review.model_dump_json(),
                parsed=review,
                usage=_USAGE["analyst"],
            )
        # 否则视为 digest 提炼（ChapterDigestFacts）：把终稿标记塞进 summary →
        # 断言 digest 来自终稿（非审稿草稿，不变量 #4）。
        facts = ChapterDigestFacts(
            summary=_FINAL_MARKER,
            events=["主角觉醒"],
            locations=["试炼之地"],
        )
        return ProviderResult(
            text=facts.model_dump_json(),
            parsed=facts,
            usage=_USAGE["light"],
        )

    async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
        for token in _TOKENS:
            yield StreamChunk(text=token)
        yield StreamChunk(usage=_USAGE["writer"])


@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()


def _parse_sse(raw: str) -> list[tuple[str, str]]:
    """把 text/event-stream 原文解析为 `(event, data)` 帧列表。"""
    frames: list[tuple[str, str]] = []
    event: str | None = None
    data: str | None = None
    for line in raw.splitlines():
        if line.startswith("event:"):
            event = line[len("event:") :].strip()
        elif line.startswith("data:"):
            data = line[len("data:") :].strip()
        elif line == "":
            if event is not None and data is not None:
                frames.append((event, data))
            event, data = None, None
    if event is not None and data is not None:
        frames.append((event, data))
    return frames


def _gateway_override_factory(
    adapter: _FakeMultiTierAdapter,
) -> Callable[[AsyncSession], Gateway]:
    """构造一个依赖覆盖：真实 Gateway + 给定假适配器 + 真实 ledger（请求 session）。

    ledger 用**请求 session**（`Depends(get_session)`，FastAPI 按请求缓存，与端点/验收
    事务同一实例）→ 由端点流末 / 验收事务的 `commit()` 落库（同 M1 回归校验）。
    """

    def _override(session: Annotated[AsyncSession, Depends(get_session)]) -> Gateway:
        return Gateway(
            adapters={_PROVIDER: adapter},
            ledger=SqlAlchemyLedgerSink(session),
            resolver=resolve_route,
        )

    return _override


async def _cleanup(e2e_sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
    """按 FK 顺序清理（无级联的先删，再删 project；chapters/digests/reviews 经 FK 级联）。"""
    async with e2e_sm() as cleanup:
        await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
        await cleanup.execute(delete(ChapterDigest).where(ChapterDigest.project_id == project_uuid))
        await cleanup.execute(delete(ChapterReview).where(ChapterReview.project_id == project_uuid))
        await cleanup.execute(delete(Chapter).where(Chapter.project_id == project_uuid))
        await cleanup.execute(delete(Project).where(Project.id == project_uuid))
        await cleanup.commit()


async def test_m2_closed_loop_write_review_adjudicate_accept_digest(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    from ww_api.main import create_app
    from ww_api.services.project_deps import (
        get_digest_gateway,
        get_review_gateway,
        get_writer_gateway,
    )

    adapter = _FakeMultiTierAdapter()  # 默认产 1 个冲突
    override = _gateway_override_factory(adapter)

    app = create_app()
    # 三档位网关均经 override 注入（真实 Gateway + 同一假适配器 + 真实 ledger）。
    app.dependency_overrides[get_writer_gateway] = override
    app.dependency_overrides[get_review_gateway] = override
    app.dependency_overrides[get_digest_gateway] = override

    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:
                # 1) 立项 → 201。
                title = "M2 E2E 闭环验证作品"
                create_resp = await client.post(
                    "/projects",
                    json={"title": title, "genre": "玄幻", "logline": "少年试炼"},
                )
                assert create_resp.status_code == 201
                project_id = create_resp.json()["id"]
                project_uuid = uuid.UUID(project_id)

                # 2) 流式写章草稿（SSE）→ 重组文本。
                draft_resp = await client.post(f"/projects/{project_id}/chapters/1/draft")
                assert draft_resp.status_code == 200
                frames = _parse_sse(draft_resp.text)
                token_frames = [d for (ev, d) in frames if ev == "token"]
                assert [d for (ev, d) in frames if ev == "error"] == []
                assert len([d for (ev, d) in frames if ev == "done"]) == 1
                draft_text = "".join(json.loads(d)["text"] for d in token_frames)
                assert draft_text == "".join(_TOKENS)

                # 自动保存草稿（version=1, status='draft'）。
                save_resp = await client.put(
                    f"/projects/{project_id}/chapters/1/draft",
                    json={"text": draft_text},
                )
                assert save_resp.status_code == 200
                assert save_resp.json()["version"] == 1
                assert save_resp.json()["status"] == "draft"

                # 3) 续审（SSE，body draft=当前草稿）→ ≥1 section(continuity,done) +
                #    ≥1 conflict + 1 done、无 error。
                review_resp = await client.post(
                    f"/projects/{project_id}/chapters/1/review",
                    json={"draft": draft_text},
                )
                assert review_resp.status_code == 200
                assert review_resp.headers["content-type"].startswith("text/event-stream")
                rframes = _parse_sse(review_resp.text)
                section_frames = [json.loads(d) for (ev, d) in rframes if ev == "section"]
                conflict_frames = [json.loads(d) for (ev, d) in rframes if ev == "conflict"]
                rdone = [d for (ev, d) in rframes if ev == "done"]
                rerror = [d for (ev, d) in rframes if ev == "error"]
                assert rerror == []
                assert len(rdone) == 1
                continuity_sections = [
                    s
                    for s in section_frames
                    if s.get("name") == "continuity" and s.get("status") == "done"
                ]
                assert len(continuity_sections) >= 1
                assert len(conflict_frames) >= 1
                assert conflict_frames[0]["type"] == _CONFLICT.type
                assert conflict_frames[0]["suggestion"] == _CONFLICT.suggestion

                # 4) GET 历史含刚落的留痕（conflicts 非空、decisions 为空）。
                hist_resp = await client.get(f"/projects/{project_id}/chapters/1/reviews")
                assert hist_resp.status_code == 200
                reviews = hist_resp.json()["reviews"]
                assert len(reviews) >= 1
                latest = reviews[0]
                assert len(latest["conflicts"]) >= 1
                assert latest["decisions"] is None

                # 5) 冲突 gate（未决禁验收）：残缺裁决（空 decisions，缺 conflict_index=0）
                #    → 409 CONFLICT_UNRESOLVED，details 命中缺项。
                final_text = draft_text + "\n\n" + _FINAL_MARKER
                gate_resp = await client.post(
                    f"/projects/{project_id}/chapters/1/accept",
                    json={"final_text": final_text, "decisions": []},
                )
                assert gate_resp.status_code == 409
                gate_body = gate_resp.json()
                assert gate_body["error"]["code"] == "CONFLICT_UNRESOLVED"
                details = gate_body["error"]["details"]
                assert 0 in details["missing_conflict_indices"]
                assert details["conflict_count"] >= 1

                # 6) 完整验收：全量裁决 + 终稿（与草稿不同，验证 #4）→ 200。
                accept_resp = await client.post(
                    f"/projects/{project_id}/chapters/1/accept",
                    json={
                        "final_text": final_text,
                        "decisions": [
                            {"conflict_index": 0, "verdict": "accept", "note": "采纳改法"}
                        ],
                    },
                )
                assert accept_resp.status_code == 200
                acc = accept_resp.json()
                assert acc["accepted_version"] == 2  # 草稿 v1 + accepted v2
                assert acc["digest_added"] is True
                assert acc["decisions_recorded"] == 1
                assert acc["review_id"] is not None

        # 7) DB 真源断言（经 e2e session）。
        assert project_uuid is not None
        async with e2e_sm() as verify:
            # chapters：草稿 v1 仍在 + 新增 accepted v2（content==final_text）。
            chapter_rows = (
                (
                    await verify.execute(
                        select(Chapter)
                        .where(Chapter.project_id == project_uuid)
                        .order_by(Chapter.version)
                    )
                )
                .scalars()
                .all()
            )
            assert len(chapter_rows) == 2
            draft_row, accepted_row = chapter_rows[0], chapter_rows[1]
            assert draft_row.version == 1
            assert draft_row.status == "draft"
            assert draft_row.content == draft_text
            assert accepted_row.version == 2
            assert accepted_row.status == "accepted"
            assert accepted_row.content == final_text

            # chapter_digests：append 一行，facts 可追溯到**终稿**（#4：summary 含终稿标记）。
            digest_rows = (
                (
                    await verify.execute(
                        select(ChapterDigest).where(ChapterDigest.project_id == project_uuid)
                    )
                )
                .scalars()
                .all()
            )
            assert len(digest_rows) == 1
            digest_facts = digest_rows[0].facts
            assert digest_facts["summary"] == _FINAL_MARKER
            # 终稿标记不在草稿里 → 证明 digest 从终稿而非审稿草稿提炼。
            assert _FINAL_MARKER not in draft_text

            # chapter_reviews：留痕行 decisions 已写（裁决留痕）。
            review_rows = (
                (
                    await verify.execute(
                        select(ChapterReview).where(ChapterReview.project_id == project_uuid)
                    )
                )
                .scalars()
                .all()
            )
            assert len(review_rows) == 1
            assert len(review_rows[0].conflicts) >= 1
            decisions = review_rows[0].decisions
            assert decisions is not None
            assert decisions["items"][0]["conflict_index"] == 0
            assert decisions["items"][0]["verdict"] == "accept"

            # usage_ledger：写章 + 续审 + digest 提炼三档位各记一条（≥3）。
            ledger_rows = (
                (
                    await verify.execute(
                        select(UsageLedger).where(UsageLedger.project_id == project_uuid)
                    )
                )
                .scalars()
                .all()
            )
            assert len(ledger_rows) >= 3
            # 三档位 input_tokens 各异 → 验证各档位调用都真落库（事务已 commit）。
            input_token_set = {r.input_tokens for r in ledger_rows}
            assert _USAGE["writer"].input_tokens in input_token_set
            assert _USAGE["analyst"].input_tokens in input_token_set
            assert _USAGE["light"].input_tokens in input_token_set
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_m2_zero_conflict_accept_passes_gate_directly(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """零冲突直通：续审产 conflicts=[] → 空裁决 accept 直接 200（gate 直通）。"""
    from ww_api.main import create_app
    from ww_api.services.project_deps import (
        get_digest_gateway,
        get_review_gateway,
        get_writer_gateway,
    )

    adapter = _FakeMultiTierAdapter(conflicts=[])  # 零冲突路径
    override = _gateway_override_factory(adapter)

    app = create_app()
    app.dependency_overrides[get_writer_gateway] = override
    app.dependency_overrides[get_review_gateway] = override
    app.dependency_overrides[get_digest_gateway] = override

    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:
                create_resp = await client.post("/projects", json={"title": "M2 零冲突直通作品"})
                assert create_resp.status_code == 201
                project_id = create_resp.json()["id"]
                project_uuid = uuid.UUID(project_id)

                draft_resp = await client.post(f"/projects/{project_id}/chapters/1/draft")
                draft_text = "".join(
                    json.loads(d)["text"]
                    for (ev, d) in _parse_sse(draft_resp.text)
                    if ev == "token"
                )
                await client.put(
                    f"/projects/{project_id}/chapters/1/draft",
                    json={"text": draft_text},
                )

                # 续审产零冲突 → 留痕 conflicts=[]。
                review_resp = await client.post(
                    f"/projects/{project_id}/chapters/1/review",
                    json={"draft": draft_text},
                )
                assert review_resp.status_code == 200
                rframes = _parse_sse(review_resp.text)
                assert [d for (ev, d) in rframes if ev == "conflict"] == []

                # 空裁决 accept → gate 直通 200。
                final_text = draft_text + "\n" + _FINAL_MARKER
                accept_resp = await client.post(
                    f"/projects/{project_id}/chapters/1/accept",
                    json={"final_text": final_text, "decisions": []},
                )
                assert accept_resp.status_code == 200
                acc = accept_resp.json()
                assert acc["accepted_version"] == 2
                assert acc["digest_added"] is True
                assert acc["decisions_recorded"] == 0

        # DB 断言：accepted 行落库、digest 从终稿。
        assert project_uuid is not None
        async with e2e_sm() as verify:
            accepted = (
                await verify.execute(
                    select(Chapter).where(
                        Chapter.project_id == project_uuid,
                        Chapter.status == "accepted",
                    )
                )
            ).scalar_one()
            assert accepted.content == final_text
            digest = (
                await verify.execute(
                    select(ChapterDigest).where(ChapterDigest.project_id == project_uuid)
                )
            ).scalar_one()
            assert digest.facts["summary"] == _FINAL_MARKER
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)