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
168 lines
4.4 KiB
Python
168 lines
4.4 KiB
Python
"""T3.3 三审契约测试：foreshadow / pace spec + 输出 schema（C6 扩）。
|
||
|
||
契约测试——构造符合 schema 的 mock 响应，校验字段、tier、四审只读（writes==[]）。
|
||
不联网、无 DB。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
from ww_agents import (
|
||
AgentSpec,
|
||
ForeshadowReview,
|
||
ForeshadowSuggestion,
|
||
PaceIssue,
|
||
PaceReview,
|
||
foreshadow_spec,
|
||
pace_spec,
|
||
)
|
||
|
||
# ---- ForeshadowReview schema ----
|
||
|
||
|
||
def test_foreshadow_review_parses_mock_response() -> None:
|
||
# Arrange：模拟网关 instructor 校验后的结构化产出
|
||
mock = {
|
||
"planted": [
|
||
{
|
||
"code": "FS-MARK",
|
||
"title": "胸口的胎记",
|
||
"where": "第 2 段主角沐浴时被瞥见",
|
||
"note": "暗示皇族血统",
|
||
}
|
||
],
|
||
"resolved": [
|
||
{
|
||
"code": "FS-BIRTH",
|
||
"title": "身世之谜",
|
||
"where": "第 9 段老者道破来历",
|
||
"note": "回收第 1 章埋设",
|
||
}
|
||
],
|
||
}
|
||
|
||
# Act
|
||
review = ForeshadowReview.model_validate(mock)
|
||
|
||
# Assert
|
||
assert len(review.planted) == 1
|
||
assert review.planted[0].code == "FS-MARK"
|
||
assert review.resolved[0].code == "FS-BIRTH"
|
||
assert review.resolved[0].where == "第 9 段老者道破来历"
|
||
|
||
|
||
def test_foreshadow_review_defaults_to_empty_lists() -> None:
|
||
review = ForeshadowReview()
|
||
assert review.planted == []
|
||
assert review.resolved == []
|
||
|
||
|
||
def test_foreshadow_suggestion_only_title_required() -> None:
|
||
sug = ForeshadowSuggestion.model_validate({"title": "新埋的悬念"})
|
||
assert sug.title == "新埋的悬念"
|
||
assert sug.code is None
|
||
assert sug.where is None
|
||
assert sug.note is None
|
||
|
||
|
||
def test_foreshadow_suggestion_requires_title() -> None:
|
||
with pytest.raises(ValidationError):
|
||
ForeshadowSuggestion.model_validate({"code": "FS-X"})
|
||
|
||
|
||
# ---- PaceReview schema ----
|
||
|
||
|
||
def test_pace_review_parses_mock_response() -> None:
|
||
mock = {
|
||
"water": [{"where": "第 4–6 段反复描写天气", "reason": "信息密度低、与主线无关"}],
|
||
"hook": True,
|
||
"beat_map": [1, 3, 2, 0, 4, 5],
|
||
}
|
||
|
||
review = PaceReview.model_validate(mock)
|
||
|
||
assert len(review.water) == 1
|
||
assert review.water[0].reason == "信息密度低、与主线无关"
|
||
assert review.hook is True
|
||
assert review.beat_map == [1, 3, 2, 0, 4, 5]
|
||
|
||
|
||
def test_pace_review_defaults() -> None:
|
||
review = PaceReview()
|
||
assert review.water == []
|
||
assert review.hook is False
|
||
assert review.beat_map == []
|
||
|
||
|
||
def test_pace_issue_requires_where_and_reason() -> None:
|
||
with pytest.raises(ValidationError):
|
||
PaceIssue.model_validate({"where": "x"})
|
||
|
||
|
||
# ---- foreshadow_spec 声明 ----
|
||
|
||
|
||
def test_foreshadow_spec_is_analyst_tier() -> None:
|
||
assert foreshadow_spec.tier == "analyst"
|
||
assert foreshadow_spec.name == "foreshadow"
|
||
|
||
|
||
def test_foreshadow_spec_is_read_only() -> None:
|
||
# 不变量 #3：四审只读
|
||
assert foreshadow_spec.writes == []
|
||
|
||
|
||
def test_foreshadow_spec_reads_foreshadow() -> None:
|
||
assert foreshadow_spec.reads == ["foreshadow"]
|
||
|
||
|
||
def test_foreshadow_spec_output_schema() -> None:
|
||
assert foreshadow_spec.output_schema is ForeshadowReview
|
||
|
||
|
||
def test_foreshadow_spec_has_nonempty_system_prompt() -> None:
|
||
assert foreshadow_spec.system_prompt.strip()
|
||
|
||
|
||
# ---- pace_spec 声明 ----
|
||
|
||
|
||
def test_pace_spec_is_light_tier() -> None:
|
||
# 不变量 #2：节奏审用轻量档，只声明 tier
|
||
assert pace_spec.tier == "light"
|
||
assert pace_spec.name == "pace"
|
||
|
||
|
||
def test_pace_spec_is_read_only() -> None:
|
||
assert pace_spec.writes == []
|
||
|
||
|
||
def test_pace_spec_reads_rules() -> None:
|
||
# genre 模板 DSL 经 rules（genre 级）注入
|
||
assert pace_spec.reads == ["rules"]
|
||
|
||
|
||
def test_pace_spec_output_schema() -> None:
|
||
assert pace_spec.output_schema is PaceReview
|
||
|
||
|
||
def test_pace_spec_has_nonempty_system_prompt() -> None:
|
||
assert pace_spec.system_prompt.strip()
|
||
|
||
|
||
# ---- 共性：immutable + 是 AgentSpec ----
|
||
|
||
|
||
def test_review_specs_are_agent_specs() -> None:
|
||
assert isinstance(foreshadow_spec, AgentSpec)
|
||
assert isinstance(pace_spec, AgentSpec)
|
||
|
||
|
||
def test_review_specs_are_immutable() -> None:
|
||
with pytest.raises(ValidationError):
|
||
foreshadow_spec.tier = "writer"
|
||
with pytest.raises(ValidationError):
|
||
pace_spec.tier = "writer"
|