feat: M2 — 写→审(一致性)→裁决→验收(事务)；未决冲突禁验收

- 续审 Agent 声明(AgentSpec) + 结构化输出契约(ContinuityReview/Conflict 五类) - LangGraph 并行审子图(可扩四审) + collect 落 chapter_reviews 留痕 + review SSE(section/conflict) - 验收-side Repository：章节 accepted 版本晋升 + digest append-only + 审稿留痕/裁决 - API：review(SSE) + reviews 历史 + accept(单原子事务：晋升 version + 终稿 digest + 裁决留痕) - 冲突 gate：未决裁决拦截(CONFLICT_UNRESOLVED)；digest 从终稿提炼(不变量#4) - 前端：审稿报告页 + 冲突就地标注 + 裁决(采纳/忽略/手改) + 未决禁验收 + 「本次将更新」清单 - M2 E2E：真实 DB + 多档位 mock 网关零 token 走通写→审→裁决→验收→摘要入库 - 多 agent 协同台账(PROGRESS.md) + 共享记忆(memory/contracts·decisions·gotchas)
2026-06-18 11:38:28 +02:00
parent b523b4fd21
commit 68f194a043
36 changed files with 3881 additions and 0 deletions
--- a/tests/test_m2_e2e.py
+++ b/tests/test_m2_e2e.py
@@ -0,0 +1,473 @@
+"""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)