"""T6.2/T6.3 创作工具箱通用端点测试（内存替身，无 DB/无网络）。

覆盖：
- GET /skills/toolbox：列出全部工具（legacy + 新）+ ingestable/legacy_route 形。
- POST .../generate：按 key 解析 spec → 结构化预览；未知 key 404；legacy key 404；无凭据 503。
- POST .../ingest：world_entities continuity 409 gate + acknowledge 放行 + partition_writes 丢越权；
  不可入库工具 422；outline 细纲入库 upsert。
"""

from __future__ import annotations

import os
import uuid
from typing import Any

import httpx
import pytest
from cryptography.fernet import Fernet
from fakes_projects import FakeProjectRepo, FakeSession
from test_projects import _empty_memory_repos
from ww_agents import Conflict, ContinuityReview
from ww_agents.schemas import (
    BookTeardownResult,
    ContinuationResult,
    DeAiResult,
    Idea,
    IdeaListResult,
    PolishResult,
)
from ww_core.domain.chapter_repo import ChapterDraftView, ChapterView
from ww_core.domain.outline_write_repo import OutlineWriteView
from ww_core.domain.project_repo import ProjectCreate
from ww_core.domain.repositories import OutlineView
from ww_core.domain.world_entity_repo import WorldEntityWriteView
from ww_llm_gateway.types import LlmRequest, LlmResponse, ServedBy, Usage
from ww_shared import AppError, ErrorCode

STUB_OWNER = uuid.UUID(int=1)


class _SchemaRoutingGateway:
    """按 `req.output_schema` 返对应 parsed（generate / precheck 各拿自己的产物）。"""

    def __init__(self, by_schema: dict[type[Any] | None, Any]) -> None:
        self._by_schema = by_schema
        self.calls: list[type[Any] | None] = []

    async def run(self, req: LlmRequest) -> LlmResponse:
        schema = req.output_schema
        self.calls.append(schema)
        parsed = self._by_schema.get(schema)
        return LlmResponse(
            text=parsed.model_dump_json() if parsed is not None else "{}",
            parsed=parsed,
            usage=Usage(
                provider="fake",
                model="fake",
                input_tokens=1,
                output_tokens=1,
                cost_minor=0,
                currency="USD",
            ),
            served_by=ServedBy(provider="fake", model="fake"),
        )


class _FakeWorldWriteRepo:
    def __init__(self) -> None:
        self.rows: list[dict[str, Any]] = []

    async def create(
        self, project_id: uuid.UUID, *, type: str, name: str, rules: list[str]
    ) -> WorldEntityWriteView:
        self.rows.append({"type": type, "name": name, "rules": list(rules)})
        return WorldEntityWriteView(id=uuid.uuid4(), type=type, name=name)


class _FakeOutlineWriteRepo:
    def __init__(self) -> None:
        self.rows: list[dict[str, Any]] = []

    async def upsert_chapter(
        self,
        project_id: uuid.UUID,
        *,
        volume: int,
        chapter_no: int,
        beats: list[str],
        foreshadow_windows: list[dict[str, Any]],
    ) -> OutlineWriteView:
        self.rows.append({"chapter_no": chapter_no, "beats": list(beats)})
        return OutlineWriteView(
            chapter_no=chapter_no, volume=volume, beats=list(beats), foreshadow_windows=[]
        )


class _FakeOutlineReadRepo:
    """续写/按章工具读节拍用的内存替身：缺章 → None（端点降级到空节拍）。"""

    def __init__(self, beats_by_chapter: dict[int, list[str]] | None = None) -> None:
        self._beats = beats_by_chapter or {}

    async def get(self, project_id: uuid.UUID, chapter_no: int) -> Any:
        beats = self._beats.get(chapter_no)
        if beats is None:
            return None
        return OutlineView(
            chapter_no=chapter_no, volume=1, beats={"beats": beats}, foreshadow_windows=[]
        )

    async def list_for_project(self, project_id: uuid.UUID) -> list[Any]:
        return []


class _FakeChapterRepo:
    """续写读前文用的内存章节替身：按 chapter_no 存 accepted/draft 正文。"""

    def __init__(
        self,
        *,
        accepted: dict[int, str] | None = None,
        drafts: dict[int, str] | None = None,
    ) -> None:
        self._accepted = accepted or {}
        self._drafts = drafts or {}

    async def latest_accepted(self, project_id: uuid.UUID, chapter_no: int) -> ChapterView | None:
        content = self._accepted.get(chapter_no)
        if content is None:
            return None
        return ChapterView(
            project_id=project_id,
            chapter_no=chapter_no,
            volume=1,
            content=content,
            status="accepted",
            version=1,
        )

    async def get_draft(self, project_id: uuid.UUID, chapter_no: int) -> ChapterDraftView | None:
        content = self._drafts.get(chapter_no)
        if content is None:
            return None
        return ChapterDraftView(
            project_id=project_id,
            chapter_no=chapter_no,
            volume=1,
            content=content,
            status="draft",
            version=1,
        )


class _CaptureGateway:
    """记录最近一次 req.input（验证续写读前文/原文路由进了 context），返指定 parsed。"""

    def __init__(self, parsed: Any) -> None:
        self._parsed = parsed
        self.last_input: str | None = None

    async def run(self, req: LlmRequest) -> LlmResponse:
        # 工具箱端点恒传序列化文本 context（str）；存为 str 供断言。
        self.last_input = req.input if isinstance(req.input, str) else str(req.input)
        return LlmResponse(
            text=self._parsed.model_dump_json(),
            parsed=self._parsed,
            usage=Usage(
                provider="fake",
                model="fake",
                input_tokens=1,
                output_tokens=1,
                cost_minor=0,
                currency="USD",
            ),
            served_by=ServedBy(provider="fake", model="fake"),
        )


def _ideas() -> IdeaListResult:
    return IdeaListResult(ideas=[Idea(premise="废柴觉醒系统", hook="开局即巅峰", genre_fit="玄幻")])


def _no_conflicts() -> ContinuityReview:
    return ContinuityReview(conflicts=[])


def _with_conflicts() -> ContinuityReview:
    return ContinuityReview(
        conflicts=[Conflict(type="设定违例", where="吞噬系统", refs=["灵脉"], suggestion="改设定")]
    )


def _make_app(
    *,
    project_repo: FakeProjectRepo,
    gateway: Any,
    session: FakeSession | None = None,
    world_repo: _FakeWorldWriteRepo | None = None,
    outline_write_repo: _FakeOutlineWriteRepo | None = None,
    chapter_repo: _FakeChapterRepo | None = None,
    outline_read_repo: _FakeOutlineReadRepo | None = None,
    memory: Any = None,
    no_creds: bool = False,
) -> tuple[Any, FakeSession, _FakeWorldWriteRepo, _FakeOutlineWriteRepo]:
    os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
    from ww_api.main import create_app
    from ww_api.services.project_deps import (
        get_chapter_repo,
        get_memory_repos,
        get_outline_read_repo,
        get_outline_write_repo,
        get_project_repo,
        get_tier_gateway_builder,
        get_world_entity_write_repo,
    )
    from ww_db import get_session

    session = session or FakeSession()
    world_repo = world_repo or _FakeWorldWriteRepo()
    outline_write_repo = outline_write_repo or _FakeOutlineWriteRepo()
    chapter_repo = chapter_repo or _FakeChapterRepo()
    outline_read_repo = outline_read_repo or _FakeOutlineReadRepo()

    async def _build_ok(_tier: str) -> Any:
        return gateway

    async def _build_no_creds(_tier: str) -> Any:
        raise AppError(ErrorCode.LLM_UNAVAILABLE, "未配置凭据", {"provider": "deepseek"})

    app = create_app()
    app.dependency_overrides[get_project_repo] = lambda: project_repo
    app.dependency_overrides[get_memory_repos] = (
        (lambda: memory) if memory is not None else _empty_memory_repos
    )
    app.dependency_overrides[get_world_entity_write_repo] = lambda: world_repo
    app.dependency_overrides[get_outline_write_repo] = lambda: outline_write_repo
    app.dependency_overrides[get_chapter_repo] = lambda: chapter_repo
    app.dependency_overrides[get_outline_read_repo] = lambda: outline_read_repo
    app.dependency_overrides[get_session] = lambda: session
    app.dependency_overrides[get_tier_gateway_builder] = lambda: (
        _build_no_creds if no_creds else _build_ok
    )
    return app, session, world_repo, outline_write_repo


async def _seed_project(repo: FakeProjectRepo) -> uuid.UUID:
    view = await repo.create(STUB_OWNER, ProjectCreate(title="测试作品", genre="玄幻"))
    return uuid.UUID(str(view.id))


def _client(app: Any) -> httpx.AsyncClient:
    transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
    return httpx.AsyncClient(transport=transport, base_url="http://test")


# ---- GET /skills/toolbox ----


@pytest.mark.asyncio
async def test_list_toolbox_contains_all_keys() -> None:
    repo = FakeProjectRepo()
    app, *_ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}))

    async with _client(app) as client:
        resp = await client.get("/skills/toolbox")

    assert resp.status_code == 200
    tools = resp.json()["tools"]
    keys = {t["key"] for t in tools}
    assert keys >= {
        "worldbuilding",
        "character",
        "outline",
        "brainstorm",
        "book-title",
        "blurb",
        "name",
        "golden-finger",
        "glossary",
        "opening",
        "fine-outline",
        "continue",
        "expand",
        "de-ai",
        "teardown",
    }
    by_key = {t["key"]: t for t in tools}
    # 4 个竞品生成器：非 legacy、纯预览（不可入库）。
    for key in ("continue", "expand", "de-ai", "teardown"):
        assert by_key[key]["is_legacy"] is False
        assert by_key[key]["ingestable"] is False
    # legacy 携 legacy_route + is_legacy；新工具 ingestable 标记。
    assert by_key["worldbuilding"]["is_legacy"] is True
    assert by_key["worldbuilding"]["legacy_route"]
    assert by_key["brainstorm"]["is_legacy"] is False
    assert by_key["golden-finger"]["ingestable"] is True
    assert by_key["brainstorm"]["ingestable"] is False
    # input_fields 形（brief textarea）。
    assert any(f["name"] == "brief" for f in by_key["brainstorm"]["input_fields"])


# ---- POST .../generate ----


@pytest.mark.asyncio
async def test_generate_resolves_spec_and_returns_preview() -> None:
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    gateway = _SchemaRoutingGateway({IdeaListResult: _ideas()})
    app, session, *_ = _make_app(project_repo=repo, gateway=gateway)

    async with _client(app) as client:
        resp = await client.post(
            f"/projects/{pid}/skills/brainstorm/generate", json={"brief": "废柴流"}
        )

    assert resp.status_code == 200
    body = resp.json()
    assert body["tool_key"] == "brainstorm"
    assert body["output_kind"] == "IdeaListResult"
    assert body["preview"]["ideas"][0]["premise"] == "废柴觉醒系统"
    assert session.commits == 1  # 预览不写业务表，但落 ledger


@pytest.mark.asyncio
async def test_generate_unknown_tool_404() -> None:
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    app, *_ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}))

    async with _client(app) as client:
        resp = await client.post(f"/projects/{pid}/skills/nope/generate", json={"brief": "x"})

    assert resp.status_code == 404
    assert resp.json()["error"]["code"] == ErrorCode.NOT_FOUND


@pytest.mark.asyncio
async def test_generate_legacy_tool_404() -> None:
    # legacy 工具无通用执行路径（前端走 legacy_route）。
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    app, *_ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}))

    async with _client(app) as client:
        resp = await client.post(
            f"/projects/{pid}/skills/worldbuilding/generate", json={"brief": "x"}
        )

    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_generate_unknown_project_404() -> None:
    repo = FakeProjectRepo()
    app, *_ = _make_app(
        project_repo=repo, gateway=_SchemaRoutingGateway({IdeaListResult: _ideas()})
    )

    async with _client(app) as client:
        resp = await client.post(
            f"/projects/{uuid.uuid4()}/skills/brainstorm/generate", json={"brief": "x"}
        )

    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_generate_no_credentials_503() -> None:
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    app, session, *_ = _make_app(project_repo=repo, gateway=object(), no_creds=True)

    async with _client(app) as client:
        resp = await client.post(f"/projects/{pid}/skills/brainstorm/generate", json={"brief": "x"})

    assert resp.status_code == 503
    assert resp.json()["error"]["code"] == ErrorCode.LLM_UNAVAILABLE
    assert session.commits == 0


# ---- POST .../generate（Scope B 竞品快赢：续写/扩写/降AI/拆书）----


@pytest.mark.asyncio
async def test_generate_continue_reads_prior_chapter() -> None:
    # 续写：端点读该章最新 accepted 正文 → 注入 context（with_prior_chapter 路由）。
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    chapter_repo = _FakeChapterRepo(accepted={2: "他握紧了那柄剑，缓缓转身。"})
    gateway = _CaptureGateway(ContinuationResult(text="剑光一闪，敌人应声倒地。"))
    app, session, *_ = _make_app(project_repo=repo, gateway=gateway, chapter_repo=chapter_repo)

    async with _client(app) as client:
        resp = await client.post(
            f"/projects/{pid}/skills/continue/generate", json={"chapter_no": 2, "brief": "续写战斗"}
        )

    assert resp.status_code == 200
    body = resp.json()
    assert body["tool_key"] == "continue"
    assert body["output_kind"] == "ContinuationResult"
    assert body["preview"]["text"] == "剑光一闪，敌人应声倒地。"
    # 前文正文确实进了喂网关的 context（读章路由生效）。
    assert gateway.last_input is not None
    assert "他握紧了那柄剑" in gateway.last_input
    assert session.commits == 1  # 预览不写业务表，落 ledger


@pytest.mark.asyncio
async def test_generate_continue_falls_back_to_draft() -> None:
    # 无 accepted → 退回草稿正文。
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    chapter_repo = _FakeChapterRepo(drafts={1: "草稿里的开头一段。"})
    gateway = _CaptureGateway(ContinuationResult(text="续写。"))
    app, *_ = _make_app(project_repo=repo, gateway=gateway, chapter_repo=chapter_repo)

    async with _client(app) as client:
        resp = await client.post(
            f"/projects/{pid}/skills/continue/generate", json={"chapter_no": 1}
        )

    assert resp.status_code == 200
    assert gateway.last_input is not None
    assert "草稿里的开头一段" in gateway.last_input


@pytest.mark.asyncio
async def test_generate_expand_routes_text_input() -> None:
    # 扩写：原文经 text 字段 → 注入 context（text_input 路由）。
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    gateway = _CaptureGateway(PolishResult(text="扩写后的丰富正文。"))
    app, session, *_ = _make_app(project_repo=repo, gateway=gateway)

    async with _client(app) as client:
        resp = await client.post(
            f"/projects/{pid}/skills/expand/generate",
            json={"text": "原始简短的一句话。", "brief": "加强描写"},
        )

    assert resp.status_code == 200
    body = resp.json()
    assert body["output_kind"] == "PolishResult"
    assert body["preview"]["text"] == "扩写后的丰富正文。"
    assert gateway.last_input is not None
    assert "原始简短的一句话" in gateway.last_input
    assert session.commits == 1


@pytest.mark.asyncio
async def test_generate_de_ai_routes_text_input() -> None:
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    gateway = _CaptureGateway(DeAiResult(text="更自然的人写质感正文。"))
    app, *_ = _make_app(project_repo=repo, gateway=gateway)

    async with _client(app) as client:
        resp = await client.post(
            f"/projects/{pid}/skills/de-ai/generate",
            json={"text": "充斥着 AI 腔的排比句。"},
        )

    assert resp.status_code == 200
    assert resp.json()["output_kind"] == "DeAiResult"
    assert gateway.last_input is not None
    assert "充斥着 AI 腔的排比句" in gateway.last_input


@pytest.mark.asyncio
async def test_generate_teardown_returns_structured() -> None:
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    parsed = BookTeardownResult(
        themes=["逆袭"],
        archetypes=["废柴主角"],
        structure="黄金三章立钩",
        hooks=["开局即巅峰"],
    )
    gateway = _CaptureGateway(parsed)
    app, *_ = _make_app(project_repo=repo, gateway=gateway)

    async with _client(app) as client:
        resp = await client.post(
            f"/projects/{pid}/skills/teardown/generate",
            json={"kind": "某爆款", "text": "样本章节正文……"},
        )

    assert resp.status_code == 200
    body = resp.json()
    assert body["output_kind"] == "BookTeardownResult"
    assert body["preview"]["themes"] == ["逆袭"]
    assert body["preview"]["structure"] == "黄金三章立钩"


# ---- POST .../ingest（world_entities gate）----


def _gf_payload() -> dict[str, Any]:
    return {
        "world_entities": [
            {"type": "力量体系", "name": "吞噬系统", "rules": ["每日上限", "不可越级"]}
        ]
    }


@pytest.mark.asyncio
async def test_ingest_world_entities_no_conflict_writes_201() -> None:
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    gateway = _SchemaRoutingGateway({ContinuityReview: _no_conflicts()})
    app, session, world_repo, _ = _make_app(project_repo=repo, gateway=gateway)

    async with _client(app) as client:
        resp = await client.post(f"/projects/{pid}/skills/golden-finger/ingest", json=_gf_payload())

    assert resp.status_code == 201
    body = resp.json()
    assert body["table"] == "world_entities"
    assert body["created"] == ["吞噬系统"]
    assert body["rejected_tables"] == []
    assert len(world_repo.rows) == 1
    assert world_repo.rows[0]["rules"] == ["每日上限", "不可越级"]
    assert session.commits == 1


@pytest.mark.asyncio
async def test_ingest_world_entities_conflict_blocks_409() -> None:
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    gateway = _SchemaRoutingGateway({ContinuityReview: _with_conflicts()})
    app, session, world_repo, _ = _make_app(project_repo=repo, gateway=gateway)

    async with _client(app) as client:
        resp = await client.post(f"/projects/{pid}/skills/golden-finger/ingest", json=_gf_payload())

    assert resp.status_code == 409
    err = resp.json()["error"]
    assert err["code"] == ErrorCode.CONFLICT_UNRESOLVED
    assert err["details"]["conflict_count"] == 1
    assert len(world_repo.rows) == 0
    assert session.commits == 1  # 落 precheck ledger，不写业务表


@pytest.mark.asyncio
async def test_ingest_world_entities_acknowledged_writes() -> None:
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    gateway = _SchemaRoutingGateway({ContinuityReview: _with_conflicts()})
    app, _, world_repo, _ = _make_app(project_repo=repo, gateway=gateway)

    payload = {**_gf_payload(), "acknowledge_conflicts": True}
    async with _client(app) as client:
        resp = await client.post(f"/projects/{pid}/skills/golden-finger/ingest", json=payload)

    assert resp.status_code == 201
    assert len(world_repo.rows) == 1


@pytest.mark.asyncio
async def test_ingest_non_ingestable_tool_422() -> None:
    # brainstorm 纯预览 → 不可入库。
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    app, *_ = _make_app(project_repo=repo, gateway=_SchemaRoutingGateway({}))

    async with _client(app) as client:
        resp = await client.post(f"/projects/{pid}/skills/brainstorm/ingest", json={})

    assert resp.status_code == 422
    assert resp.json()["error"]["code"] == ErrorCode.VALIDATION


@pytest.mark.asyncio
async def test_ingest_outline_upserts_scenes() -> None:
    repo = FakeProjectRepo()
    pid = await _seed_project(repo)
    gateway = _SchemaRoutingGateway({})
    app, session, _, outline_repo = _make_app(project_repo=repo, gateway=gateway)

    payload = {
        "chapter_no": 3,
        "scenes": [
            {"idx": 1, "beat": "初遇反派", "purpose": "推进", "conflict": "对峙", "hook": "悬念"},
            {"idx": 0, "beat": "主角觉醒", "purpose": "塑造", "conflict": "内心", "hook": "钩子"},
        ],
    }
    async with _client(app) as client:
        resp = await client.post(f"/projects/{pid}/skills/fine-outline/ingest", json=payload)

    assert resp.status_code == 201
    body = resp.json()
    assert body["table"] == "outline"
    assert len(outline_repo.rows) == 1
    # 场景按 idx 升序拼装成 beats（确定性）。
    assert outline_repo.rows[0]["beats"] == ["主角觉醒", "初遇反派"]
    assert outline_repo.rows[0]["chapter_no"] == 3
    assert session.commits == 1