"""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 assert fp["dimensions"]["句长节奏"] == "短句为主" assert fp["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)