5fb7bfb1de
- 伏笔账本：纯函数状态机(OPEN/PARTIAL/CLOSED/OVERDUE) + ForeshadowLedger repo；验收后到期扫描(BackgroundTask 自建 session 置 OVERDUE)；登记/状态变更端点 - 节奏 + 三审齐：foreshadow-analyst + pace-checker 并入 LangGraph 并行审(REVIEW_SPECS)，collect 分列落 chapter_reviews(conflicts/foreshadow_sug/pace)，review SSE 加 foreshadow/pace 事件 - 大纲：outliner Agent 产 OutlineResult(含 foreshadow_windows)，POST /outline 逐章 upsert outline 表；GET /foreshadow?status= 看板 - 前端：伏笔四泳道看板(OVERDUE 琥珀) + 大纲编辑器(窗口徽标) + 节奏节拍图(▁▃▅) + 审稿页消费 foreshadow/pace SSE - bugfix(T3.8)：并行三审共用请求 session 记账触发 'Session is already flushing' → foreshadow/pace 静默丢失；SqlAlchemyLedgerSink.record 改 add-only(靠端点/事务 commit)，加并发回归测试 - M3 E2E：真实 DB + mock 网关零 token 走通 埋设→进展→验收后扫描 OVERDUE→看板 + 大纲含窗口 + 三审齐 SSE/留痕；E2E 暴露并钉住上述 bug - 门禁绿：mypy 111 / pytest 228(0 xfailed) / alembic 无漂移；前端 gen:api/lint/tsc/vitest 69/build
436 lines
15 KiB
Python
436 lines
15 KiB
Python
"""T2.2 审稿子图单测——审稿节点 + collect 留痕 + review SSE 归一（ARCH §5.2/§7.3）。
|
||
|
||
全部注入 mock 网关 / 内存 fake repo，无真 LLM、无真 Postgres（图用 MemorySaver）。
|
||
asyncio_mode=auto。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from typing import Any
|
||
|
||
from fakes_orchestrator import (
|
||
FailingRunGateway,
|
||
FakeReviewRepo,
|
||
FakeRunGateway,
|
||
SchemaRoutingRunGateway,
|
||
)
|
||
from langchain_core.runnables import RunnableConfig
|
||
from langgraph.checkpoint.memory import MemorySaver
|
||
from ww_agents import (
|
||
Conflict,
|
||
ContinuityReview,
|
||
ForeshadowReview,
|
||
ForeshadowSuggestion,
|
||
PaceIssue,
|
||
PaceReview,
|
||
continuity_spec,
|
||
foreshadow_spec,
|
||
pace_spec,
|
||
)
|
||
from ww_core.orchestrator import (
|
||
CONTINUITY,
|
||
EVENT_CONFLICT,
|
||
EVENT_DONE,
|
||
EVENT_ERROR,
|
||
EVENT_FORESHADOW,
|
||
EVENT_PACE,
|
||
EVENT_SECTION,
|
||
FORESHADOW,
|
||
PACE,
|
||
REVIEW_INCOMPLETE,
|
||
REVIEW_OK,
|
||
SECTION_DONE,
|
||
SECTION_INCOMPLETE,
|
||
ChapterState,
|
||
build_review_context,
|
||
build_review_graph,
|
||
build_review_request,
|
||
collect_reviews,
|
||
extract_conflicts,
|
||
extract_foreshadow_sug,
|
||
extract_pace,
|
||
make_review_node,
|
||
normalize_review,
|
||
run_review,
|
||
)
|
||
from ww_shared import AppError, ErrorCode
|
||
|
||
PROJECT = uuid.UUID("00000000-0000-0000-0000-000000000001")
|
||
USER = uuid.UUID("00000000-0000-0000-0000-0000000000aa")
|
||
|
||
STABLE = "## 世界观硬规则\n灵气可凝丹"
|
||
VOLATILE = "## 近况\n主角已破境"
|
||
DRAFT = "他一拳轰碎山岳。"
|
||
|
||
_CONFLICT = Conflict(
|
||
type="能力不符",
|
||
where="第3段「一拳轰碎山岳」",
|
||
refs=["人物卡:主角:力量上限=裂石"],
|
||
suggestion="改为震裂巨石",
|
||
)
|
||
|
||
|
||
def _state(reviews: dict[str, Any] | None = None) -> ChapterState:
|
||
state: ChapterState = {
|
||
"project_id": PROJECT,
|
||
"chapter_no": 7,
|
||
"user_id": USER,
|
||
"stable_core": STABLE,
|
||
"volatile": VOLATILE,
|
||
"draft": DRAFT,
|
||
"review_context": build_review_context(draft=DRAFT, stable_core=STABLE, volatile=VOLATILE),
|
||
}
|
||
if reviews is not None:
|
||
state["reviews"] = reviews
|
||
return state
|
||
|
||
|
||
# ---- build_review_context：确定性、含草稿+材料 ----
|
||
|
||
|
||
def test_build_review_context_is_deterministic_and_contains_parts() -> None:
|
||
ctx1 = build_review_context(draft=DRAFT, stable_core=STABLE, volatile=VOLATILE)
|
||
ctx2 = build_review_context(draft=DRAFT, stable_core=STABLE, volatile=VOLATILE)
|
||
|
||
assert ctx1 == ctx2 # 无时间戳/随机 → 同输入同输出
|
||
assert DRAFT in ctx1
|
||
assert STABLE in ctx1
|
||
assert VOLATILE in ctx1
|
||
|
||
|
||
# ---- build_review_request：tier 不变量、缓存断点、output_schema、run 非 stream ----
|
||
|
||
|
||
def test_build_review_request_uses_spec_tier_and_cached_prompt() -> None:
|
||
req = build_review_request(continuity_spec, _state())
|
||
|
||
assert req.tier == "analyst" # 不变量②：spec 档位透传，不传 model
|
||
assert req.stream is False # 续审用 run() 非 stream()
|
||
assert len(req.system) == 1
|
||
assert req.system[0].text == continuity_spec.system_prompt
|
||
assert req.system[0].cache is True # 不变量#9
|
||
assert req.output_schema is ContinuityReview
|
||
assert req.scope.user_id == USER
|
||
assert req.scope.project_id == PROJECT
|
||
|
||
|
||
# ---- run_review：结构化结果回 state、只读不写库 ----
|
||
|
||
|
||
async def test_run_review_returns_ok_with_parsed_result() -> None:
|
||
gateway = FakeRunGateway(ContinuityReview(conflicts=[_CONFLICT]))
|
||
|
||
result = await run_review(continuity_spec, _state(), gateway=gateway)
|
||
|
||
assert gateway.call_count == 1
|
||
entry = result["reviews"][CONTINUITY]
|
||
assert entry["status"] == REVIEW_OK
|
||
assert entry["result"]["conflicts"][0]["type"] == "能力不符"
|
||
|
||
|
||
async def test_run_review_isolates_gateway_failure_as_incomplete() -> None:
|
||
gateway = FailingRunGateway(AppError(ErrorCode.LLM_UNAVAILABLE, "boom"))
|
||
|
||
result = await run_review(continuity_spec, _state(), gateway=gateway)
|
||
|
||
entry = result["reviews"][CONTINUITY]
|
||
assert entry["status"] == REVIEW_INCOMPLETE # §5.2：失败隔离、不上抛
|
||
assert entry["result"] is None
|
||
|
||
|
||
async def test_make_review_node_binds_gateway() -> None:
|
||
gateway = FakeRunGateway(ContinuityReview(conflicts=[]))
|
||
node = make_review_node(continuity_spec, gateway)
|
||
|
||
result = await node(_state())
|
||
|
||
assert result["reviews"][CONTINUITY]["status"] == REVIEW_OK
|
||
|
||
|
||
# ---- collect：抽冲突 + 落 chapter_reviews 留痕（只 flush，不在节点 commit） ----
|
||
|
||
|
||
def test_extract_conflicts_only_from_ok_continuity() -> None:
|
||
ok = {CONTINUITY: {"status": REVIEW_OK, "result": {"conflicts": [_CONFLICT.model_dump()]}}}
|
||
incomplete = {CONTINUITY: {"status": REVIEW_INCOMPLETE, "result": None}}
|
||
|
||
assert len(extract_conflicts(ok)) == 1
|
||
assert extract_conflicts(incomplete) == []
|
||
assert extract_conflicts({}) == []
|
||
|
||
|
||
async def test_collect_records_conflicts_to_review_repo() -> None:
|
||
repo = FakeReviewRepo()
|
||
reviews = {CONTINUITY: {"status": REVIEW_OK, "result": {"conflicts": [_CONFLICT.model_dump()]}}}
|
||
|
||
out = await collect_reviews(_state(reviews), review_repo=repo)
|
||
|
||
assert out == {} # 不作真相源：留痕在表（不变量#5）
|
||
assert len(repo.records) == 1
|
||
rec = repo.records[0]
|
||
assert rec["project_id"] == PROJECT
|
||
assert rec["chapter_no"] == 7
|
||
assert rec["chapter_version"] is None # version 绑定/裁决归 T2.4
|
||
assert rec["conflicts"][0]["type"] == "能力不符"
|
||
|
||
|
||
async def test_collect_records_empty_when_review_incomplete() -> None:
|
||
repo = FakeReviewRepo()
|
||
reviews = {CONTINUITY: {"status": REVIEW_INCOMPLETE, "result": None}}
|
||
|
||
await collect_reviews(_state(reviews), review_repo=repo)
|
||
|
||
assert repo.records[0]["conflicts"] == [] # 未完成 → 不臆造冲突
|
||
|
||
|
||
# ---- 图：START→continuity→collect→END，全程 mock，MemorySaver ----
|
||
|
||
|
||
def test_build_review_graph_compiles_with_nodes() -> None:
|
||
graph = build_review_graph(
|
||
FakeRunGateway(ContinuityReview(conflicts=[])),
|
||
FakeReviewRepo(),
|
||
checkpointer=MemorySaver(),
|
||
)
|
||
|
||
nodes = graph.get_graph().nodes
|
||
assert CONTINUITY in nodes
|
||
assert "collect" in nodes
|
||
assert graph.checkpointer is not None
|
||
|
||
|
||
async def test_review_graph_runs_continuity_then_collect_end_to_end() -> None:
|
||
gateway = FakeRunGateway(ContinuityReview(conflicts=[_CONFLICT]))
|
||
repo = FakeReviewRepo()
|
||
graph = build_review_graph(gateway, repo, checkpointer=MemorySaver())
|
||
config: RunnableConfig = {"configurable": {"thread_id": "rev-1"}}
|
||
|
||
final = await graph.ainvoke(_state(), config=config)
|
||
|
||
assert final["reviews"][CONTINUITY]["status"] == REVIEW_OK
|
||
assert len(repo.records) == 1 # collect 落了留痕
|
||
assert repo.records[0]["conflicts"][0]["type"] == "能力不符"
|
||
|
||
|
||
async def test_review_graph_does_not_block_on_review_failure() -> None:
|
||
gateway = FailingRunGateway(RuntimeError("down"))
|
||
repo = FakeReviewRepo()
|
||
graph = build_review_graph(gateway, repo, checkpointer=MemorySaver())
|
||
config: RunnableConfig = {"configurable": {"thread_id": "rev-2"}}
|
||
|
||
final = await graph.ainvoke(_state(), config=config)
|
||
|
||
# 审失败被隔离为 incomplete，图照常走到 collect 并留痕（空冲突）
|
||
assert final["reviews"][CONTINUITY]["status"] == REVIEW_INCOMPLETE
|
||
assert repo.records[0]["conflicts"] == []
|
||
|
||
|
||
# ---- review SSE 归一：section + conflict + done；incomplete 路径；异常→error ----
|
||
|
||
|
||
async def test_normalize_review_emits_section_conflict_done() -> None:
|
||
reviews = {CONTINUITY: {"status": REVIEW_OK, "result": {"conflicts": [_CONFLICT.model_dump()]}}}
|
||
|
||
events = [e async for e in normalize_review(reviews)]
|
||
|
||
assert [e.event for e in events] == [EVENT_SECTION, EVENT_CONFLICT, EVENT_DONE]
|
||
assert events[0].data == {"name": CONTINUITY, "status": SECTION_DONE}
|
||
assert events[1].data["type"] == "能力不符"
|
||
assert events[1].data["refs"] == ["人物卡:主角:力量上限=裂石"]
|
||
assert events[-1].data == {"length": 1}
|
||
|
||
|
||
async def test_normalize_review_marks_incomplete_section_without_conflicts() -> None:
|
||
reviews = {CONTINUITY: {"status": REVIEW_INCOMPLETE, "result": None}}
|
||
|
||
events = [e async for e in normalize_review(reviews)]
|
||
|
||
assert [e.event for e in events] == [EVENT_SECTION, EVENT_DONE]
|
||
assert events[0].data == {"name": CONTINUITY, "status": SECTION_INCOMPLETE}
|
||
|
||
|
||
async def test_normalize_review_maps_unexpected_error_to_error_event() -> None:
|
||
# 畸形 entry（result 非 dict）触发归一内部异常 → error 事件，不上抛
|
||
reviews = {CONTINUITY: {"status": REVIEW_OK, "result": 123}}
|
||
|
||
events = [e async for e in normalize_review(reviews, request_id="req-9")]
|
||
|
||
assert events[-1].event == EVENT_ERROR
|
||
assert events[-1].data["code"] == ErrorCode.INTERNAL
|
||
assert events[-1].data["request_id"] == "req-9"
|
||
|
||
|
||
# ============================================================================
|
||
# T3.3 三审齐：foreshadow + pace 并入并行审 / collect 列映射 / SSE surface
|
||
# ============================================================================
|
||
|
||
_FORESHADOW = ForeshadowReview(
|
||
planted=[
|
||
ForeshadowSuggestion(code="FS-MARK", title="胸口胎记", where="第2段", note="皇族暗示")
|
||
],
|
||
resolved=[ForeshadowSuggestion(code="FS-BIRTH", title="身世", where="第9段", note="回收")],
|
||
)
|
||
_PACE = PaceReview(
|
||
water=[PaceIssue(where="第4-6段", reason="信息密度低")],
|
||
hook=True,
|
||
beat_map=[1, 3, 5, 0, 4],
|
||
)
|
||
|
||
|
||
def _three_reviews() -> dict[str, Any]:
|
||
"""三审皆 ok 的 state['reviews']（continuity/foreshadow/pace）。"""
|
||
return {
|
||
CONTINUITY: {"status": REVIEW_OK, "result": {"conflicts": [_CONFLICT.model_dump()]}},
|
||
FORESHADOW: {"status": REVIEW_OK, "result": _FORESHADOW.model_dump()},
|
||
PACE: {"status": REVIEW_OK, "result": _PACE.model_dump()},
|
||
}
|
||
|
||
|
||
# ---- collect 三审列映射 ----
|
||
|
||
|
||
def test_extract_foreshadow_sug_flattens_planted_and_resolved() -> None:
|
||
sug = extract_foreshadow_sug(_three_reviews())
|
||
|
||
assert len(sug) == 2 # 1 planted + 1 resolved 扁平
|
||
kinds = {s["kind"] for s in sug}
|
||
assert kinds == {"planted", "resolved"}
|
||
planted = next(s for s in sug if s["kind"] == "planted")
|
||
assert planted["code"] == "FS-MARK"
|
||
|
||
|
||
def test_extract_pace_returns_dict() -> None:
|
||
pace = extract_pace(_three_reviews())
|
||
|
||
assert pace is not None
|
||
assert pace["hook"] is True
|
||
assert pace["beat_map"] == [1, 3, 5, 0, 4]
|
||
|
||
|
||
def test_extract_foreshadow_and_pace_empty_when_incomplete() -> None:
|
||
reviews = {
|
||
FORESHADOW: {"status": REVIEW_INCOMPLETE, "result": None},
|
||
PACE: {"status": REVIEW_INCOMPLETE, "result": None},
|
||
}
|
||
assert extract_foreshadow_sug(reviews) == []
|
||
assert extract_pace(reviews) is None
|
||
|
||
|
||
async def test_collect_records_three_audits_to_their_columns() -> None:
|
||
repo = FakeReviewRepo()
|
||
|
||
await collect_reviews(_state(_three_reviews()), review_repo=repo)
|
||
|
||
assert len(repo.records) == 1 # 一次 record 落齐三审
|
||
rec = repo.records[0]
|
||
assert rec["conflicts"][0]["type"] == "能力不符" # continuity→conflicts
|
||
assert len(rec["foreshadow_sug"]) == 2 # foreshadow→foreshadow_sug（扁平）
|
||
assert rec["pace"]["hook"] is True # pace→pace
|
||
|
||
|
||
# ---- 三审并行图跑通（mock 网关按 schema 路由产三 parsed）----
|
||
|
||
|
||
async def test_three_review_graph_runs_all_audits_end_to_end() -> None:
|
||
gateway = SchemaRoutingRunGateway(
|
||
{
|
||
ContinuityReview: ContinuityReview(conflicts=[_CONFLICT]),
|
||
ForeshadowReview: _FORESHADOW,
|
||
PaceReview: _PACE,
|
||
}
|
||
)
|
||
repo = FakeReviewRepo()
|
||
graph = build_review_graph(
|
||
gateway,
|
||
repo,
|
||
review_specs=(continuity_spec, foreshadow_spec, pace_spec),
|
||
checkpointer=MemorySaver(),
|
||
)
|
||
config: RunnableConfig = {"configurable": {"thread_id": "rev-3"}}
|
||
|
||
final = await graph.ainvoke(_state(), config=config)
|
||
|
||
assert final["reviews"][CONTINUITY]["status"] == REVIEW_OK
|
||
assert final["reviews"][FORESHADOW]["status"] == REVIEW_OK
|
||
assert final["reviews"][PACE]["status"] == REVIEW_OK
|
||
assert len(gateway.calls) == 3 # 三审各调一次网关
|
||
# collect 一次 record 落齐三列
|
||
rec = repo.records[0]
|
||
assert rec["conflicts"] and rec["foreshadow_sug"] and rec["pace"]
|
||
|
||
|
||
async def test_three_review_graph_isolates_one_failing_audit() -> None:
|
||
# foreshadow schema 缺失 → SchemaRoutingRunGateway 抛 KeyError → run_review 隔离为 incomplete
|
||
gateway = SchemaRoutingRunGateway(
|
||
{
|
||
ContinuityReview: ContinuityReview(conflicts=[_CONFLICT]),
|
||
PaceReview: _PACE,
|
||
}
|
||
)
|
||
repo = FakeReviewRepo()
|
||
graph = build_review_graph(
|
||
gateway,
|
||
repo,
|
||
review_specs=(continuity_spec, foreshadow_spec, pace_spec),
|
||
checkpointer=MemorySaver(),
|
||
)
|
||
config: RunnableConfig = {"configurable": {"thread_id": "rev-4"}}
|
||
|
||
final = await graph.ainvoke(_state(), config=config)
|
||
|
||
# foreshadow 失败被隔离，其余两审照常 + collect 照常留痕（foreshadow_sug 空）
|
||
assert final["reviews"][FORESHADOW]["status"] == REVIEW_INCOMPLETE
|
||
assert final["reviews"][CONTINUITY]["status"] == REVIEW_OK
|
||
assert final["reviews"][PACE]["status"] == REVIEW_OK
|
||
rec = repo.records[0]
|
||
assert rec["conflicts"] # continuity 落库
|
||
assert rec["foreshadow_sug"] == [] # 失败列留空，不阻塞其余
|
||
assert rec["pace"]["hook"] is True # pace 落库
|
||
|
||
|
||
# ---- SSE：normalize_review surface 三审（section + conflict + foreshadow + pace + done）----
|
||
|
||
|
||
async def test_normalize_review_surfaces_three_audits() -> None:
|
||
events = [e async for e in normalize_review(_three_reviews())]
|
||
kinds = [e.event for e in events]
|
||
|
||
# 字典序遍历：continuity < foreshadow < pace
|
||
# continuity: section + 1 conflict
|
||
# foreshadow: section + 2（planted+resolved）；pace: section + 1
|
||
assert kinds == [
|
||
EVENT_SECTION,
|
||
EVENT_CONFLICT,
|
||
EVENT_SECTION,
|
||
EVENT_FORESHADOW,
|
||
EVENT_FORESHADOW,
|
||
EVENT_SECTION,
|
||
EVENT_PACE,
|
||
EVENT_DONE,
|
||
]
|
||
# foreshadow 事件带 kind + code
|
||
fs_events = [e for e in events if e.event == EVENT_FORESHADOW]
|
||
assert {e.data["kind"] for e in fs_events} == {"planted", "resolved"}
|
||
# pace 事件带节拍图
|
||
pace_evt = next(e for e in events if e.event == EVENT_PACE)
|
||
assert pace_evt.data["beat_map"] == [1, 3, 5, 0, 4]
|
||
assert pace_evt.data["hook"] is True
|
||
assert events[-1].data == {"length": 3} # 三审项
|
||
|
||
|
||
async def test_normalize_review_skips_results_for_incomplete_audit() -> None:
|
||
reviews = {
|
||
CONTINUITY: {"status": REVIEW_OK, "result": {"conflicts": []}},
|
||
FORESHADOW: {"status": REVIEW_INCOMPLETE, "result": None},
|
||
PACE: {"status": REVIEW_OK, "result": _PACE.model_dump()},
|
||
}
|
||
|
||
events = [e async for e in normalize_review(reviews)]
|
||
kinds = [e.event for e in events]
|
||
|
||
# foreshadow incomplete → 仅 section，无 foreshadow 事件
|
||
assert kinds == [EVENT_SECTION, EVENT_SECTION, EVENT_SECTION, EVENT_PACE, EVENT_DONE]
|
||
fs_section = next(e for e in events if e.data.get("name") == FORESHADOW)
|
||
assert fs_section.data["status"] == SECTION_INCOMPLETE
|