writer-work-flow/tests/test_m4_e2e.py

"""M4 端到端：学文风(指纹+证据,版本化) + 漂移打分(第四审) + 回炉（真实 DB，零 token）。

证明 M4 闭环（DoD：学文风指纹+证据；漂移打分作第四审真持久化；回炉返新旧 diff）：
  用例 1 · 学文风闭环（jobs 异步长任务，M4-c）
    `POST /projects` → `POST .../style`(samples) → **202 `{job_id}`**（写一行 job 返回）；
    BackgroundTask 经 `run_job` 自建独立 session 跑 `run_style_extraction`(analyst 假适配器
    产 `StyleFingerprintResult`)→拆指纹→`SqlStyleFingerprintWriteRepo.append`(真落 pg)。
    ASGITransport 下 `await client.post` 等 background task 跑完（同 M3 时序坑）；override
    `get_session_factory`→`e2e_sm`(真 sessionmaker/同测试 engine/loop)，并 monkeypatch
    `style.build_gateway_for_tier`→真 Gateway 包假适配器(用 background task 的真 session 记账)。
    → `GET /jobs/{id}` status=done(result.version/dims_count) → 断言 `style_fingerprint` 真有行
    (dimensions_json/evidence_json/version)。再 `mode="update"` 学一次 → version+1。
  用例 2 · 写章第四审（drift scoring）
    学文风后 `assemble` 的 `stable_core` 含指纹 → 第四审能在 review_context 拿指纹对照。
    `POST .../chapters/N/review`(SSE) → 断言 `section{name:"style"}` 帧 + `style{score,segments}`
    帧真发；`GET .../reviews` 留痕行 `style`(dict) 已填；DB 真源 `chapter_reviews.style` 真落库
    (score + segments)。四审齐（continuity/foreshadow/pace/style）无 error。
  用例 3 · 回炉（refine，非持久 / 不变量 #3）
    `POST .../chapters/N/refine` → 200 `{original, refined}`；断言 `refined != original`
    （writer 假适配器据 `req.output_schema is None` 返改写串）+ 不写库。

确定性 & 零成本（同 M1/M2/M3）：真实 `Gateway` + 多档位假适配器（据 `req.output_schema`
分支返回固定 `parsed`/`text`，绝不联网）+ 真实 `SqlAlchemyLedgerSink`。无 pg → skip。

坑（见 memory/gotchas）：
- 学文风 BackgroundTask 在 `run_job` 自建独立 session 上**自造网关**(`build_gateway_for_tier`,
  从凭据建真 OpenAI 适配器) → E2E 无真凭据，须 monkeypatch `style.build_gateway_for_tier`
  返真 `Gateway`(包假适配器, ledger 绑入参 session)；写侧 `SqlStyleFingerprintWriteRepo` 保持
  真实 → 指纹真落 pg。`run_job` 默认 `repo_factory`→真 `SqlJobRepo`（落真 pg job 行），不动。
- `get_session_factory` 必 override → `e2e_sm`（真 sessionmaker，同测试 engine/loop），否则
  默认 `get_sessionmaker()` 另建 engine 绑别的 loop（跨 loop 坑）。
- `get_sessionmaker` engine 绑首个 loop → 每 DB 测试清缓存重建、结束 dispose（同 M3）。
- ASGITransport 默认 `raise_app_exceptions=True`；SSE 解析 `event:`/`data:` 块；snake_case。
"""

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 (
    ContinuityReview,
    ForeshadowReview,
    ForeshadowSuggestion,
    PaceIssue,
    PaceReview,
    StyleDimension,
    StyleDriftReview,
    StyleDriftSegment,
    StyleFingerprintResult,
)
from ww_api.services.digest_extraction import ChapterDigestFacts
from ww_db import get_session, get_sessionmaker
from ww_db.models import (
    Chapter,
    ChapterDigest,
    ChapterReview,
    Foreshadow,
    Job,
    Outline,
    Project,
    StyleFingerprint,
    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 = ["第", "一", "章", "：", "文风漂移之时。"]

# 学文风假产物：两维指纹，每维带原文证据（StyleFingerprintResult 提取轨）。
_FINGERPRINT = StyleFingerprintResult(
    dimensions=[
        StyleDimension(name="句长节奏", value="短句为主", evidence=["他来了。", "她走了。"]),
        StyleDimension(name="叙事人称", value="第三人称", evidence=["他看着远方的山。"]),
    ]
)

# 第四审（文风漂移）假产物：整体相似度 + 一个低相似段。
_STYLE_DRIFT = StyleDriftReview(
    score=72,
    segments=[StyleDriftSegment(idx=2, score=40, label="机翻腔")],
)

# 三审固定产物（与 M3 一致，让四审图跑通）。
_FORESHADOW_REVIEW = ForeshadowReview(
    planted=[ForeshadowSuggestion(code="F9", title="新埋：山门禁制", where="第2段")],
    resolved=[],
)
_PACE_REVIEW = PaceReview(
    water=[PaceIssue(where="第3段", reason="信息密度低")],
    hook=True,
    beat_map=[1, 3, 5, 2],
)

# 回炉改写标记（writer 纯文本返回）：保证 refined != original。
_REFINE_PREFIX = "【已回炉】"

# 各档位假用量（喂记账；证明各档位调用各自落库）。
_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),
}


def _req_text(req: LlmRequest) -> str:
    """把 `LlmRequest.input`（str | list[Block]）压成纯文本（供回炉改写引用原文）。"""
    if isinstance(req.input, str):
        return req.input
    return "\n".join(block.text for block in req.input)


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

    - `complete()`：据 `req.output_schema` 路由：
      `StyleFingerprintResult`→学文风提取（analyst）；四审 schema
      （`ContinuityReview`/`ForeshadowReview`/`PaceReview`/`StyleDriftReview`）→各审；
      `output_schema is None`→**回炉纯文本**（writer，返改写串，保证 refined≠original）；
      其余视作 digest 提炼（light）。
    - `stream()`：吐固定写章 token（writer/draft），末尾喂 writer 用量。
    每路带 `ProviderUsage` → 网关据此各落一条 usage_ledger。
    """

    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 None:
            # 回炉：writer 纯文本重写选中段（保证 refined != original）。
            return ProviderResult(
                text=_REFINE_PREFIX + _req_text(req),
                parsed=None,
                usage=_USAGE["writer"],
            )
        if schema is StyleFingerprintResult:
            return ProviderResult(
                text=_FINGERPRINT.model_dump_json(),
                parsed=_FINGERPRINT,
                usage=_USAGE["analyst"],
            )
        if schema is StyleDriftReview:
            return ProviderResult(
                text=_STYLE_DRIFT.model_dump_json(),
                parsed=_STYLE_DRIFT,
                usage=_USAGE["light"],
            )
        if schema is ContinuityReview:
            review = ContinuityReview(conflicts=[])  # 零冲突
            return ProviderResult(
                text=review.model_dump_json(), parsed=review, usage=_USAGE["analyst"]
            )
        if schema is ForeshadowReview:
            return ProviderResult(
                text=_FORESHADOW_REVIEW.model_dump_json(),
                parsed=_FORESHADOW_REVIEW,
                usage=_USAGE["analyst"],
            )
        if schema is PaceReview:
            return ProviderResult(
                text=_PACE_REVIEW.model_dump_json(), parsed=_PACE_REVIEW, usage=_USAGE["light"]
            )
        # 否则视为 digest 提炼（ChapterDigestFacts）。
        facts = ChapterDigestFacts(summary="终稿摘要", 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）。"""

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

    return _override


def _patched_build_gateway(
    adapter: _FakeMultiTierAdapter,
) -> Callable[..., object]:
    """构造 monkeypatch 替身：`style.build_gateway_for_tier`→真 Gateway 包假适配器。

    学文风 BackgroundTask 在 `run_job` 自建的独立 session 上自造网关（从凭据建真 OpenAI
    适配器）——E2E 无真凭据，故替换成返真 `Gateway`(包假适配器, ledger 绑 background task
    的真 session)。写侧 `SqlStyleFingerprintWriteRepo` + `run_job` 默认 `SqlJobRepo` 保持
    真实 → 指纹/job 真落 pg。
    """

    async def _build(session: AsyncSession, store: object, tier: str) -> Gateway:
        return Gateway(
            adapters={_PROVIDER: adapter},
            ledger=SqlAlchemyLedgerSink(session),
            resolver=resolve_route,
        )

    return _build


async def _cleanup(e2e_sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
    """按 FK 顺序清理（多数子表经 FK CASCADE；显式删可独立行）。"""
    async with e2e_sm() as cleanup:
        await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
        await cleanup.execute(delete(Job).where(Job.project_id == project_uuid))
        await cleanup.execute(
            delete(StyleFingerprint).where(StyleFingerprint.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(Outline).where(Outline.project_id == project_uuid))
        await cleanup.execute(delete(Foreshadow).where(Foreshadow.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_m4_learn_style_job_done_and_fingerprint_persisted(
    e2e_sm: async_sessionmaker[AsyncSession],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """用例 1：学文风闭环 → 202 job → 后台跑完 → job done → 指纹真落 pg（含 update version+1）。"""
    import ww_api.routers.style as style_mod
    from ww_api.main import create_app
    from ww_api.services.project_deps import (
        get_session_factory,
        get_style_extract_gateway,
    )

    adapter = _FakeMultiTierAdapter()
    override = _gateway_override_factory(adapter)

    app = create_app()
    # 凭据探测（请求阶段）→ 注假网关，绕过无凭据 503。
    app.dependency_overrides[get_style_extract_gateway] = override
    # BackgroundTask 独立 session 工厂 → 真 sessionmaker（同测试 engine/loop）。
    app.dependency_overrides[get_session_factory] = lambda: e2e_sm
    # BackgroundTask 自造网关 → 替成真 Gateway 包假适配器（写侧 repo + SqlJobRepo 保持真）。
    monkeypatch.setattr(style_mod, "build_gateway_for_tier", _patched_build_gateway(adapter))

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

                # 1) 学文风（首学）→ 202 {job_id}（ASGITransport 等 background task 跑完）。
                learn_resp = await client.post(
                    f"/projects/{project_id}/style",
                    json={"samples": ["他来了。她走了。", "他看着远方的山。"], "mode": "create"},
                )
                assert learn_resp.status_code == 202
                job_id = learn_resp.json()["job_id"]

                # 2) 轮询 job → done，result 含 version/dims_count。
                job_resp = await client.get(f"/jobs/{job_id}")
                assert job_resp.status_code == 200
                job = job_resp.json()
                assert job["status"] == "done"
                assert job["kind"] == "style_learn"
                assert job["result"]["version"] == 1
                assert job["result"]["dims_count"] == 2

                # 3) GET /style → 最新指纹（完整维度 + 证据 + 版本）。
                style_resp = await client.get(f"/projects/{project_id}/style")
                assert style_resp.status_code == 200
                fp = style_resp.json()
                assert fp["version"] == 1
                # 新契约：dimensions 为 list[DimensionEntry{name,value,evidence}]（不再裸 dict）。
                dims = {d["name"]: d for d in fp["dimensions"]}
                assert dims["句长节奏"]["value"] == "短句为主"
                assert dims["句长节奏"]["evidence"] == ["他来了。", "她走了。"]

                # 4) mode="update" 再学一次 → version+1。
                update_resp = await client.post(
                    f"/projects/{project_id}/style",
                    json={"samples": ["新样本正文。"], "mode": "update"},
                )
                assert update_resp.status_code == 202
                update_job = (await client.get(f"/jobs/{update_resp.json()['job_id']}")).json()
                assert update_job["status"] == "done"
                assert update_job["result"]["version"] == 2

        # DB 真源断言。
        assert project_uuid is not None
        async with e2e_sm() as verify:
            fp_rows = (
                (
                    await verify.execute(
                        select(StyleFingerprint)
                        .where(StyleFingerprint.project_id == project_uuid)
                        .order_by(StyleFingerprint.version)
                    )
                )
                .scalars()
                .all()
            )
            # 两次学文风 → 两行（版本化 INSERT，不覆盖历史）。
            assert [r.version for r in fp_rows] == [1, 2]
            first = fp_rows[0]
            assert first.dimensions_json == {"句长节奏": "短句为主", "叙事人称": "第三人称"}
            assert first.evidence_json["句长节奏"] == ["他来了。", "她走了。"]

            # job 真落 pg：两行 done。
            job_rows = (
                (await verify.execute(select(Job).where(Job.project_id == project_uuid)))
                .scalars()
                .all()
            )
            assert len(job_rows) == 2
            assert all(r.status == "done" and r.kind == "style_learn" for r in job_rows)
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_m4_review_fourth_style_audit_sse_and_trace(
    e2e_sm: async_sessionmaker[AsyncSession],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """用例 2：写章第四审（drift）→ SSE section{style} + style{score,segments} + style 列真填。

    先学文风 → `assemble` 的 stable_core 含指纹 → 第四审能在 review_context 拿指纹对照打分。
    四审齐（continuity/foreshadow/pace/style）无 error。
    """
    import ww_api.routers.style as style_mod
    from ww_api.main import create_app
    from ww_api.services.project_deps import (
        get_review_gateway,
        get_session_factory,
        get_style_extract_gateway,
        get_writer_gateway,
    )

    adapter = _FakeMultiTierAdapter()
    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_style_extract_gateway] = override
    app.dependency_overrides[get_session_factory] = lambda: e2e_sm
    monkeypatch.setattr(style_mod, "build_gateway_for_tier", _patched_build_gateway(adapter))

    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": "M4 第四审验证作品"})
                assert create_resp.status_code == 201
                project_id = create_resp.json()["id"]
                project_uuid = uuid.UUID(project_id)

                # 1) 学文风 → 指纹入库（供 assemble 注入第四审做对照）。
                learn_resp = await client.post(
                    f"/projects/{project_id}/style",
                    json={"samples": ["他来了。她走了。"], "mode": "create"},
                )
                assert learn_resp.status_code == 202
                assert (await client.get(f"/jobs/{learn_resp.json()['job_id']}")).json()[
                    "status"
                ] == "done"

                # 2) 写第1章草稿 + 自动保存（供续审）。
                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},
                )

                # 3) 续审（SSE）：四审齐 + section{style} + style{score,segments}，无 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)
                assert [d for (ev, d) in rframes if ev == "error"] == []
                section_names = {
                    json.loads(d).get("name") for (ev, d) in rframes if ev == "section"
                }
                assert {"continuity", "foreshadow", "pace", "style"} <= section_names
                style_events = [json.loads(d) for (ev, d) in rframes if ev == "style"]
                assert len(style_events) == 1
                style_ev = style_events[0]
                assert style_ev["score"] == 72
                assert style_ev["segments"][0]["idx"] == 2
                assert style_ev["segments"][0]["score"] == 40
                assert style_ev["segments"][0]["label"] == "机翻腔"

                # 4) GET 历史：留痕行 style(dict) 已填。
                hist_resp = await client.get(f"/projects/{project_id}/chapters/1/reviews")
                assert hist_resp.status_code == 200
                latest = hist_resp.json()["reviews"][0]
                assert isinstance(latest["style"], dict)
                assert latest["style"]["score"] == 72

        # DB 真源断言：chapter_reviews.style 列真填。
        assert project_uuid is not None
        async with e2e_sm() as verify:
            review_rows = (
                (
                    await verify.execute(
                        select(ChapterReview).where(ChapterReview.project_id == project_uuid)
                    )
                )
                .scalars()
                .all()
            )
            assert len(review_rows) == 1
            row = review_rows[0]
            assert isinstance(row.style, dict)
            assert row.style["score"] == 72
            assert row.style["segments"][0]["idx"] == 2
            assert row.style["segments"][0]["label"] == "机翻腔"
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_m4_refine_returns_diff_without_persisting(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 3：回炉 → 200 {original, refined}；refined != original；不写库（不变量 #3）。"""
    from ww_api.main import create_app
    from ww_api.services.project_deps import get_refine_gateway

    adapter = _FakeMultiTierAdapter()
    override = _gateway_override_factory(adapter)

    app = create_app()
    app.dependency_overrides[get_refine_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": "M4 回炉验证作品"})
                assert create_resp.status_code == 201
                project_id = create_resp.json()["id"]
                project_uuid = uuid.UUID(project_id)

                segment = "他緩緩走進房間。"
                refine_resp = await client.post(
                    f"/projects/{project_id}/chapters/1/refine",
                    json={"segment": segment, "instruction": "去掉机翻腔"},
                )
                assert refine_resp.status_code == 200
                body = refine_resp.json()
                assert body["original"] == segment
                # writer 假适配器返改写串（前缀），保证 refined != original。
                assert body["refined"] != body["original"]
                assert body["refined"].startswith(_REFINE_PREFIX)
                # 指令进了输入文本（writer 据 output_schema is None 走纯文本路径）。
                assert "去掉机翻腔" in body["refined"]

        # DB 真源断言：回炉不写库 → 无 chapter/chapter_review 行（不变量 #3）。
        assert project_uuid is not None
        async with e2e_sm() as verify:
            chapter_rows = (
                (await verify.execute(select(Chapter).where(Chapter.project_id == project_uuid)))
                .scalars()
                .all()
            )
            assert chapter_rows == []
            review_rows = (
                (
                    await verify.execute(
                        select(ChapterReview).where(ChapterReview.project_id == project_uuid)
                    )
                )
                .scalars()
                .all()
            )
            assert review_rows == []
            # 但回炉记账落库（writer 网关 ledger，端点末尾 commit）。
            ledger_rows = (
                (
                    await verify.execute(
                        select(UsageLedger).where(UsageLedger.project_id == project_uuid)
                    )
                )
                .scalars()
                .all()
            )
            assert len(ledger_rows) == 1
            assert ledger_rows[0].input_tokens == _USAGE["writer"].input_tokens
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)