453 lines
20 KiB
Python
453 lines
20 KiB
Python
"""WFW-9 端到端：两阶段 AI 反问澄清预检（真实 DB，零 token，MOCK 网关）。
|
||
|
||
证明 WFW-9 路线A 两阶段之「问题」阶段的两条预检端点闭环（refine 侧 M1 + rewrite 侧 M2），
|
||
锁定其跨栈契约：HTTP → dep 解析 → 真实 `Gateway` → 结构化 `ClarifyDecision` →
|
||
`usage_ledger` 落真 pg → JSON 响应。两端点共用 `get_clarify_gateway`（analyst 档，默认路由
|
||
deepseek），均**非流式、只读**（绝不写业务表，只有末尾 `session.commit()` 让网关 ledger 落库）
|
||
且**失败开放**（网关产非 `ClarifyDecision` → `run_clarify` 确定性回退 need_clarification=false，
|
||
绝不 500）。
|
||
|
||
范式与 `tests/test_m2_e2e.py` / `tests/test_t6_toolbox_e2e.py` 一致：**在 provider/adapter 层
|
||
mock**（真实 `Gateway` + 据调用返回 `parsed` 的假适配器，绝不联网）+ 真实
|
||
`SqlAlchemyLedgerSink`（用量记账真落 pg）+ DB 真源断言。**不**在 dep 层塞假网关——那是端点单测
|
||
（`apps/api/tests/`）的层级；E2E 走真网关证明记账/路由/结构化透传全链路。无 pg → skip。
|
||
|
||
用例（真 pg + mock 网关 → 脚本化 ClarifyDecision，零 token）：
|
||
1. refine 含糊意见 → 反问：mock 产 need_clarification=true + 恰 1 问（2–3 选项 + 自由输入）
|
||
→ 200，need_clarification is True，恰 1 问带选项，ledger 落 analyst 一行。
|
||
2. refine 明确意见 → 放行：mock 产 need_clarification=false + verification → questions==[]、
|
||
verification 非空。
|
||
3. rewrite 含糊意见 → 反问：POST rewrite/clarify（带 prior_draft 摘录）→ True + 1 问。
|
||
4. rewrite 明确意见 → 放行：→ need_clarification is False。
|
||
5. 失败开放：mock 让网关返回**非** ClarifyDecision（parsed=None）→ 端点仍 200 且
|
||
need_clarification is False（绝不 500），且 ledger 仍落一行（预检是真计费 LLM 调用）。
|
||
6. 只读 + 记账：clarify 后真 pg 中业务表**零变更**（seed 的 chapter 内容字节一致、无新
|
||
chapter/review/digest 行）但 usage_ledger **真落一行**；响应 JSON 绝不回灌原文摘录。
|
||
7. 404：clarify 打不存在的 project_id → 404 错误信封，且无 ledger/业务行。
|
||
|
||
坑（见 memory/gotchas，同 M2/T6）：
|
||
- `LifespanManager` 触发 lifespan → `seed_stub_user`（project/chapter owner FK 依赖它）。
|
||
- `get_sessionmaker` engine 绑定首个 loop → 每 DB 测试清缓存重建、结束 dispose。
|
||
- 网关 ledger 只 flush；端点末 `session.commit()` → usage_ledger 方落库。
|
||
- ASGITransport `raise_app_exceptions=True`：404 是 `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, func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||
from ww_agents import ClarifyDecision, ClarifyOption, ClarifyQuestion
|
||
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
|
||
|
||
# analyst 默认路由 deepseek（config.tier_defaults），同 M2/T6。
|
||
_PROVIDER = "deepseek"
|
||
|
||
# analyst 档假用量（喂记账；证明预检记账真落 pg + input_tokens 可回指本档位调用）。
|
||
_ANALYST_USAGE = ProviderUsage(input_tokens=23, output_tokens=7)
|
||
|
||
# 脚本化「反问」决策：need_clarification=true + 恰 1 问（v1 硬上限）+ 3 选项 + 自由输入兜底。
|
||
_ASK_DECISION = ClarifyDecision(
|
||
need_clarification=True,
|
||
questions=[
|
||
ClarifyQuestion(
|
||
question="你想让这段往哪个方向改？",
|
||
options=[
|
||
ClarifyOption(label="加快节奏", value="拆短句、压缩铺陈"),
|
||
ClarifyOption(label="加重冲突", value="强化人物对立、提高赌注"),
|
||
ClarifyOption(label="补足细节", value="增加环境与心理描写"),
|
||
],
|
||
allow_free_text=True,
|
||
)
|
||
],
|
||
verification=None,
|
||
)
|
||
|
||
# 脚本化「放行」决策：need_clarification=false + 一句确认语，questions 留空。
|
||
_PROCEED_DECISION = ClarifyDecision(
|
||
need_clarification=False,
|
||
verification="我的理解是把这段节奏压紧、突出主角的犹豫，对吗？",
|
||
)
|
||
|
||
# 原文标记串：塞进 prior_draft/segment，断言**响应绝不回灌**（预检只判不回抄原文）。
|
||
_MANUSCRIPT_MARKER = "秘密正文标记：主角在雨夜独自登上钟楼敲响了第七下钟"
|
||
|
||
|
||
class _FakeClarifyAdapter:
|
||
"""实现 `ProviderAdapter` Protocol：`complete()` 返回脚本化 `parsed`，绝不联网。
|
||
|
||
`parsed` 由构造注入——正常用例给 `ClarifyDecision`；失败开放用例给 `None`（模拟
|
||
instructor 重试上限后仍畸形），验证 `run_clarify` 端到端确定性回退。每路带
|
||
`ProviderUsage` → 网关据此落一条 usage_ledger（预检是真计费调用）。clarify 走 run()
|
||
非流式，`stream()` 仅为满足 Protocol、被调即判失败。
|
||
"""
|
||
|
||
provider = _PROVIDER
|
||
|
||
def __init__(self, parsed: ClarifyDecision | None) -> None:
|
||
self._parsed = parsed
|
||
|
||
def capabilities(self) -> Capabilities:
|
||
return Capabilities(structured_output=True)
|
||
|
||
async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
|
||
text = self._parsed.model_dump_json() if self._parsed is not None else "{}"
|
||
return ProviderResult(text=text, parsed=self._parsed, usage=_ANALYST_USAGE)
|
||
|
||
async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
|
||
yield StreamChunk(usage=_ANALYST_USAGE)
|
||
raise AssertionError("clarify 预检必须非流式，不得走 stream()")
|
||
|
||
|
||
@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_override_factory(
|
||
adapter: _FakeClarifyAdapter,
|
||
) -> Callable[[AsyncSession], Gateway]:
|
||
"""依赖覆盖：真实 Gateway + 给定假适配器 + 真实 ledger（请求 session）。
|
||
|
||
ledger 用**请求 session**（`Depends(get_session)`，FastAPI 按请求缓存，与端点末尾
|
||
`commit()` 同一实例）→ 由端点 `session.commit()` 落库（同 M2 记账闭环）。
|
||
"""
|
||
|
||
def _override(session: Annotated[AsyncSession, Depends(get_session)]) -> Gateway:
|
||
return Gateway(
|
||
adapters={_PROVIDER: adapter},
|
||
ledger=SqlAlchemyLedgerSink(session),
|
||
resolver=resolve_route,
|
||
)
|
||
|
||
return _override
|
||
|
||
|
||
async def _count(
|
||
sm: async_sessionmaker[AsyncSession],
|
||
model: type[Chapter] | type[ChapterReview] | type[ChapterDigest] | type[UsageLedger],
|
||
project_uuid: uuid.UUID,
|
||
) -> int:
|
||
async with sm() as s:
|
||
return int(
|
||
(
|
||
await s.execute(
|
||
select(func.count()).select_from(model).where(model.project_id == project_uuid)
|
||
)
|
||
).scalar_one()
|
||
)
|
||
|
||
|
||
async def _seed_chapter(
|
||
sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID, content: str
|
||
) -> None:
|
||
"""直插一行草稿 chapter（供只读用例断言 clarify 后内容字节一致、未被触碰）。"""
|
||
async with sm() as s:
|
||
s.add(
|
||
Chapter(
|
||
project_id=project_uuid,
|
||
volume=1,
|
||
chapter_no=1,
|
||
content=content,
|
||
status="draft",
|
||
version=1,
|
||
)
|
||
)
|
||
await s.commit()
|
||
|
||
|
||
async def _cleanup(sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
|
||
"""按 FK 顺序清理。"""
|
||
async with 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()
|
||
|
||
|
||
def _assert_no_manuscript_leak(payload: object) -> None:
|
||
"""负向：响应体序列化后绝不含原文标记（预检只判含糊，不回抄原文）。"""
|
||
blob = json.dumps(payload, ensure_ascii=False)
|
||
assert _MANUSCRIPT_MARKER not in blob, "原文标记泄漏进 clarify 响应"
|
||
|
||
|
||
def _build_app_with_adapter(adapter: _FakeClarifyAdapter): # type: ignore[no-untyped-def]
|
||
"""建 app 并把 `get_clarify_gateway` 覆盖为「真 Gateway + 假适配器」——refine/rewrite 共用。"""
|
||
from ww_api.main import create_app
|
||
from ww_api.services.project_deps import get_clarify_gateway
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_clarify_gateway] = _gateway_override_factory(adapter)
|
||
return app
|
||
|
||
|
||
async def _create_project(client: httpx.AsyncClient, title: str) -> uuid.UUID:
|
||
resp = await client.post("/projects", json={"title": title, "genre": "玄幻"})
|
||
assert resp.status_code == 201
|
||
return uuid.UUID(resp.json()["id"])
|
||
|
||
|
||
async def test_refine_vague_instruction_asks_one_clarifying_question(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 1：refine 含糊意见「改改」→ 200，need_clarification True，恰 1 问带选项 + ledger。"""
|
||
app = _build_app_with_adapter(_FakeClarifyAdapter(_ASK_DECISION))
|
||
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:
|
||
project_uuid = await _create_project(client, "WFW-9 refine 含糊作品")
|
||
|
||
resp = await client.post(
|
||
f"/projects/{project_uuid}/chapters/1/refine/clarify",
|
||
json={"segment": "他走进房间坐下。", "instruction": "改改"},
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["need_clarification"] is True
|
||
assert len(body["questions"]) == 1 # v1 硬上限 1 问
|
||
question = body["questions"][0]
|
||
assert question["question"]
|
||
assert 2 <= len(question["options"]) <= 3
|
||
assert question["allow_free_text"] is True
|
||
_assert_no_manuscript_leak(body)
|
||
|
||
# DB 真源：预检是真计费 analyst 调用 → ledger 落 1 行；业务表零新增。
|
||
assert project_uuid is not None
|
||
assert await _count(e2e_sm, UsageLedger, project_uuid) == 1
|
||
assert await _count(e2e_sm, Chapter, project_uuid) == 0
|
||
async with e2e_sm() as verify:
|
||
row = (
|
||
await verify.execute(
|
||
select(UsageLedger).where(UsageLedger.project_id == project_uuid)
|
||
)
|
||
).scalar_one()
|
||
assert row.provider == _PROVIDER
|
||
assert row.input_tokens == _ANALYST_USAGE.input_tokens
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|
||
|
||
|
||
async def test_refine_clear_instruction_proceeds_without_questions(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 2：refine 明确意见 → need_clarification False，questions==[]，verification 非空。"""
|
||
app = _build_app_with_adapter(_FakeClarifyAdapter(_PROCEED_DECISION))
|
||
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:
|
||
project_uuid = await _create_project(client, "WFW-9 refine 明确作品")
|
||
|
||
resp = await client.post(
|
||
f"/projects/{project_uuid}/chapters/1/refine/clarify",
|
||
json={
|
||
"segment": "他走进房间坐下。",
|
||
"instruction": "把这句拆成两个短句，突出他的迟疑",
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["need_clarification"] is False
|
||
assert body["questions"] == []
|
||
assert body["verification"] # 明确时给一句确认语
|
||
|
||
# 明确放行同样是真计费调用 → ledger 落 1 行。
|
||
assert project_uuid is not None
|
||
assert await _count(e2e_sm, UsageLedger, project_uuid) == 1
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|
||
|
||
|
||
async def test_rewrite_vague_feedback_asks_one_clarifying_question(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 3：rewrite 含糊意见「节奏太慢」（带 prior_draft）→ True + 恰 1 问。"""
|
||
app = _build_app_with_adapter(_FakeClarifyAdapter(_ASK_DECISION))
|
||
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:
|
||
project_uuid = await _create_project(client, "WFW-9 rewrite 含糊作品")
|
||
|
||
resp = await client.post(
|
||
f"/projects/{project_uuid}/chapters/1/rewrite/clarify",
|
||
json={
|
||
"feedback": "节奏太慢",
|
||
"prior_draft": "第一章：清晨的雾还没散。" + _MANUSCRIPT_MARKER,
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["need_clarification"] is True
|
||
assert len(body["questions"]) == 1
|
||
assert body["questions"][0]["options"]
|
||
_assert_no_manuscript_leak(body)
|
||
|
||
assert project_uuid is not None
|
||
assert await _count(e2e_sm, UsageLedger, project_uuid) == 1
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|
||
|
||
|
||
async def test_rewrite_clear_feedback_proceeds_without_questions(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 4：rewrite 明确意见 → need_clarification False。"""
|
||
app = _build_app_with_adapter(_FakeClarifyAdapter(_PROCEED_DECISION))
|
||
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:
|
||
project_uuid = await _create_project(client, "WFW-9 rewrite 明确作品")
|
||
|
||
resp = await client.post(
|
||
f"/projects/{project_uuid}/chapters/1/rewrite/clarify",
|
||
json={
|
||
"feedback": "把第 3 段主角的犹豫删掉，直接进入冲突，全章提速",
|
||
"prior_draft": "第一章：清晨的雾还没散。",
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["need_clarification"] is False
|
||
assert body["questions"] == []
|
||
assert body["verification"]
|
||
|
||
assert project_uuid is not None
|
||
assert await _count(e2e_sm, UsageLedger, project_uuid) == 1
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|
||
|
||
|
||
async def test_clarify_fails_open_when_gateway_returns_non_decision(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 5：网关返回非 ClarifyDecision（parsed=None）→ 端点仍 200、need_clarification False。
|
||
|
||
证明 `run_clarify` 的确定性回退端到端生效（不 500、不阻塞润色），且该次调用仍真计费落库。
|
||
"""
|
||
app = _build_app_with_adapter(_FakeClarifyAdapter(None)) # 畸形：parsed=None
|
||
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:
|
||
project_uuid = await _create_project(client, "WFW-9 失败开放作品")
|
||
|
||
resp = await client.post(
|
||
f"/projects/{project_uuid}/chapters/1/refine/clarify",
|
||
json={"segment": "他走进房间坐下。", "instruction": "随便改"},
|
||
)
|
||
# 绝不 500——确定性回退为「不问、放行」。
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["need_clarification"] is False
|
||
assert body["questions"] == []
|
||
|
||
# 回退不改「这是一次真 LLM 调用」的事实 → ledger 仍落一行。
|
||
assert project_uuid is not None
|
||
assert await _count(e2e_sm, UsageLedger, project_uuid) == 1
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|
||
|
||
|
||
async def test_clarify_is_read_only_and_records_ledger(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 6：clarify 后业务表零变更（seed 的 chapter 字节一致）但 usage_ledger 真落一行。"""
|
||
app = _build_app_with_adapter(_FakeClarifyAdapter(_ASK_DECISION))
|
||
transport = httpx.ASGITransport(app=app)
|
||
project_uuid: uuid.UUID | None = None
|
||
seeded_content = "第一章：清晨的雾还没散。" + _MANUSCRIPT_MARKER
|
||
try:
|
||
async with LifespanManager(app):
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
project_uuid = await _create_project(client, "WFW-9 只读记账作品")
|
||
await _seed_chapter(e2e_sm, project_uuid, seeded_content)
|
||
|
||
resp = await client.post(
|
||
f"/projects/{project_uuid}/chapters/1/rewrite/clarify",
|
||
json={"feedback": "改改", "prior_draft": seeded_content},
|
||
)
|
||
assert resp.status_code == 200
|
||
_assert_no_manuscript_leak(resp.json())
|
||
|
||
assert project_uuid is not None
|
||
# 业务表零变更：seed 的 chapter 仍是唯一一行且内容字节一致；无 review/digest 落库。
|
||
assert await _count(e2e_sm, Chapter, project_uuid) == 1
|
||
assert await _count(e2e_sm, ChapterReview, project_uuid) == 0
|
||
assert await _count(e2e_sm, ChapterDigest, project_uuid) == 0
|
||
async with e2e_sm() as verify:
|
||
chapter = (
|
||
await verify.execute(select(Chapter).where(Chapter.project_id == project_uuid))
|
||
).scalar_one()
|
||
assert chapter.content == seeded_content
|
||
assert chapter.status == "draft"
|
||
assert chapter.version == 1
|
||
# 但预检本身是真计费 analyst 调用 → usage_ledger 恰一行。
|
||
assert await _count(e2e_sm, UsageLedger, project_uuid) == 1
|
||
finally:
|
||
if project_uuid is not None:
|
||
await _cleanup(e2e_sm, project_uuid)
|
||
|
||
|
||
async def test_clarify_nonexistent_project_returns_404(
|
||
e2e_sm: async_sessionmaker[AsyncSession],
|
||
) -> None:
|
||
"""用例 7：clarify 打不存在的 project_id → 404 错误信封，且无 ledger/业务行。"""
|
||
app = _build_app_with_adapter(_FakeClarifyAdapter(_ASK_DECISION))
|
||
transport = httpx.ASGITransport(app=app)
|
||
missing = uuid.uuid4()
|
||
async with LifespanManager(app):
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
resp = await client.post(
|
||
f"/projects/{missing}/chapters/1/refine/clarify",
|
||
json={"segment": "他走进房间坐下。", "instruction": "改改"},
|
||
)
|
||
assert resp.status_code == 404
|
||
err = resp.json()["error"]
|
||
assert err["code"] == "NOT_FOUND"
|
||
|
||
# 项目不存在触网关前 fail-fast → ledger/业务表均无该 project 的行。
|
||
assert await _count(e2e_sm, UsageLedger, missing) == 0
|
||
assert await _count(e2e_sm, Chapter, missing) == 0
|