"""T6.4 端到端：创作工具箱通用生成器端点（真实 DB，零 token，MOCK 网关）。

证明 T6 通用生成器框架闭环（声明驱动、一条执行路径 + 不变量 #3 预览不写库 / 入库必过
continuity gate + 白名单）。镜像 `tests/test_m5_e2e.py` 的范式：真实 `Gateway` + 据
`req.output_schema` 分支返回 `parsed` 的假适配器（绝不联网）+ 真实 `SqlAlchemyLedgerSink`
（记账真落 pg）+ DB 真源断言。无 pg → skip。

可注入缝（Wave A）：`get_tier_gateway_builder` 返回 `TierGatewayBuilder`
（`Callable[[Tier], Awaitable[Gateway]]`）。toolbox 端点 generate 调 `build_gateway(spec.tier)`、
ingest 预检调 `build_gateway("analyst")`——故 override **这一个 builder**，让它返回单个真
`Gateway`（假适配器据 output_schema 分支：生成产物 vs ContinuityReview）即可覆盖两路。

用例：
1. `GET /skills/toolbox` 返回全 11 key（legacy 3 带 legacy_route+is_legacy=true；新 8）；
   brainstorm/glossary/golden-finger 的 ingestable 标志符合预期。
2. brainstorm `generate` → 200 ToolGeneratePreviewResponse（output_kind + preview.ideas）；
   断言 **ledger 真落行** 且 **业务表零新增**（world_entities 计数不变）——不变量 #3。
3. glossary `ingest` → world_entities：先**不带 acknowledge** + 预检有冲突 → **409
   CONFLICT_UNRESOLVED**，断言 world_entities **此刻无新行**；再带 acknowledge_conflicts=true
   → 201，world_entities 行**真落 pg**（rules list→JSONB dict 形变保留）；rejected_tables 空。
4. 未知 tool_key `generate` → 404 NOT_FOUND envelope。
5. 无凭据（该 tier 无配置）→ 503 LLM_UNAVAILABLE envelope（不 override builder，走真依赖）。
负向：响应体绝不泄露 token/secret；预览路径只 commit ledger。
"""

from __future__ import annotations

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

import httpx
import pytest
from asgi_lifespan import LifespanManager
from fastapi import Depends
from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from ww_agents import (
    Conflict,
    ContinuityReview,
    GlossaryResult,
    GlossaryTerm,
    GoldenFingerResult,
    IdeaListResult,
)
from ww_agents.schemas import GoldenFinger, Idea
from ww_db import get_session, get_sessionmaker
from ww_db.models import Project, UsageLedger, WorldEntity
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, Tier

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

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

# brainstorm 假产物：3 条脑洞（纯预览，不入库）。
_IDEAS = IdeaListResult(
    ideas=[
        Idea(premise="天降系统但只在深夜生效", hook="主角被迫昼伏夜出", genre_fit="都市异能"),
        Idea(premise="转生成反派的家养灵宠", hook="一边卖萌一边算计", genre_fit="搞笑修真"),
        Idea(premise="末世里唯一能种田的人", hook="种子即权力", genre_fit="末世种田"),
    ]
)

# glossary 假产物（生成预览用；ingest 由前端映射成 world_entities，故端点不强依赖此形）。
_GLOSSARY = GlossaryResult(
    terms=[
        GlossaryTerm(
            name="玄晶", type="货币", definition="修真界硬通货", rules=["不可伪造", "遇火则碎"]
        ),
    ]
)

# golden-finger 假产物（生成预览用）。
_GOLDEN = GoldenFingerResult(
    systems=[
        GoldenFinger(
            name="九转炼体诀",
            mechanism="吞噬天地灵气淬体",
            growth="每转突破一境",
            limits="每转需一枚玄晶，不可逆修",
        )
    ]
)

# 待入库的 world_entities（贴 WorldEntityCardView：type/name/rules）。
_INGEST_ENTITIES = [
    {"type": "货币", "name": "玄晶", "rules": ["不可伪造", "遇火则碎"]},
    {"type": "力量体系", "name": "九转炼体诀", "rules": ["每转需一枚玄晶", "不可逆修"]},
]

# precheck 假冲突（让入库 gate 触发 409）。
_PRECHECK_CONFLICT = ContinuityReview(
    conflicts=[
        Conflict(
            type="设定违例",
            where="玄晶",
            refs=["既有货币体系"],
            suggestion="玄晶『遇火则碎』与既有设定冲突，请裁决",
        )
    ]
)


class _FakeToolboxAdapter:
    """实现 `ProviderAdapter` Protocol：按 `req.output_schema` 分支返回 parsed，绝不联网。

    - `IdeaListResult` → 3 条脑洞（light）。
    - `GlossaryResult` → 术语表（analyst）。
    - `GoldenFingerResult` → 金手指（writer）。
    - `ContinuityReview` → 带冲突的预检结果（analyst）——触发 ingest 409 gate。
    每路带 `ProviderUsage` → 网关据此落 usage_ledger。
    """

    def __init__(self, provider: str = _PROVIDER) -> None:
        self.provider = provider

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

    async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
        schema = req.output_schema
        if schema is IdeaListResult:
            return ProviderResult(
                text=_IDEAS.model_dump_json(), parsed=_IDEAS, usage=_USAGE["light"]
            )
        if schema is GlossaryResult:
            return ProviderResult(
                text=_GLOSSARY.model_dump_json(), parsed=_GLOSSARY, usage=_USAGE["analyst"]
            )
        if schema is GoldenFingerResult:
            return ProviderResult(
                text=_GOLDEN.model_dump_json(), parsed=_GOLDEN, usage=_USAGE["writer"]
            )
        if schema is ContinuityReview:
            return ProviderResult(
                text=_PRECHECK_CONFLICT.model_dump_json(),
                parsed=_PRECHECK_CONFLICT,
                usage=_USAGE["analyst"],
            )
        raise AssertionError(f"unexpected output_schema in T6 fake adapter: {schema!r}")

    async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
        # 工具箱生成走 run()（结构化），不走 stream——但 Protocol 要求实现。
        yield StreamChunk(usage=_USAGE["light"])
        raise AssertionError("T6 toolbox generation must not stream")


@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 _tier_builder_override(
    adapter: _FakeToolboxAdapter,
) -> Callable[[AsyncSession], Callable[[Tier], Awaitable[Gateway]]]:
    """依赖覆盖：`get_tier_gateway_builder` → 返回「按 tier 建网关」的 builder。

    builder 忽略 tier 返回同一个真 `Gateway`（单 provider 假适配器据 output_schema 分支 +
    真实 ledger，请求 session）——同时覆盖 generate（按 spec.tier）与 ingest 预检（analyst）。
    """

    def _override(
        session: Annotated[AsyncSession, Depends(get_session)],
    ) -> Callable[[Tier], Awaitable[Gateway]]:
        async def _build(_tier: Tier) -> Gateway:
            return Gateway(
                adapters={adapter.provider: adapter},
                ledger=SqlAlchemyLedgerSink(session),
                resolver=resolve_route,
            )

        return _build

    return _override


async def _ledger_count(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> int:
    async with sm() as s:
        return int(
            (
                await s.execute(
                    select(func.count())
                    .select_from(UsageLedger)
                    .where(UsageLedger.project_id == project_uuid)
                )
            ).scalar_one()
        )


async def _world_count(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> int:
    async with sm() as s:
        return int(
            (
                await s.execute(
                    select(func.count())
                    .select_from(WorldEntity)
                    .where(WorldEntity.project_id == project_uuid)
                )
            ).scalar_one()
        )


async def _cleanup(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
    """按 FK 顺序清理。"""
    async with sm() as cleanup:
        await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
        await cleanup.execute(delete(WorldEntity).where(WorldEntity.project_id == project_uuid))
        await cleanup.execute(delete(Project).where(Project.id == project_uuid))
        await cleanup.commit()


def _assert_no_secret_leak(payload: object) -> None:
    """负向：响应体序列化后绝不含 token/secret/api_key 等敏感字样。"""
    import json

    blob = json.dumps(payload, ensure_ascii=False).lower()
    for needle in ("api_key", "secret", "token", "bearer", "sk-", "credential"):
        assert needle not in blob, f"sensitive marker leaked into response: {needle!r}"


async def test_t6_list_toolbox_returns_all_eleven_tools(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 1：GET /skills/toolbox 返回全 11 key；legacy 3 带 legacy_route，新 8 走通用执行器。"""
    from ww_api.main import create_app

    app = create_app()
    transport = httpx.ASGITransport(app=app)
    async with LifespanManager(app):
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            resp = await client.get("/skills/toolbox")
            assert resp.status_code == 200
            body = resp.json()

    tools = {t["key"]: t for t in body["tools"]}
    # 全 11 key（legacy 3 + 新 8）。
    assert set(tools) == {
        "worldbuilding",
        "character",
        "outline",
        "brainstorm",
        "book-title",
        "blurb",
        "name",
        "golden-finger",
        "glossary",
        "opening",
        "fine-outline",
    }
    # legacy（spec=None）：is_legacy=true + 带 legacy_route，不可经通用执行入库。
    legacy_keys = {"worldbuilding", "character", "outline"}
    for key in legacy_keys:
        assert tools[key]["is_legacy"] is True
        assert tools[key]["legacy_route"]
    # 新工具：is_legacy=false。
    for key in set(tools) - legacy_keys:
        assert tools[key]["is_legacy"] is False
    # ingestable 标志：glossary/golden-finger/fine-outline 可入库；brainstorm 纯预览。
    assert tools["glossary"]["ingestable"] is True
    assert tools["golden-finger"]["ingestable"] is True
    assert tools["fine-outline"]["ingestable"] is True
    assert tools["brainstorm"]["ingestable"] is False
    # brainstorm 描述符含一句话需求字段。
    assert any(f["name"] == "brief" for f in tools["brainstorm"]["input_fields"])
    _assert_no_secret_leak(body)


async def test_t6_brainstorm_generate_previews_without_writing_business_tables(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 2：brainstorm generate → 200 预览；ledger 真落行 + 业务表零新增（不变量 #3）。"""
    from ww_api.main import create_app
    from ww_api.services.project_deps import get_tier_gateway_builder

    app = create_app()
    app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
        _FakeToolboxAdapter()
    )

    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": "T6 脑洞验证作品", "genre": "玄幻"}
                )
                assert create_resp.status_code == 201
                project_id = create_resp.json()["id"]
                project_uuid = uuid.UUID(project_id)

                world_before = await _world_count(e2e_sm, project_uuid)

                gen_resp = await client.post(
                    f"/projects/{project_id}/skills/brainstorm/generate",
                    json={"brief": "想要三个突破常规的脑洞"},
                )
                assert gen_resp.status_code == 200
                body = gen_resp.json()
                assert body["tool_key"] == "brainstorm"
                assert body["output_kind"] == "IdeaListResult"
                ideas = body["preview"]["ideas"]
                assert [i["premise"] for i in ideas] == [
                    "天降系统但只在深夜生效",
                    "转生成反派的家养灵宠",
                    "末世里唯一能种田的人",
                ]
                _assert_no_secret_leak(body)

        # DB 真源：ledger 落 1 行（light）；业务表（world_entities）零新增（预览不写库）。
        assert project_uuid is not None
        assert await _ledger_count(e2e_sm, project_uuid) == 1
        assert await _world_count(e2e_sm, project_uuid) == world_before
        async with e2e_sm() as verify:
            row = (
                (
                    await verify.execute(
                        select(UsageLedger).where(UsageLedger.project_id == project_uuid)
                    )
                )
                .scalars()
                .one()
            )
            assert row.provider == _PROVIDER
            assert row.input_tokens == _USAGE["light"].input_tokens
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_t6_glossary_ingest_gate_then_acknowledge_persists(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 3：glossary ingest → 409 gate（无 ack + 有冲突）→ ack 放行 → world_entities 真落 pg。"""
    from ww_api.main import create_app
    from ww_api.services.project_deps import get_tier_gateway_builder

    app = create_app()
    app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
        _FakeToolboxAdapter()
    )

    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": "T6 词条入库作品", "genre": "玄幻"}
                )
                assert create_resp.status_code == 201
                project_id = create_resp.json()["id"]
                project_uuid = uuid.UUID(project_id)

                # 3a) 入库（未 acknowledge）→ precheck 有冲突 → 409 CONFLICT_UNRESOLVED。
                blocked = await client.post(
                    f"/projects/{project_id}/skills/glossary/ingest",
                    json={"world_entities": _INGEST_ENTITIES},
                )
                assert blocked.status_code == 409
                err = blocked.json()["error"]
                assert err["code"] == "CONFLICT_UNRESOLVED"
                details = err["details"]
                assert details["conflict_count"] == 1
                assert details["conflicts"][0]["type"] == "设定违例"
                assert details["conflicts"][0]["where"] == "玄晶"
                _assert_no_secret_leak(blocked.json())

                # 入库被拦 → 此刻 world_entities 表应无行（不静默入库，守不变量 #3）。
                assert await _world_count(e2e_sm, project_uuid) == 0

                # 3b) 作者裁决后带 acknowledge_conflicts=true 重发 → 201 入库。
                ok = await client.post(
                    f"/projects/{project_id}/skills/glossary/ingest",
                    json={"world_entities": _INGEST_ENTITIES, "acknowledge_conflicts": True},
                )
                assert ok.status_code == 201
                ok_body = ok.json()
                assert ok_body["table"] == "world_entities"
                assert ok_body["created"] == ["玄晶", "九转炼体诀"]
                # 合法路径（glossary writes=["world_entities"]）无越权 → rejected_tables 空。
                assert ok_body["rejected_tables"] == []
                _assert_no_secret_leak(ok_body)

        # DB 真源断言：world_entities 行真落 pg + rules list→JSONB dict 形变保留。
        assert project_uuid is not None
        async with e2e_sm() as verify:
            rows = (
                (
                    await verify.execute(
                        select(WorldEntity)
                        .where(WorldEntity.project_id == project_uuid)
                        .order_by(WorldEntity.name)
                    )
                )
                .scalars()
                .all()
            )
            assert {r.name for r in rows} == {"玄晶", "九转炼体诀"}
            xuanjing = next(r for r in rows if r.name == "玄晶")
            assert xuanjing.type == "货币"
            # 形变：rules list → JSONB dict {"rules":[...]}（rules 列表保留）。
            assert xuanjing.rules == {"rules": ["不可伪造", "遇火则碎"]}

            # 记账真落 pg：precheck(409 路径) + precheck(入库路径) = 2 条（入库本体不调网关）。
            ledger_rows = (
                (
                    await verify.execute(
                        select(UsageLedger).where(UsageLedger.project_id == project_uuid)
                    )
                )
                .scalars()
                .all()
            )
            assert len(ledger_rows) == 2
            assert all(r.provider == _PROVIDER for r in ledger_rows)
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_t6_unknown_tool_key_generate_returns_404(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 4：未知 tool_key generate → 404 NOT_FOUND envelope。"""
    from ww_api.main import create_app
    from ww_api.services.project_deps import get_tier_gateway_builder

    app = create_app()
    # builder 即使被覆盖，也应在工具解析阶段 404（不触达网关）。
    app.dependency_overrides[get_tier_gateway_builder] = _tier_builder_override(
        _FakeToolboxAdapter()
    )

    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": "T6 未知工具作品"})
                assert create_resp.status_code == 201
                project_id = create_resp.json()["id"]
                project_uuid = uuid.UUID(project_id)

                resp = await client.post(
                    f"/projects/{project_id}/skills/does-not-exist/generate",
                    json={"brief": "x"},
                )
                assert resp.status_code == 404
                err = resp.json()["error"]
                assert err["code"] == "NOT_FOUND"
                assert err["details"]["tool_key"] == "does-not-exist"
                _assert_no_secret_leak(resp.json())
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_t6_no_provider_returns_503(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 5：无凭据（不 override builder，走真 build_gateway_for_tier）→ 503 LLM_UNAVAILABLE。"""
    from ww_api.main import create_app

    app = create_app()
    # 不 override get_tier_gateway_builder → 真依赖在无凭据时抛 LLM_UNAVAILABLE。

    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": "T6 无凭据作品"})
                assert create_resp.status_code == 201
                project_id = create_resp.json()["id"]
                project_uuid = uuid.UUID(project_id)

                resp = await client.post(
                    f"/projects/{project_id}/skills/brainstorm/generate",
                    json={"brief": "脑洞"},
                )
                assert resp.status_code == 503
                err = resp.json()["error"]
                assert err["code"] == "LLM_UNAVAILABLE"
                _assert_no_secret_leak(resp.json())

                # 503 后业务表与 ledger 均无新增（凭据探测在生成前拦下）。
                assert await _world_count(e2e_sm, project_uuid) == 0
                assert await _ledger_count(e2e_sm, project_uuid) == 0
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)