writer-work-flow/tests/test_chain_workflow_e2e.py

"""C4 多章工作流链 端到端（chain-workflow §9 C4 / §10）——真 pg + mock 网关零 token。

证明杀手锏闭环（不变量 #1/#3/#4/#5/#9）：批量量产，但每章过 `write→四审→decide→accept`，
遇冲突 interrupt 交人裁决再 resume 续跑——星月架构做不到的一致性闸。

  用例 1 · 两章无冲突全自动
    `POST /projects` → `POST .../chains/draft_volume/run {start=1,count=2}` → 202 {job_id}；
    BackgroundTask `run_chain_job` 自建独立 session 驱动链图（mock 网关零冲突）→ 两章
    `write→四审(无冲突)→accept` 自动跑完。DB 真源逐项断言：
      - `chapters` 第 1/2 章各一行 `accepted`（write 落 draft → accept 晋升终稿）；
      - `chapter_digests` 两行（digest 从终稿提炼，不变量 #4）；
      - `GET /jobs/{id}` status=done，result.written=[1,2]、completed=True、awaiting_chapter=None。
    负向：job result/状态绝不含正文/prompt/token（§5）。

  用例 2 · 注入冲突 → interrupt → awaiting → resume → accept
    续审第 1 章报冲突 → `decide` 命中 → `accept` 节点 `interrupt()` 暂停 → job=awaiting_input、
    result.awaiting_chapter=1、`chapters` 第 1 章仍 `draft`（未晋升）、`chapter_digests` 空。
    `POST .../chains/runs/{job_id}/resume {decisions:[...]}` → 202 → resume 续跑 → accept →
    job done、第 1 章 `accepted`、digest 一行。
    interrupt+resume 横跨**两次端点调用**，靠单进程单 `MemorySaver`（经依赖覆盖注入
    `get_checkpointer_factory` 的同一 saver 实例）持久控制流位置 + pending handle。

确定性 & 零成本（同 M3/M4）：真 `Gateway` + 假适配器（据 `req.output_schema` 分支返固定
parsed/text，绝不联网）+ 真 `SqlAlchemyLedgerSink`。检查点用 MemorySaver（绝不连真 Postgres
检查点）。无 pg → skip。

坑（见 memory/gotchas）：
- `get_checkpointer_factory` 必 override 成产**同一** MemorySaver 的工厂——run 与 resume 两次
  端点调用须落同一 thread_id 的同一检查点存储，否则 resume 找不到 interrupt 现场。
- `get_session_factory` 必 override → `e2e_sm`（真 sessionmaker，同测试 engine/loop）；
  BackgroundTask 在请求 session 关闭后才跑。
- `get_digest_gateway_builder` 必 override → 返「按 session 建真 Gateway 包假适配器」的 builder
  （链 accept 节点终稿提炼用 light 档），否则会从空凭据建真 OpenAI 适配器。
- ASGITransport 下 `await client.post` 等 ASGI app（含 background task）跑完才返回。
"""

from __future__ import annotations

import uuid
from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from typing import Annotated, Any

import httpx
import pytest
from asgi_lifespan import LifespanManager
from fastapi import Depends
from langgraph.checkpoint.memory import MemorySaver
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from ww_agents import (
    Conflict,
    ContinuityReview,
    ForeshadowReview,
    PaceReview,
    StyleDriftReview,
)
from ww_api.services.digest_extraction import ChapterDigestFacts
from ww_db import get_session, get_sessionmaker
from ww_db.models import (
    Chapter,
    ChapterDigest,
    ChapterReview,
    Job,
    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"

# 链写章收集版正文（非 SSE，经 gateway.run → adapter.complete，output_schema is None）。
_CHAPTER_TEXT = "灵气如潮，少年握剑而立，山门之上风雷大作。"

# 各档假用量（喂记账；证明各档调用各落一条 usage_ledger）。
_USAGE = ProviderUsage(input_tokens=13, output_tokens=7)

# 三审固定无冲突产物（让四审图跑通；continuity 的冲突由 fixture 参数注入）。
_FORESHADOW_REVIEW = ForeshadowReview(planted=[], resolved=[])
_PACE_REVIEW = PaceReview(water=[], hook=True, beat_map=[1, 2, 3])
_STYLE_DRIFT = StyleDriftReview(score=88, segments=[])


class _FakeChainAdapter:
    """实现 `ProviderAdapter`：据 `req.output_schema` 分支返 parsed/text，绝不联网。

    - `output_schema is None` → 链写章收集版纯文本（writer 档，`gateway.run`）。
    - 四审 schema → 各审 parsed（continuity 注入冲突，其余无冲突）。
    - `ChapterDigestFacts` → 终稿 digest 提炼（light 档）。
    `stream()` 实现但链不走（收集版量产非 SSE）。
    """

    provider = _PROVIDER

    def __init__(self, *, continuity_conflicts: list[Conflict]) -> None:
        self._continuity = ContinuityReview(conflicts=continuity_conflicts)

    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:
            return ProviderResult(text=_CHAPTER_TEXT, parsed=None, usage=_USAGE)
        if schema is ContinuityReview:
            return ProviderResult(
                text=self._continuity.model_dump_json(), parsed=self._continuity, usage=_USAGE
            )
        if schema is ForeshadowReview:
            return ProviderResult(
                text=_FORESHADOW_REVIEW.model_dump_json(), parsed=_FORESHADOW_REVIEW, usage=_USAGE
            )
        if schema is PaceReview:
            return ProviderResult(
                text=_PACE_REVIEW.model_dump_json(), parsed=_PACE_REVIEW, usage=_USAGE
            )
        if schema is StyleDriftReview:
            return ProviderResult(
                text=_STYLE_DRIFT.model_dump_json(), parsed=_STYLE_DRIFT, usage=_USAGE
            )
        # 否则视为终稿 digest 提炼（ChapterDigestFacts）。
        facts = ChapterDigestFacts(summary="终稿摘要", events=["少年出山"], locations=["山门"])
        return ProviderResult(text=facts.model_dump_json(), parsed=facts, usage=_USAGE)

    async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
        yield StreamChunk(text=_CHAPTER_TEXT)
        yield StreamChunk(usage=_USAGE)


@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 _gateway_for(adapter: _FakeChainAdapter, session: AsyncSession) -> Gateway:
    """真 Gateway + 给定假适配器 + 真 ledger（绑给定 session）。"""
    return Gateway(
        adapters={_PROVIDER: adapter},
        ledger=SqlAlchemyLedgerSink(session),
        resolver=resolve_route,
    )


def _chain_gateway_override(
    adapter: _FakeChainAdapter,
) -> Callable[[AsyncSession], Gateway]:
    """`get_chain_gateway` 覆盖：请求阶段凭据探测 → 真 Gateway 包假适配器（绕过 503）。"""

    def _override(session: Annotated[AsyncSession, Depends(get_session)]) -> Gateway:
        return _gateway_for(adapter, session)

    return _override


def _digest_builder_override(
    adapter: _FakeChainAdapter,
) -> Callable[[], Callable[[AsyncSession], Awaitable[Gateway]]]:
    """`get_digest_gateway_builder` 覆盖：返「按 session 建真 Gateway 包假适配器」的 builder。

    链 accept 节点在自建短事务里据该 session 现建 light 档 digest 网关（终稿提炼）。
    """

    def _get_builder() -> Callable[[AsyncSession], Awaitable[Gateway]]:
        async def _build(session: AsyncSession) -> Gateway:
            return _gateway_for(adapter, session)

        return _build

    return _get_builder


def _memsaver_override(
    saver: MemorySaver,
) -> Callable[[], Callable[[], AbstractAsyncContextManager[MemorySaver]]]:
    """`get_checkpointer_factory` 覆盖：返产**同一** saver 的工厂（run/resume 跨调用共用）。"""

    @asynccontextmanager
    async def _ctx() -> AsyncIterator[MemorySaver]:
        yield saver

    return lambda: lambda: _ctx()


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


def _build_app(adapter: _FakeChainAdapter, saver: MemorySaver, e2e_sm: Any) -> Any:
    """组装注入了全 fake 缝的 app（凭据探测/session 工厂/checkpointer/digest builder）。"""
    from ww_api.main import create_app
    from ww_api.services.chain_deps import get_checkpointer_factory
    from ww_api.services.project_deps import (
        get_chain_gateway,
        get_chain_gateway_builder,
        get_digest_gateway_builder,
        get_session_factory,
    )

    app = create_app()
    app.dependency_overrides[get_chain_gateway] = _chain_gateway_override(adapter)
    # 链 write/review 节点经 builder 在节点 session 上现建网关（ledger 绑节点 session，不变量 #1）。
    app.dependency_overrides[get_chain_gateway_builder] = _digest_builder_override(adapter)
    app.dependency_overrides[get_session_factory] = lambda: e2e_sm
    app.dependency_overrides[get_checkpointer_factory] = _memsaver_override(saver)
    app.dependency_overrides[get_digest_gateway_builder] = _digest_builder_override(adapter)
    return app


async def test_chain_two_chapters_no_conflict_full_auto(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 1：两章无冲突全自动 → 两 accepted 章 + 两 digest 行 + job done written=[1,2]。"""
    adapter = _FakeChainAdapter(continuity_conflicts=[])  # 零冲突 → 全自动
    saver = MemorySaver()
    app = _build_app(adapter, saver, e2e_sm)

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

                run_resp = await client.post(
                    f"/projects/{project_id}/chains/draft_volume/run",
                    json={"start_chapter_no": 1, "count": 2},
                )
                assert run_resp.status_code == 202
                run_body = run_resp.json()
                assert run_body["chain_key"] == "draft_volume"
                job_id = run_body["job_id"]

                # 链经 BackgroundTask 已跑完（ASGITransport 等其完成）→ job done。
                job_resp = await client.get(f"/jobs/{job_id}")
                assert job_resp.status_code == 200
                job = job_resp.json()
                assert job["status"] == "done", job
                assert job["kind"] == "chain"
                result = job["result"]
                assert result["written"] == [1, 2]
                assert result["completed"] is True
                assert result["awaiting_chapter"] is None
                # 负向：job result 绝不含正文/token（只章号/计数/标志）。
                assert _CHAPTER_TEXT not in str(result)
                assert "终稿摘要" not in str(result)

        # DB 真源逐项断言。
        assert project_uuid is not None
        async with e2e_sm() as verify:
            chapters = (
                (
                    await verify.execute(
                        select(Chapter)
                        .where(Chapter.project_id == project_uuid)
                        .order_by(Chapter.chapter_no, Chapter.version)
                    )
                )
                .scalars()
                .all()
            )
            # 第 1/2 章各晋升一个 accepted 版次。
            accepted = [c for c in chapters if c.status == "accepted"]
            assert {c.chapter_no for c in accepted} == {1, 2}
            assert all(c.content == _CHAPTER_TEXT for c in accepted)

            digests = (
                (
                    await verify.execute(
                        select(ChapterDigest)
                        .where(ChapterDigest.project_id == project_uuid)
                        .order_by(ChapterDigest.chapter_no)
                    )
                )
                .scalars()
                .all()
            )
            assert [d.chapter_no for d in digests] == [1, 2]

            # 回归守卫（审评 #1）：链每次 gateway.run 都应落 usage_ledger 行——证明网关 ledger
            # 绑节点活 session、节点 commit 持久化。
            # 修复前：请求网关 session 已关闭，行被静默丢弃。
            ledger_rows = (
                (
                    await verify.execute(
                        select(UsageLedger).where(UsageLedger.project_id == project_uuid)
                    )
                )
                .scalars()
                .all()
            )
            # 两章 × (1 写 + 四审 + 1 digest)；至少应有若干条，绝不为空。
            assert len(ledger_rows) > 0, "chain usage_ledger 行不应为空（成本记账断裂回归守卫）"

            # 每审过一章应在 chapter_reviews 留痕（不变量 #3 只读留痕；审评 #7b）。
            review_rows = (
                (
                    await verify.execute(
                        select(ChapterReview)
                        .where(ChapterReview.project_id == project_uuid)
                        .order_by(ChapterReview.chapter_no)
                    )
                )
                .scalars()
                .all()
            )
            assert {r.chapter_no for r in review_rows} == {1, 2}

            job_rows = (
                (await verify.execute(select(Job).where(Job.project_id == project_uuid)))
                .scalars()
                .all()
            )
            assert len(job_rows) == 1
            assert job_rows[0].status == "done"
            assert job_rows[0].kind == "chain"
            # 负向（DB 真源）：job result 列不含正文/token。
            assert _CHAPTER_TEXT not in str(job_rows[0].result)
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)


async def test_chain_conflict_interrupts_then_resume_accepts(
    e2e_sm: async_sessionmaker[AsyncSession],
) -> None:
    """用例 2：注入冲突 → interrupt → job awaiting → resume 带裁决 → 续跑 → accept。

    interrupt+resume 横跨两次端点调用，靠单进程单 MemorySaver（同一 thread_id 检查点）。
    """
    conflict = Conflict(type="性格漂移", where="第2段", suggestion="统一主角姓名为「萧寒」")
    adapter = _FakeChainAdapter(continuity_conflicts=[conflict])
    saver = MemorySaver()  # 单实例横跨 run + resume 两次端点调用
    app = _build_app(adapter, saver, e2e_sm)

    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": "链 E2E 冲突作品"})
                assert create_resp.status_code == 201
                project_id = create_resp.json()["id"]
                project_uuid = uuid.UUID(project_id)

                # 1) 发起：第 1 章四审报冲突 → interrupt → job awaiting_input。
                run_resp = await client.post(
                    f"/projects/{project_id}/chains/draft_volume/run",
                    json={"start_chapter_no": 1, "count": 1},
                )
                assert run_resp.status_code == 202
                job_id = run_resp.json()["job_id"]

                job = (await client.get(f"/jobs/{job_id}")).json()
                assert job["status"] == "awaiting_input", job
                assert job["result"]["awaiting_chapter"] == 1
                assert job["result"]["completed"] is False
                assert job["result"]["written"] == []

                # awaiting 时第 1 章仍 draft（未经 accept 晋升），无 digest。
                async with e2e_sm() as mid:
                    mid_chapters = (
                        (
                            await mid.execute(
                                select(Chapter).where(Chapter.project_id == project_uuid)
                            )
                        )
                        .scalars()
                        .all()
                    )
                    assert all(c.status == "draft" for c in mid_chapters)
                    assert mid_chapters != []
                    mid_digests = (
                        (
                            await mid.execute(
                                select(ChapterDigest).where(
                                    ChapterDigest.project_id == project_uuid
                                )
                            )
                        )
                        .scalars()
                        .all()
                    )
                    assert mid_digests == []

                # 2) resume：作者裁决冲突 → 续跑 accept → done。
                resume_resp = await client.post(
                    f"/projects/{project_id}/chains/runs/{job_id}/resume",
                    json={
                        "decisions": [
                            {"conflict_index": 0, "verdict": "ignore", "note": "笔误，忽略"}
                        ]
                    },
                )
                assert resume_resp.status_code == 202

                job = (await client.get(f"/jobs/{job_id}")).json()
                assert job["status"] == "done", job
                assert job["result"]["written"] == [1]
                assert job["result"]["completed"] is True
                assert job["result"]["awaiting_chapter"] is None

        # DB 真源断言：resume 后第 1 章 accepted + digest 一行。
        assert project_uuid is not None
        async with e2e_sm() as verify:
            chapters = (
                (await verify.execute(select(Chapter).where(Chapter.project_id == project_uuid)))
                .scalars()
                .all()
            )
            accepted = [c for c in chapters if c.status == "accepted"]
            assert {c.chapter_no for c in accepted} == {1}

            digests = (
                (
                    await verify.execute(
                        select(ChapterDigest).where(ChapterDigest.project_id == project_uuid)
                    )
                )
                .scalars()
                .all()
            )
            assert [d.chapter_no for d in digests] == [1]

            # 负向：job 状态/result 不含正文/裁决正文外的 token/prompt。
            job_row = (
                await verify.execute(select(Job).where(Job.project_id == project_uuid))
            ).scalar_one()
            assert _CHAPTER_TEXT not in str(job_row.result)
            assert job_row.status == "done"
    finally:
        if project_uuid is not None:
            await _cleanup(e2e_sm, project_uuid)