feat: M3 — 伏笔账本 + 节奏引擎 + 大纲（含并发记账 bugfix）

- 伏笔账本：纯函数状态机(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
2026-06-18 14:21:17 +02:00
parent 68f194a043
commit 5fb7bfb1de
74 changed files with 6529 additions and 126 deletions
--- a/packages/agents/tests/test_outliner_spec.py
+++ b/packages/agents/tests/test_outliner_spec.py
@@ -0,0 +1,117 @@
+"""T3.4 大纲 Agent 契约测试：OutlineResult schema + outliner_spec 声明（C6 扩）。
+
+契约测试——构造符合 schema 的 mock 响应，校验伏笔窗口字段、章节结构、
+声明式 writes（验收/T3.5 才真写库，不变量 #3）。不联网、无 DB。
+"""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+from ww_agents import (
+    AgentSpec,
+    ForeshadowWindow,
+    OutlineChapter,
+    OutlineResult,
+    outliner_spec,
+)
+
+
+def test_outline_result_parses_mock_response() -> None:
+    # Arrange：模拟网关 instructor 校验后的结构化大纲产出
+    mock = {
+        "chapters": [
+            {
+                "no": 1,
+                "beats": ["主角登场，立下目标", "埋下身世伏笔"],
+                "foreshadow_windows": [
+                    {
+                        "code": "FS-BIRTH",
+                        "plant_chapter": 1,
+                        "expected_close_from": 20,
+                        "expected_close_to": 30,
+                    }
+                ],
+            },
+            {
+                "no": 2,
+                "beats": ["遭遇对手"],
+                "foreshadow_windows": [],
+            },
+        ]
+    }
+
+    # Act
+    outline = OutlineResult.model_validate(mock)
+
+    # Assert
+    assert len(outline.chapters) == 2
+    assert outline.chapters[0].no == 1
+    assert outline.chapters[0].beats == ["主角登场，立下目标", "埋下身世伏笔"]
+    window = outline.chapters[0].foreshadow_windows[0]
+    assert window.code == "FS-BIRTH"
+    assert window.plant_chapter == 1
+    assert window.expected_close_from == 20
+    assert window.expected_close_to == 30
+
+
+def test_outline_chapter_defaults_empty_windows() -> None:
+    chapter = OutlineChapter.model_validate({"no": 3, "beats": ["收束支线"]})
+    assert chapter.foreshadow_windows == []
+
+
+def test_outline_result_defaults_to_empty_chapters() -> None:
+    assert OutlineResult().chapters == []
+
+
+def test_foreshadow_window_optional_bounds_default_none() -> None:
+    window = ForeshadowWindow.model_validate({"code": "FS-X"})
+    assert window.plant_chapter is None
+    assert window.expected_close_from is None
+    assert window.expected_close_to is None
+
+
+def test_foreshadow_window_requires_code() -> None:
+    with pytest.raises(ValidationError):
+        ForeshadowWindow.model_validate({"plant_chapter": 1})
+
+
+def test_outline_chapter_requires_no() -> None:
+    with pytest.raises(ValidationError):
+        OutlineChapter.model_validate({"beats": ["x"]})
+
+
+def test_outliner_spec_is_analyst_tier() -> None:
+    assert outliner_spec.tier == "analyst"
+    assert outliner_spec.name == "outliner"
+
+
+def test_outliner_spec_declares_expected_reads() -> None:
+    assert outliner_spec.reads == [
+        "projects",
+        "foreshadow",
+        "characters",
+        "world_entities",
+    ]
+
+
+def test_outliner_spec_declares_outline_write() -> None:
+    # 声明式 writes（经验收/T3.5 才真写库，不变量 #3）
+    assert outliner_spec.writes == ["outline"]
+
+
+def test_outliner_spec_output_schema_is_outline_result() -> None:
+    assert outliner_spec.output_schema is OutlineResult
+
+
+def test_outliner_spec_has_nonempty_system_prompt() -> None:
+    assert outliner_spec.system_prompt.strip()
+
+
+def test_outliner_spec_is_immutable() -> None:
+    with pytest.raises(ValidationError):
+        outliner_spec.tier = "writer"
+
+
+def test_outliner_spec_is_agent_spec() -> None:
+    assert isinstance(outliner_spec, AgentSpec)
--- a/packages/agents/tests/test_review_specs.py
+++ b/packages/agents/tests/test_review_specs.py
@@ -0,0 +1,167 @@
+"""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"
--- a/packages/agents/ww_agents/init.py
+++ b/packages/agents/ww_agents/init.py
@@ -1,17 +1,44 @@
 """ww-agents：内置 Agent 声明（AgentSpec）+ 结构化输出 schema（C6）。

-对齐 ARCH §5.1（AgentSpec）/ §5.4（I/O 契约）/ §6.1（冲突分类）。
+对齐 ARCH §5.1（AgentSpec）/ §5.4（I/O 契约）/ §6.1（冲突分类）/ §6.2 §6.4（伏笔/节奏）。
 """

 from __future__ import annotations

-from .schemas import Conflict, ConflictType, ContinuityReview
-from .specs import AgentSpec, continuity_spec
+from .schemas import (
+    Conflict,
+    ConflictType,
+    ContinuityReview,
+    ForeshadowReview,
+    ForeshadowSuggestion,
+    ForeshadowWindow,
+    OutlineChapter,
+    OutlineResult,
+    PaceIssue,
+    PaceReview,
+)
+from .specs import (
+    AgentSpec,
+    continuity_spec,
+    foreshadow_spec,
+    outliner_spec,
+    pace_spec,
+)

 __all__ = [
    "AgentSpec",
    "Conflict",
    "ConflictType",
    "ContinuityReview",
+    "ForeshadowReview",
+    "ForeshadowSuggestion",
+    "ForeshadowWindow",
+    "OutlineChapter",
+    "OutlineResult",
+    "PaceIssue",
+    "PaceReview",
    "continuity_spec",
+    "foreshadow_spec",
+    "outliner_spec",
+    "pace_spec",
 ]
--- a/packages/agents/ww_agents/schemas.py
+++ b/packages/agents/ww_agents/schemas.py
@@ -44,3 +44,106 @@ class ContinuityReview(BaseModel):
        default_factory=list,
        description="检出的一致性冲突；无冲突时为空列表",
    )
+
+
+# ---- 大纲 Agent 结构化输出（ARCH §5.4 outliner 行 / §6.2 伏笔窗口）----
+
+
+class ForeshadowWindow(BaseModel):
+    """伏笔回收窗口：关联本章与某条伏笔（ARCH §6.2）。
+
+    排大纲时把伏笔的「埋设章 → 期望回收区间」绑到具体章节；接近回收窗口的章节
+    提示作者推进/收束。下界/上界可缺省（仅埋设、尚未排定回收时间）。
+    """
+
+    code: str = Field(description="伏笔编码（关联 foreshadow.code）")
+    plant_chapter: int | None = Field(
+        default=None, description="埋设章号（伏笔在本章/此章首次出现）"
+    )
+    expected_close_from: int | None = Field(default=None, description="期望回收区间下界（章号）")
+    expected_close_to: int | None = Field(default=None, description="期望回收区间上界（章号）")
+
+
+class OutlineChapter(BaseModel):
+    """单章大纲：章号 + 节拍清单 + 关联伏笔窗口。"""
+
+    no: int = Field(description="章号")
+    beats: list[str] = Field(default_factory=list, description="本章节拍/情节要点（有序）")
+    foreshadow_windows: list[ForeshadowWindow] = Field(
+        default_factory=list,
+        description="本章关联的伏笔回收窗口；无则空列表",
+    )
+
+
+class OutlineResult(BaseModel):
+    """大纲结构化产出：分卷分章的章节大纲清单（ARCH §5.4 writes=outline）。
+
+    声明式 writes（不变量 #3）：本结构是排大纲的纯产物，落 `outline` 表经验收/
+    端点（T3.5），不在 agent 节点直接写库。
+    """
+
+    chapters: list[OutlineChapter] = Field(
+        default_factory=list,
+        description="分章大纲；无则空列表",
+    )
+
+
+# ---- 伏笔续审 Agent 结构化输出（ARCH §5.4 foreshadow-analyst 行 / §6.2）----
+
+
+class ForeshadowSuggestion(BaseModel):
+    """单条伏笔建议：本章新埋线 或 疑似回收（ARCH §6.2）。
+
+    只读建议——作者在验收裁决里确认后才登记/改状态（不变量 #3）；本审不写库。
+    `code` 可缺（新埋线尚无编码时由作者命名）；`where` 给本章定位。
+    """
+
+    code: str | None = Field(
+        default=None, description="伏笔编码（关联已登记 foreshadow.code；新埋可缺）"
+    )
+    title: str = Field(description="伏笔标题/一句话描述")
+    where: str | None = Field(default=None, description="本章定位：哪一段埋下/疑似回收")
+    note: str | None = Field(default=None, description="建议说明（埋设要点 / 回收理由）")
+
+
+class ForeshadowReview(BaseModel):
+    """伏笔续审结构化产出：本章新埋 + 疑似回收两组建议（ARCH §5.4 / §6.2）。
+
+    只读建议清单（不变量 #3）；登记/状态变更经验收裁决（作者确认）。
+    """
+
+    planted: list[ForeshadowSuggestion] = Field(
+        default_factory=list,
+        description="本章疑似新埋的伏笔；无则空列表",
+    )
+    resolved: list[ForeshadowSuggestion] = Field(
+        default_factory=list,
+        description="本章疑似回收/收束的伏笔；无则空列表",
+    )
+
+
+# ---- 节奏续审 Agent 结构化输出（ARCH §5.4 pace-checker 行 / §6.4）----
+
+
+class PaceIssue(BaseModel):
+    """单条节奏问题：注水段定位 + 原因（ARCH §6.4）。"""
+
+    where: str = Field(description="本章定位：哪一段疑似注水/拖沓")
+    reason: str = Field(description="判定理由（信息密度低 / 重复 / 偏题等）")
+
+
+class PaceReview(BaseModel):
+    """节奏续审结构化产出：注水段 + 章末钩子 + 爽点节拍图（ARCH §5.4 / §6.4）。
+
+    只读（不变量 #3）。`beat_map` 是逐段节拍强度序列（整数），供前端 ▁▃▅ 可视化。
+    """
+
+    water: list[PaceIssue] = Field(
+        default_factory=list,
+        description="疑似注水/拖沓段；无则空列表",
+    )
+    hook: bool = Field(default=False, description="章末是否有有效钩子（悬念/反转/期待）")
+    beat_map: list[int] = Field(
+        default_factory=list,
+        description="逐段爽点节拍强度序列（整数，供前端节拍图可视化）",
+    )
--- a/packages/agents/ww_agents/specs.py
+++ b/packages/agents/ww_agents/specs.py
@@ -12,7 +12,12 @@ from __future__ import annotations
 from pydantic import BaseModel, ConfigDict, Field
 from ww_llm_gateway.types import Tier

-from .schemas import ContinuityReview
+from .schemas import (
+    ContinuityReview,
+    ForeshadowReview,
+    OutlineResult,
+    PaceReview,
+)


 class AgentSpec(BaseModel):
@@ -63,3 +68,99 @@ continuity_spec = AgentSpec(
    writes=[],  # 只读（不变量 #3）
    scope="builtin",
 )
+
+
+OUTLINER_SYSTEM_PROMPT = """你是长篇连载小说的「大纲架构师」。职责：依据作品立意、\
+人物与世界观，排出分卷分章的章节大纲，并为每条伏笔标注回收窗口。
+
+输入材料：
+- 作品设定（projects：题材、立意、主线、卖点）；
+- 已登记伏笔（foreshadow：编码、标题、埋设/期望回收线索）；
+- 主要人物（characters）与世界观实体（world_entities）。
+
+产出要求（每章）：
+- 章号有序、按卷推进；
+- beats：本章核心节拍/情节要点，推动主线、服务人物弧光；
+- foreshadow_windows：本章关联的伏笔回收窗口——给出伏笔编码、埋设章号、\
+期望回收区间（下界/上界章号）。把每条伏笔的「埋设 → 回收」绑到具体章节。
+
+纪律：
+- 接近某伏笔的回收窗口时，在对应章 beats 里安排推进/收束该伏笔的情节，避免伏笔悬置过久；
+- 你只产结构化大纲（章节 + 节拍 + 伏笔窗口），**不改稿、不写库**（落 outline 表经验收/端点）；
+- 伏笔编码须与已登记 foreshadow.code 对应，便于后续按窗口提示与校验。"""
+
+
+outliner_spec = AgentSpec(
+    name="outliner",
+    tier="analyst",  # 不变量 #2：只声明档位，不写 model
+    system_prompt=OUTLINER_SYSTEM_PROMPT,
+    input_schema=None,  # 注入材料为序列化文本（设定/伏笔/人物/世界观），非结构化入参
+    output_schema=OutlineResult,
+    reads=["projects", "foreshadow", "characters", "world_entities"],
+    writes=["outline"],  # 声明式（经验收/T3.5 才真写库，不变量 #3）
+    scope="builtin",
+)
+
+
+FORESHADOW_SYSTEM_PROMPT = """你是长篇连载小说的「伏笔续审」（foreshadow-analyst）。\
+职责：把本章草稿与作品已登记伏笔逐项比对，找出本章**新埋的伏笔**与**疑似回收/收束**\
+的伏笔，产出结构化建议清单。
+
+比对依据（注入材料）：
+- 已登记伏笔（foreshadow：编码、标题、状态、期望回收窗口）；
+- 本章草稿正文。
+
+产出两组建议：
+- planted（新埋）：本章疑似首次埋下的伏笔——给出标题、本章定位、要点说明；\
+若与某已登记编码相关则填 code，否则留空由作者命名。
+- resolved（回收）：本章疑似回收/收束已登记伏笔——给出对应 code、本章定位、回收理由。
+
+纪律：
+- 你**只读、只产建议**，不改稿、不写库、不直接登记或改伏笔状态——\
+登记/状态变更经作者在验收时裁决确认（不变量 #3 / #4）。
+- 只报有据可依的；无则两组都返回空列表，不要臆造。
+- 引用具体（伏笔编码、本章段落定位），便于作者就地确认。"""
+
+
+foreshadow_spec = AgentSpec(
+    name="foreshadow",
+    tier="analyst",  # 不变量 #2：只声明档位，不写 model
+    system_prompt=FORESHADOW_SYSTEM_PROMPT,
+    input_schema=None,  # 注入材料为序列化文本（已登记伏笔 + 草稿），非结构化入参
+    output_schema=ForeshadowReview,
+    reads=["foreshadow"],
+    writes=[],  # 只读（不变量 #3）
+    scope="builtin",
+)
+
+
+PACE_SYSTEM_PROMPT = """你是长篇连载小说的「节奏续审」（pace-checker）。职责：按题材的\
+节奏模板审本章草稿，找出注水段、判定章末钩子有无、给出爽点节拍图。
+
+题材节奏模板（genre 级规则，注入材料里带本作题材的具体模板；无则用网文通用基线）：
+- 黄金三章：开篇前三章须快速立爽点/钩子/代入感，信息密度高、少铺垫慢热；
+- 章末钩子：每章结尾应留悬念/反转/期待，驱动追更；
+- 爽点密度：按题材基线维持爽点节拍，避免长段平淡注水。
+
+产出：
+- water（注水段）：逐段审，标出信息密度低/重复/偏题/拖沓的段落，给本章定位与原因；
+- hook（章末钩子）：本章结尾是否存在有效钩子（true/false）；
+- beat_map（节拍图）：把本章按段切分，给每段一个爽点强度整数（如 0–5），\
+形成逐段强度序列，供前端 ▁▃▅ 可视化节奏起伏。
+
+纪律：
+- 你**只读、只报节奏诊断**，不改稿、不写库（不变量 #3）。
+- 依据题材模板判定，不臆造；无注水段则 water 为空列表。
+- beat_map 长度应与切分段数一致、顺序即正文顺序，便于前端对齐渲染。"""
+
+
+pace_spec = AgentSpec(
+    name="pace",
+    tier="light",  # 不变量 #2：只声明档位，不写 model（节奏审用轻量档）
+    system_prompt=PACE_SYSTEM_PROMPT,
+    input_schema=None,  # 注入材料为序列化文本（题材模板 + 草稿），非结构化入参
+    output_schema=PaceReview,
+    reads=["rules"],  # genre 模板 DSL 经 rules（genre 级）注入，复用 review_context 规则合并
+    writes=[],  # 只读（不变量 #3）
+    scope="builtin",
+)
--- a/packages/core/tests/fakes_orchestrator.py
+++ b/packages/core/tests/fakes_orchestrator.py
@@ -77,6 +77,29 @@ class FakeRunGateway:
        )


+class SchemaRoutingRunGateway:
+    """按 `req.output_schema` 路由返对应 `parsed`——并行三审各拿自己 schema 的产物。
+
+    `by_schema`：{schema 类型: 该 schema 的实例}。命中即回该 parsed；未命中抛 KeyError
+    （测试要保证三审 schema 都登记）。统计每个 schema 的调用次数。
+    """
+
+    def __init__(self, by_schema: dict[type[BaseModel], BaseModel]) -> None:
+        self._by_schema = by_schema
+        self.calls: list[type[BaseModel] | None] = []
+
+    async def run(self, req: LlmRequest) -> LlmResponse:
+        schema = req.output_schema
+        self.calls.append(schema)
+        parsed = self._by_schema[schema] if schema is not None else None
+        return LlmResponse(
+            text=parsed.model_dump_json() if parsed is not None else "",
+            parsed=parsed,
+            usage=_stub_usage(),
+            served_by=ServedBy(provider="fake", model="fake-analyst"),
+        )
+
+
 class FailingRunGateway:
    """`run` 抛指定异常——验「某审失败不阻塞其余」+ 未完成占位归一。"""

--- a/packages/core/tests/test_foreshadow_repo.py
+++ b/packages/core/tests/test_foreshadow_repo.py
@@ -0,0 +1,293 @@
+"""T3.1 伏笔账本写侧 repo 单测（ARCH §6.2）。
+
+纯内存 fake，无 DB：断言 register 唯一、list_by_status 过滤、record_progress append
+不覆盖、transition 走状态机校验、scan_overdue 命中集合、按 project_id 隔离、frozen
+View 不可变。`(project_id, code)` 唯一性是 **DB 级**（models UniqueConstraint），纯 fake
+不引 pg 断言——真实 DB 路径由 T3.7 E2E 覆盖。
+"""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass, field
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+from ww_core.domain.foreshadow_repo import ForeshadowLedgerRepo, ForeshadowLedgerView
+from ww_core.domain.foreshadow_state import (
+    CLOSED,
+    OVERDUE,
+    PARTIAL,
+    InvalidTransition,
+)
+
+PROJECT = uuid.UUID("00000000-0000-0000-0000-000000000001")
+OTHER = uuid.UUID("00000000-0000-0000-0000-000000000002")
+
+
+@dataclass
+class _Row:
+    project_id: uuid.UUID
+    code: str
+    title: str
+    status: str
+    planted_at: int | None
+    content: str | None
+    expected_close_from: int | None
+    expected_close_to: int | None
+    importance: str | None
+    links: list[Any]
+    progress: list[Any]
+
+
+def _view(row: _Row) -> ForeshadowLedgerView:
+    return ForeshadowLedgerView(
+        code=row.code,
+        title=row.title,
+        status=row.status,
+        planted_at=row.planted_at,
+        content=row.content,
+        expected_close_from=row.expected_close_from,
+        expected_close_to=row.expected_close_to,
+        importance=row.importance,
+        links=list(row.links),
+        progress=list(row.progress),
+    )
+
+
+@dataclass
+class FakeForeshadowLedgerRepo:
+    """内存替身——镜像 SqlForeshadowLedgerRepo 语义（按 (project_id, code) 唯一）。"""
+
+    rows: list[_Row] = field(default_factory=list)
+
+    def _find(self, project_id: uuid.UUID, code: str) -> _Row | None:
+        return next((r for r in self.rows if r.project_id == project_id and r.code == code), None)
+
+    async def register(
+        self,
+        project_id: uuid.UUID,
+        *,
+        code: str,
+        title: str,
+        planted_at: int | None = None,
+        content: str | None = None,
+        expected_close_from: int | None = None,
+        expected_close_to: int | None = None,
+        importance: str | None = None,
+    ) -> ForeshadowLedgerView:
+        if self._find(project_id, code) is not None:
+            raise ValueError(f"duplicate foreshadow code: {code}")
+        row = _Row(
+            project_id=project_id,
+            code=code,
+            title=title,
+            status="OPEN",
+            planted_at=planted_at,
+            content=content,
+            expected_close_from=expected_close_from,
+            expected_close_to=expected_close_to,
+            importance=importance,
+            links=[],
+            progress=[],
+        )
+        self.rows.append(row)
+        return _view(row)
+
+    async def get(self, project_id: uuid.UUID, code: str) -> ForeshadowLedgerView | None:
+        row = self._find(project_id, code)
+        return _view(row) if row is not None else None
+
+    async def list_by_status(
+        self, project_id: uuid.UUID, status: str | None = None
+    ) -> list[ForeshadowLedgerView]:
+        return [
+            _view(r)
+            for r in self.rows
+            if r.project_id == project_id and (status is None or r.status == status)
+        ]
+
+    async def transition(
+        self, project_id: uuid.UUID, code: str, *, to_status: str
+    ) -> ForeshadowLedgerView:
+        from ww_core.domain.foreshadow_state import ForeshadowStatus
+        from ww_core.domain.foreshadow_state import transition as apply_transition
+
+        row = self._find(project_id, code)
+        if row is None:
+            raise LookupError(f"foreshadow not found: {code}")
+        row.status = apply_transition(ForeshadowStatus(row.status), ForeshadowStatus(to_status))
+        return _view(row)
+
+    async def record_progress(
+        self, project_id: uuid.UUID, code: str, *, entry: dict[str, Any]
+    ) -> ForeshadowLedgerView:
+        row = self._find(project_id, code)
+        if row is None:
+            raise LookupError(f"foreshadow not found: {code}")
+        row.progress = [*row.progress, dict(entry)]
+        return _view(row)
+
+    async def scan_overdue(
+        self, project_id: uuid.UUID, *, current_chapter: int
+    ) -> list[ForeshadowLedgerView]:
+        from ww_core.domain.foreshadow_state import ForeshadowStatus, apply_overdue_scan
+
+        changed: list[ForeshadowLedgerView] = []
+        for r in self.rows:
+            if r.project_id != project_id:
+                continue
+            new = apply_overdue_scan(
+                ForeshadowStatus(r.status),
+                current_chapter=current_chapter,
+                expected_close_to=r.expected_close_to,
+            )
+            if new != r.status:
+                r.status = new
+                changed.append(_view(r))
+        return changed
+
+
+def _repo() -> FakeForeshadowLedgerRepo:
+    return FakeForeshadowLedgerRepo()
+
+
+# ---- register ----
+
+
+async def test_register_starts_open() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    view = await repo.register(
+        PROJECT, code="F1", title="神秘玉佩", planted_at=3, expected_close_to=20
+    )
+    assert isinstance(view, ForeshadowLedgerView)
+    assert view.code == "F1"
+    assert view.status == "OPEN"
+    assert view.planted_at == 3
+    assert view.expected_close_to == 20
+    assert view.progress == []
+
+
+async def test_register_duplicate_code_rejected() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="F1", title="a")
+    with pytest.raises(ValueError):
+        await repo.register(PROJECT, code="F1", title="b")
+
+
+async def test_register_same_code_other_project_ok() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="F1", title="a")
+    other = await repo.register(OTHER, code="F1", title="b")
+    assert other.code == "F1"
+
+
+# ---- get / list_by_status ----
+
+
+async def test_get_returns_none_when_absent() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    assert await repo.get(PROJECT, "nope") is None
+
+
+async def test_list_by_status_filters() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="A", title="a")
+    b = await repo.register(PROJECT, code="B", title="b")
+    assert b.status == "OPEN"
+    await repo.transition(PROJECT, "B", to_status=PARTIAL)
+
+    opens = await repo.list_by_status(PROJECT, "OPEN")
+    assert {v.code for v in opens} == {"A"}
+    partials = await repo.list_by_status(PROJECT, PARTIAL)
+    assert {v.code for v in partials} == {"B"}
+
+
+async def test_list_by_status_none_returns_all() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="A", title="a")
+    await repo.register(PROJECT, code="B", title="b")
+    assert len(await repo.list_by_status(PROJECT)) == 2
+
+
+async def test_list_by_status_isolated_by_project() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="A", title="a")
+    await repo.register(OTHER, code="B", title="b")
+    rows = await repo.list_by_status(PROJECT)
+    assert len(rows) == 1 and rows[0].code == "A"
+
+
+# ---- transition (走状态机) ----
+
+
+async def test_transition_applies_state_machine() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="A", title="a")
+    v = await repo.transition(PROJECT, "A", to_status=PARTIAL)
+    assert v.status == "PARTIAL"
+    v2 = await repo.transition(PROJECT, "A", to_status=CLOSED)
+    assert v2.status == "CLOSED"
+
+
+async def test_transition_rejects_illegal() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="A", title="a")
+    await repo.transition(PROJECT, "A", to_status=CLOSED)
+    with pytest.raises(InvalidTransition):
+        await repo.transition(PROJECT, "A", to_status=PARTIAL)
+
+
+# ---- record_progress (append-only) ----
+
+
+async def test_record_progress_appends_not_overwrites() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="A", title="a")
+    await repo.record_progress(PROJECT, "A", entry={"chapter": 5, "note": "first hint"})
+    v = await repo.record_progress(PROJECT, "A", entry={"chapter": 9, "note": "second"})
+    assert len(v.progress) == 2
+    assert v.progress[0]["chapter"] == 5
+    assert v.progress[1]["chapter"] == 9
+
+
+# ---- scan_overdue ----
+
+
+async def test_scan_overdue_marks_due_rows() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="DUE", title="d", expected_close_to=10)
+    await repo.register(PROJECT, code="NOTYET", title="n", expected_close_to=30)
+    await repo.register(PROJECT, code="NOWIN", title="w")  # 无窗口
+    changed = await repo.scan_overdue(PROJECT, current_chapter=15)
+    assert {v.code for v in changed} == {"DUE"}
+    assert all(v.status == "OVERDUE" for v in changed)
+    # NOTYET / NOWIN 保持 OPEN
+    assert (await repo.get(PROJECT, "NOTYET")).status == "OPEN"  # type: ignore[union-attr]
+
+
+async def test_scan_overdue_skips_closed() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(PROJECT, code="C", title="c", expected_close_to=10)
+    await repo.transition(PROJECT, "C", to_status=CLOSED)
+    changed = await repo.scan_overdue(PROJECT, current_chapter=99)
+    assert changed == []
+    assert (await repo.get(PROJECT, "C")).status == "CLOSED"  # type: ignore[union-attr]
+
+
+async def test_scan_overdue_isolated_by_project() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    await repo.register(OTHER, code="X", title="x", expected_close_to=5)
+    changed = await repo.scan_overdue(PROJECT, current_chapter=99)
+    assert changed == []  # OTHER 的行不受 PROJECT 扫描影响
+
+
+# ---- frozen View ----
+
+
+async def test_view_is_frozen() -> None:
+    repo: ForeshadowLedgerRepo = _repo()
+    v = await repo.register(PROJECT, code="A", title="a")
+    with pytest.raises(ValidationError):
+        v.status = OVERDUE
--- a/packages/core/tests/test_foreshadow_state.py
+++ b/packages/core/tests/test_foreshadow_state.py
@@ -0,0 +1,143 @@
+"""T3.1 伏笔状态机纯函数单测（ARCH §6.2）。
+
+确定性纯函数，无 IO：覆盖全转移（登记/首次照应/完成回收）+ OVERDUE 到期判据边界
+（current ==/</> expected_close_to；CLOSED 不被覆盖；无 expected_close_to 不 OVERDUE）+
+非法转移显式处理。
+"""
+
+from __future__ import annotations
+
+import pytest
+from ww_core.domain.foreshadow_state import (
+    CLOSED,
+    OPEN,
+    OVERDUE,
+    PARTIAL,
+    ForeshadowStatus,
+    InvalidTransition,
+    apply_overdue_scan,
+    is_overdue,
+    transition,
+)
+
+# ---- 合法转移（登记→OPEN→PARTIAL→CLOSED）----
+
+
+def test_open_to_partial() -> None:
+    assert transition(OPEN, PARTIAL) == PARTIAL
+
+
+def test_partial_to_closed() -> None:
+    assert transition(PARTIAL, CLOSED) == CLOSED
+
+
+def test_open_to_closed_direct_recover() -> None:
+    # 一步回收（埋后直接收）也合法
+    assert transition(OPEN, CLOSED) == CLOSED
+
+
+def test_overdue_to_closed_recover_after_due() -> None:
+    # 逾期后仍可回收
+    assert transition(OVERDUE, CLOSED) == CLOSED
+
+
+def test_overdue_to_partial_partial_recover_after_due() -> None:
+    assert transition(OVERDUE, PARTIAL) == PARTIAL
+
+
+# ---- 非法转移显式拒绝 ----
+
+
+def test_closed_cannot_reopen() -> None:
+    with pytest.raises(InvalidTransition):
+        transition(CLOSED, OPEN)
+
+
+def test_closed_cannot_become_overdue_via_transition() -> None:
+    with pytest.raises(InvalidTransition):
+        transition(CLOSED, OVERDUE)
+
+
+def test_closed_cannot_become_partial() -> None:
+    with pytest.raises(InvalidTransition):
+        transition(CLOSED, PARTIAL)
+
+
+def test_partial_cannot_go_back_to_open() -> None:
+    with pytest.raises(InvalidTransition):
+        transition(PARTIAL, OPEN)
+
+
+def test_same_status_is_noop_allowed() -> None:
+    # 幂等：同态转移允许（看板手动设回同值不报错）
+    assert transition(OPEN, OPEN) == OPEN
+    assert transition(PARTIAL, PARTIAL) == PARTIAL
+
+
+# ---- is_overdue 判据边界 ----
+
+
+def test_is_overdue_strictly_past_window() -> None:
+    assert is_overdue(current_chapter=11, expected_close_to=10, status=OPEN) is True
+
+
+def test_is_overdue_at_window_boundary_not_overdue() -> None:
+    # current == expected_close_to：仍在窗口内，未逾期
+    assert is_overdue(current_chapter=10, expected_close_to=10, status=OPEN) is False
+
+
+def test_is_overdue_before_window_not_overdue() -> None:
+    assert is_overdue(current_chapter=5, expected_close_to=10, status=OPEN) is False
+
+
+def test_is_overdue_closed_never_overdue() -> None:
+    assert is_overdue(current_chapter=99, expected_close_to=10, status=CLOSED) is False
+
+
+def test_is_overdue_partial_can_be_overdue() -> None:
+    assert is_overdue(current_chapter=11, expected_close_to=10, status=PARTIAL) is True
+
+
+def test_is_overdue_without_window_never_overdue() -> None:
+    # 无 expected_close_to → 永不逾期
+    assert is_overdue(current_chapter=999, expected_close_to=None, status=OPEN) is False
+
+
+def test_is_overdue_already_overdue_stays() -> None:
+    assert is_overdue(current_chapter=11, expected_close_to=10, status=OVERDUE) is True
+
+
+# ---- apply_overdue_scan ----
+
+
+def test_apply_overdue_scan_sets_overdue_when_due() -> None:
+    new = apply_overdue_scan(OPEN, current_chapter=11, expected_close_to=10)
+    assert new == OVERDUE
+
+
+def test_apply_overdue_scan_keeps_status_when_not_due() -> None:
+    assert apply_overdue_scan(OPEN, current_chapter=10, expected_close_to=10) == OPEN
+
+
+def test_apply_overdue_scan_does_not_touch_closed() -> None:
+    # CLOSED 不被 OVERDUE 覆盖（不变量）
+    assert apply_overdue_scan(CLOSED, current_chapter=99, expected_close_to=10) == CLOSED
+
+
+def test_apply_overdue_scan_no_window_keeps_status() -> None:
+    assert apply_overdue_scan(PARTIAL, current_chapter=99, expected_close_to=None) == PARTIAL
+
+
+def test_apply_overdue_scan_idempotent_on_overdue() -> None:
+    assert apply_overdue_scan(OVERDUE, current_chapter=12, expected_close_to=10) == OVERDUE
+
+
+# ---- 枚举值与 DB 文本一致 ----
+
+
+def test_status_values_match_db_text() -> None:
+    assert OPEN == "OPEN"
+    assert PARTIAL == "PARTIAL"
+    assert CLOSED == "CLOSED"
+    assert OVERDUE == "OVERDUE"
+    assert {s.value for s in ForeshadowStatus} == {"OPEN", "PARTIAL", "CLOSED", "OVERDUE"}
--- a/packages/core/tests/test_outline_node.py
+++ b/packages/core/tests/test_outline_node.py
@@ -0,0 +1,137 @@
+"""T3.4 大纲节点单测——build_outline_request + run_outline（C6 扩 / ARCH §5.4 §6.2）。
+
+注入 mock 网关（产 `OutlineResult` parsed），无真 LLM、无真 Postgres。
+大纲节点是裸函数（独立生成，不在写章/审稿流水线）；只产结构化大纲，**不写库**
+（持久化属 T3.5 验收/端点，不变量 #3）。asyncio_mode=auto。
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import pytest
+from fakes_orchestrator import FailingRunGateway, FakeRunGateway
+from ww_agents import (
+    ForeshadowWindow,
+    OutlineChapter,
+    OutlineResult,
+    outliner_spec,
+)
+from ww_core.orchestrator import build_outline_request, run_outline
+
+PROJECT = uuid.UUID("00000000-0000-0000-0000-000000000001")
+USER = uuid.UUID("00000000-0000-0000-0000-0000000000aa")
+
+CONTEXT = "## 立意\n复仇成长\n## 已开伏笔\nFS-BIRTH: 身世之谜（待第20-30章回收）"
+
+_OUTLINE = OutlineResult(
+    chapters=[
+        OutlineChapter(
+            no=1,
+            beats=["主角登场", "埋身世伏笔"],
+            foreshadow_windows=[
+                ForeshadowWindow(
+                    code="FS-BIRTH",
+                    plant_chapter=1,
+                    expected_close_from=20,
+                    expected_close_to=30,
+                )
+            ],
+        ),
+        OutlineChapter(no=2, beats=["遭遇对手"], foreshadow_windows=[]),
+    ]
+)
+
+
+# ---- build_outline_request：tier 不变量、缓存断点、output_schema、run 非 stream ----
+
+
+def test_build_outline_request_uses_spec_tier_and_cached_prompt() -> None:
+    req = build_outline_request(outliner_spec, context=CONTEXT, user_id=USER, project_id=PROJECT)
+
+    assert req.tier == "analyst"  # 不变量②：spec 档位透传，不传 model
+    assert req.stream is False  # 排大纲用 run() 非 stream()
+    assert len(req.system) == 1
+    assert req.system[0].text == outliner_spec.system_prompt
+    assert req.system[0].cache is True  # 不变量#9
+    assert req.input == CONTEXT
+    assert req.output_schema is OutlineResult
+    assert req.scope.user_id == USER
+    assert req.scope.project_id == PROJECT
+
+
+# ---- run_outline：返回结构化大纲含 foreshadow_windows，只读不写库 ----
+
+
+async def test_run_outline_returns_structured_outline() -> None:
+    gateway = FakeRunGateway(_OUTLINE)
+
+    result = await run_outline(
+        outliner_spec, context=CONTEXT, gateway=gateway, user_id=USER, project_id=PROJECT
+    )
+
+    assert isinstance(result, OutlineResult)
+    assert len(result.chapters) == 2
+    assert result.chapters[0].no == 1
+    window = result.chapters[0].foreshadow_windows[0]
+    assert window.code == "FS-BIRTH"
+    assert window.expected_close_to == 30
+    assert gateway.call_count == 1
+
+
+async def test_run_outline_forwards_constructed_request() -> None:
+    gateway = FakeRunGateway(_OUTLINE)
+
+    await run_outline(
+        outliner_spec, context=CONTEXT, gateway=gateway, user_id=USER, project_id=PROJECT
+    )
+
+    req = gateway.last_request
+    assert req is not None
+    assert req.tier == "analyst"
+    assert req.output_schema is OutlineResult
+    assert req.input == CONTEXT
+
+
+async def test_run_outline_raises_when_gateway_fails() -> None:
+    # 大纲是独立生成（非流水线并行审）；网关失败直接上抛，端点/T3.5 处理。
+    gateway = FailingRunGateway(RuntimeError("provider down"))
+
+    with pytest.raises(RuntimeError):
+        await run_outline(
+            outliner_spec,
+            context=CONTEXT,
+            gateway=gateway,
+            user_id=USER,
+            project_id=PROJECT,
+        )
+
+
+async def test_run_outline_raises_when_parsed_missing() -> None:
+    # 带 output_schema 时 parsed 必非 None（C1）；若网关违约返回 None，明确报错而非静默。
+    class _NoParseGateway:
+        async def run(self, req: object) -> object:
+            from ww_llm_gateway.types import LlmResponse, ServedBy, Usage
+
+            return LlmResponse(
+                text="{}",
+                parsed=None,
+                usage=Usage(
+                    provider="fake",
+                    model="fake-analyst",
+                    input_tokens=1,
+                    output_tokens=1,
+                    cost_minor=0,
+                    currency="USD",
+                ),
+                served_by=ServedBy(provider="fake", model="fake-analyst"),
+            )
+
+    with pytest.raises(ValueError, match="parsed"):
+        await run_outline(
+            outliner_spec,
+            context=CONTEXT,
+            gateway=_NoParseGateway(),  # type: ignore[arg-type]
+            user_id=USER,
+            project_id=PROJECT,
+        )
--- a/packages/core/tests/test_outline_write_repo.py
+++ b/packages/core/tests/test_outline_write_repo.py
@@ -0,0 +1,125 @@
+"""T3.5 大纲写侧 repo 单测（ARCH §5.4 / §7.2）。
+
+纯内存 fake，无 DB：断言 upsert 幂等覆盖、按 project_id 隔离、frozen View 不可变、
+`_to_view` 的 beats 包裹/解包 round-trip。`(project_id, chapter_no)` 唯一性是 **DB 级**
+（models UniqueConstraint），纯 fake 不引 pg 断言——真实 DB 路径由 T3.7 E2E 覆盖。
+"""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass, field
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+from ww_core.domain.outline_write_repo import (
+    OutlineWriteRepo,
+    OutlineWriteView,
+    _to_view,
+)
+
+PROJECT = uuid.UUID("00000000-0000-0000-0000-000000000001")
+OTHER = uuid.UUID("00000000-0000-0000-0000-000000000002")
+
+
+@dataclass
+class _Row:
+    """模拟 ORM Outline 行（beats 是 JSONB dict、foreshadow_windows 是 JSONB list）。"""
+
+    project_id: uuid.UUID
+    volume: int
+    chapter_no: int
+    beats: dict[str, Any] | None
+    foreshadow_windows: list[Any] = field(default_factory=list)
+
+
+@dataclass
+class FakeOutlineWriteRepo:
+    """实现 `OutlineWriteRepo` Protocol（内存；幂等覆盖同 (project_id, chapter_no)）。"""
+
+    rows: dict[tuple[uuid.UUID, int], _Row] = field(default_factory=dict)
+
+    async def upsert_chapter(
+        self,
+        project_id: uuid.UUID,
+        *,
+        volume: int,
+        chapter_no: int,
+        beats: list[str],
+        foreshadow_windows: list[dict[str, Any]],
+    ) -> OutlineWriteView:
+        row = self.rows.get((project_id, chapter_no))
+        beats_payload = {"beats": list(beats)}
+        windows = [dict(w) for w in foreshadow_windows]
+        if row is None:
+            row = _Row(
+                project_id=project_id,
+                volume=volume,
+                chapter_no=chapter_no,
+                beats=beats_payload,
+                foreshadow_windows=windows,
+            )
+        else:
+            row.volume = volume
+            row.beats = beats_payload
+            row.foreshadow_windows = windows
+        self.rows[(project_id, chapter_no)] = row
+        return _to_view(row)  # type: ignore[arg-type]
+
+
+def _repo() -> OutlineWriteRepo:
+    return FakeOutlineWriteRepo()
+
+
+@pytest.mark.asyncio
+async def test_upsert_inserts_new_chapter() -> None:
+    repo = _repo()
+    view = await repo.upsert_chapter(
+        PROJECT,
+        volume=1,
+        chapter_no=1,
+        beats=["a", "b"],
+        foreshadow_windows=[{"code": "F1", "expected_close_to": 10}],
+    )
+    assert view.chapter_no == 1
+    assert view.volume == 1
+    assert view.beats == ["a", "b"]
+    assert view.foreshadow_windows[0]["code"] == "F1"
+
+
+@pytest.mark.asyncio
+async def test_upsert_overwrites_same_chapter() -> None:
+    repo = _repo()
+    await repo.upsert_chapter(PROJECT, volume=1, chapter_no=1, beats=["old"], foreshadow_windows=[])
+    view = await repo.upsert_chapter(
+        PROJECT, volume=2, chapter_no=1, beats=["new"], foreshadow_windows=[]
+    )
+    assert view.beats == ["new"]
+    assert view.volume == 2
+
+
+@pytest.mark.asyncio
+async def test_upsert_isolated_by_project() -> None:
+    repo = FakeOutlineWriteRepo()
+    await repo.upsert_chapter(PROJECT, volume=1, chapter_no=1, beats=["p"], foreshadow_windows=[])
+    await repo.upsert_chapter(OTHER, volume=1, chapter_no=1, beats=["o"], foreshadow_windows=[])
+    assert len(repo.rows) == 2
+
+
+def test_to_view_unpacks_wrapped_beats() -> None:
+    row = _Row(project_id=PROJECT, volume=1, chapter_no=3, beats={"beats": ["x", "y"]})
+    view = _to_view(row)  # type: ignore[arg-type]
+    assert view.beats == ["x", "y"]
+
+
+def test_to_view_handles_null_beats() -> None:
+    row = _Row(project_id=PROJECT, volume=1, chapter_no=3, beats=None)
+    view = _to_view(row)  # type: ignore[arg-type]
+    assert view.beats == []
+
+
+def test_write_view_is_frozen() -> None:
+    view = OutlineWriteView(chapter_no=1, volume=1, beats=["a"])
+    with pytest.raises(ValidationError):
+        view.volume = 2
--- a/packages/core/tests/test_review.py
+++ b/packages/core/tests/test_review.py
@@ -13,16 +13,31 @@ 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, continuity_spec
+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,
@@ -33,6 +48,8 @@ from ww_core.orchestrator import (
    build_review_request,
    collect_reviews,
    extract_conflicts,
+    extract_foreshadow_sug,
+    extract_pace,
    make_review_node,
    normalize_review,
    run_review,
@@ -242,3 +259,177 @@ async def test_normalize_review_maps_unexpected_error_to_error_event() -> None:
    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
--- a/packages/core/ww_core/domain/init.py
+++ b/packages/core/ww_core/domain/init.py
@@ -9,6 +9,27 @@ from ww_core.domain.chapter_repo import (
    SqlChapterRepo,
 )
 from ww_core.domain.digest_repo import DigestAppendRepo, SqlDigestAppendRepo
+from ww_core.domain.foreshadow_repo import (
+    ForeshadowLedgerRepo,
+    ForeshadowLedgerView,
+    SqlForeshadowLedgerRepo,
+)
+from ww_core.domain.foreshadow_state import (
+    CLOSED,
+    OPEN,
+    OVERDUE,
+    PARTIAL,
+    ForeshadowStatus,
+    InvalidTransition,
+    apply_overdue_scan,
+    is_overdue,
+    transition,
+)
+from ww_core.domain.outline_write_repo import (
+    OutlineWriteRepo,
+    OutlineWriteView,
+    SqlOutlineWriteRepo,
+)
 from ww_core.domain.project_repo import (
    ProjectCreate,
    ProjectRepo,
@@ -26,7 +47,22 @@ __all__ = [
    "DigestAppendRepo",
    "SqlDigestAppendRepo",
    "DigestView",
+    "ForeshadowLedgerRepo",
+    "ForeshadowLedgerView",
+    "SqlForeshadowLedgerRepo",
+    "ForeshadowStatus",
+    "InvalidTransition",
+    "OPEN",
+    "PARTIAL",
+    "CLOSED",
+    "OVERDUE",
+    "apply_overdue_scan",
+    "is_overdue",
+    "transition",
    "MemoryRepos",
+    "OutlineWriteRepo",
+    "OutlineWriteView",
+    "SqlOutlineWriteRepo",
    "ProjectCreate",
    "ProjectRepo",
    "ProjectView",
--- a/packages/core/ww_core/domain/foreshadow_repo.py
+++ b/packages/core/ww_core/domain/foreshadow_repo.py
@@ -0,0 +1,219 @@
+"""伏笔账本**写侧** Repository（登记/状态机/进展/到期扫描；ARCH §6.2 / PRODUCT_SPEC §4.2）。
+
+读侧（`list_for_codes`，供 assemble 注入伏笔窗口）已在 `ww_core.memory.sql_repositories`
+ `domain.repositories.ForeshadowRepo`/`ForeshadowView` 提供（最小 View，C5 稳定，不动）。
+本模块加**写**能力 + 看板查询，命名加 `Ledger` 前缀避免与读侧同名歧义
+（同 T2.3 `DigestAppendRepo` 区别于读侧 `DigestRepo` 的先例，见 memory/decisions）。
+`ForeshadowLedgerView` 比读侧 View 多 `importance/links/progress`（看板/账本需要）。
+
+**提交边界**：所有写方法**只 `flush()` 不 `commit()`**——提交交调用方：
+- T3.2 验收后扫描挂 BackgroundTask / 接 §5.5 `TODO(M3)`，由那条事务/任务提交；
+- T3.5 登记/状态端点由端点事务提交。
+与 M2 验收-side repos 一致（写库副作用归编排/事务层，见 memory/gotchas）。
+含 nullable 列的唯一约束 `(project_id, code)` 不用 PG `ON CONFLICT`——`register` 仅
+显式 INSERT（重复 code 让 DB 唯一约束/调用方处理），不做 upsert（见 gotcha）。
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any, Protocol
+
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from ww_db.models import Foreshadow
+
+from ww_core.domain.foreshadow_state import (
+    ForeshadowStatus,
+    apply_overdue_scan,
+)
+from ww_core.domain.foreshadow_state import (
+    transition as apply_transition,
+)
+
+
+class ForeshadowLedgerView(BaseModel):
+    """伏笔账本只读快照（snake_case，frozen）——看板/账本用，含进展与重要度。"""
+
+    model_config = {"frozen": True}
+
+    code: str
+    title: str
+    status: str
+    planted_at: int | None = None
+    content: str | None = None
+    expected_close_from: int | None = None
+    expected_close_to: int | None = None
+    importance: str | None = None
+    links: list[dict[str, Any]] = Field(default_factory=list)
+    progress: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class ForeshadowLedgerRepo(Protocol):
+    """伏笔账本写侧接口（按 project_id 隔离；写方法只 flush 不 commit）。"""
+
+    async def register(
+        self,
+        project_id: uuid.UUID,
+        *,
+        code: str,
+        title: str,
+        planted_at: int | None = None,
+        content: str | None = None,
+        expected_close_from: int | None = None,
+        expected_close_to: int | None = None,
+        importance: str | None = None,
+    ) -> ForeshadowLedgerView:
+        """登记一条伏笔（status=OPEN）；`(project_id, code)` 唯一。"""
+        ...
+
+    async def get(self, project_id: uuid.UUID, code: str) -> ForeshadowLedgerView | None:
+        """按 code 取单条，无则 None。"""
+        ...
+
+    async def list_by_status(
+        self, project_id: uuid.UUID, status: str | None = None
+    ) -> list[ForeshadowLedgerView]:
+        """看板查询：按状态过滤（None=全部）。供 T3.5 `GET /foreshadow?status=`。"""
+        ...
+
+    async def transition(
+        self, project_id: uuid.UUID, code: str, *, to_status: str
+    ) -> ForeshadowLedgerView:
+        """应用状态机校验后写 `status`（非法转移抛 `InvalidTransition`）。"""
+        ...
+
+    async def record_progress(
+        self, project_id: uuid.UUID, code: str, *, entry: dict[str, Any]
+    ) -> ForeshadowLedgerView:
+        """append 一条进展到 `progress` JSONB（append-only，不覆盖历史）。"""
+        ...
+
+    async def scan_overdue(
+        self, project_id: uuid.UUID, *, current_chapter: int
+    ) -> list[ForeshadowLedgerView]:
+        """到期扫描：命中行置 OVERDUE，返回**被改的行**。供 T3.2 验收后扫描。"""
+        ...
+
+
+def _to_view(row: Foreshadow) -> ForeshadowLedgerView:
+    return ForeshadowLedgerView(
+        code=row.code,
+        title=row.title,
+        status=row.status,
+        planted_at=row.planted_at,
+        content=row.content,
+        expected_close_from=row.expected_close_from,
+        expected_close_to=row.expected_close_to,
+        importance=row.importance,
+        links=list(row.links or []),
+        progress=list(row.progress or []),
+    )
+
+
+class SqlForeshadowLedgerRepo:
+    """SQLAlchemy 实现：登记 + 状态机转移 + 进展 append + 到期扫描（只 flush 不 commit）。"""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._s = session
+
+    async def _find(self, project_id: uuid.UUID, code: str) -> Foreshadow | None:
+        return (
+            await self._s.execute(
+                select(Foreshadow).where(
+                    Foreshadow.project_id == project_id,
+                    Foreshadow.code == code,
+                )
+            )
+        ).scalar_one_or_none()
+
+    async def register(
+        self,
+        project_id: uuid.UUID,
+        *,
+        code: str,
+        title: str,
+        planted_at: int | None = None,
+        content: str | None = None,
+        expected_close_from: int | None = None,
+        expected_close_to: int | None = None,
+        importance: str | None = None,
+    ) -> ForeshadowLedgerView:
+        # 仅 INSERT；唯一约束 (project_id, code) 由 DB 强制（不做 ON CONFLICT upsert，见模块注释）。
+        row = Foreshadow(
+            project_id=project_id,
+            code=code,
+            title=title,
+            status=ForeshadowStatus.OPEN.value,
+            planted_at=planted_at,
+            content=content,
+            expected_close_from=expected_close_from,
+            expected_close_to=expected_close_to,
+            importance=importance,
+            links=[],
+            progress=[],
+        )
+        self._s.add(row)
+        await self._s.flush()
+        await self._s.refresh(row)
+        return _to_view(row)
+
+    async def get(self, project_id: uuid.UUID, code: str) -> ForeshadowLedgerView | None:
+        row = await self._find(project_id, code)
+        return _to_view(row) if row is not None else None
+
+    async def list_by_status(
+        self, project_id: uuid.UUID, status: str | None = None
+    ) -> list[ForeshadowLedgerView]:
+        stmt = select(Foreshadow).where(Foreshadow.project_id == project_id)
+        if status is not None:
+            stmt = stmt.where(Foreshadow.status == status)
+        rows = (await self._s.execute(stmt.order_by(Foreshadow.code))).scalars()
+        return [_to_view(r) for r in rows]
+
+    async def transition(
+        self, project_id: uuid.UUID, code: str, *, to_status: str
+    ) -> ForeshadowLedgerView:
+        row = await self._find(project_id, code)
+        if row is None:
+            raise LookupError(f"foreshadow not found: {code}")
+        # 状态机校验（非法转移抛 InvalidTransition，事务回滚由调用方控制）。
+        row.status = apply_transition(
+            ForeshadowStatus(row.status), ForeshadowStatus(to_status)
+        ).value
+        await self._s.flush()
+        await self._s.refresh(row)
+        return _to_view(row)
+
+    async def record_progress(
+        self, project_id: uuid.UUID, code: str, *, entry: dict[str, Any]
+    ) -> ForeshadowLedgerView:
+        row = await self._find(project_id, code)
+        if row is None:
+            raise LookupError(f"foreshadow not found: {code}")
+        # 不可变 append：建新 list 重赋值，确保 ORM 侦测 JSONB 变更（避免原地 mutate 不脏标记）。
+        row.progress = [*(row.progress or []), dict(entry)]
+        await self._s.flush()
+        await self._s.refresh(row)
+        return _to_view(row)
+
+    async def scan_overdue(
+        self, project_id: uuid.UUID, *, current_chapter: int
+    ) -> list[ForeshadowLedgerView]:
+        rows = (
+            await self._s.execute(select(Foreshadow).where(Foreshadow.project_id == project_id))
+        ).scalars()
+        changed: list[ForeshadowLedgerView] = []
+        for row in rows:
+            new = apply_overdue_scan(
+                ForeshadowStatus(row.status),
+                current_chapter=current_chapter,
+                expected_close_to=row.expected_close_to,
+            ).value
+            if new != row.status:
+                row.status = new
+                changed.append(_to_view(row))
+        if changed:
+            await self._s.flush()
+        return changed
--- a/packages/core/ww_core/domain/foreshadow_state.py
+++ b/packages/core/ww_core/domain/foreshadow_state.py
@@ -0,0 +1,105 @@
+"""伏笔账本状态机——确定性纯函数（ARCH §6.2 / PRODUCT_SPEC §4.2）。
+
+状态机（§6.2）::
+
+            登记            首次照应          完成回收
+      ─────▶ OPEN ──────▶ PARTIAL ─────────▶ CLOSED
+               │             │
+               └─────────────┴──▶ OVERDUE  (章号 > expected_close_to 且未 CLOSED)
+                                  (到期扫描置位; 回收后可转 CLOSED)
+
+- 登记 → OPEN（DB server_default）。
+- 首次照应 → PARTIAL。
+- 完成回收 → CLOSED（可从 OPEN/PARTIAL/OVERDUE 一步到达）。
+- 到期判定（M3-d 纯函数，非 Agent）：
+  `current_chapter > expected_close_to AND status≠CLOSED → OVERDUE`。
+- **CLOSED 终态**：不被 OVERDUE 覆盖、不可重开（非法转移显式抛 `InvalidTransition`）。
+
+全部为纯函数：不读 DB、不可变、可单测。写库由 `foreshadow_repo` 落地（只 flush）。
+枚举值与 DB `foreshadow.status` 文本严格一致（`StrEnum` → 直接当字符串用）。
+"""
+
+from __future__ import annotations
+
+from enum import StrEnum
+
+
+class ForeshadowStatus(StrEnum):
+    """伏笔状态——值与 DB `status` 文本一致。"""
+
+    OPEN = "OPEN"
+    PARTIAL = "PARTIAL"
+    CLOSED = "CLOSED"
+    OVERDUE = "OVERDUE"
+
+
+# 便捷常量（供 repo / 测试 / 调用方按名引用，避免散落字面量）。
+OPEN = ForeshadowStatus.OPEN
+PARTIAL = ForeshadowStatus.PARTIAL
+CLOSED = ForeshadowStatus.CLOSED
+OVERDUE = ForeshadowStatus.OVERDUE
+
+
+class InvalidTransition(ValueError):
+    """非法状态转移（如 CLOSED 重开）。"""
+
+
+# 合法转移图：from → 允许到达的状态集合（不含同态幂等，单独放行）。
+_ALLOWED: dict[ForeshadowStatus, frozenset[ForeshadowStatus]] = {
+    OPEN: frozenset({PARTIAL, CLOSED, OVERDUE}),
+    PARTIAL: frozenset({CLOSED, OVERDUE}),
+    OVERDUE: frozenset({PARTIAL, CLOSED}),  # 逾期后仍可（部分）回收
+    CLOSED: frozenset(),  # 终态：不可再转移
+}
+
+
+def transition(current: ForeshadowStatus, to_status: ForeshadowStatus) -> ForeshadowStatus:
+    """应用状态机校验后返回新状态（纯函数）。
+
+    - 同态（current == to_status）幂等放行。
+    - CLOSED 为终态：任何离开 CLOSED 的转移都非法。
+    - 其余按 `_ALLOWED` 校验；非法 → `InvalidTransition`。
+    """
+    if current == to_status:
+        return to_status
+    if to_status not in _ALLOWED[current]:
+        raise InvalidTransition(f"非法伏笔转移: {current.value} → {to_status.value}")
+    return to_status
+
+
+def is_overdue(
+    *,
+    current_chapter: int,
+    expected_close_to: int | None,
+    status: ForeshadowStatus,
+) -> bool:
+    """到期判据（M3-d，§6.2）：`current_chapter > expected_close_to AND status≠CLOSED`。
+
+    - 无 `expected_close_to`（未设回收窗口）→ 永不逾期。
+    - CLOSED → 永不逾期（已回收）。
+    - `current_chapter == expected_close_to` → 仍在窗口内，未逾期（严格大于才算）。
+    """
+    if expected_close_to is None:
+        return False
+    if status == CLOSED:
+        return False
+    return current_chapter > expected_close_to
+
+
+def apply_overdue_scan(
+    status: ForeshadowStatus,
+    *,
+    current_chapter: int,
+    expected_close_to: int | None,
+) -> ForeshadowStatus:
+    """到期扫描应用（验收后触发，§6.2）：命中判据 → 置 `OVERDUE`，否则保持原状态。
+
+    幂等：已 OVERDUE 且仍逾期 → 保持 OVERDUE；CLOSED 永不被改写（is_overdue 已排除）。
+    """
+    if is_overdue(
+        current_chapter=current_chapter,
+        expected_close_to=expected_close_to,
+        status=status,
+    ):
+        return OVERDUE
+    return status
--- a/packages/core/ww_core/domain/outline_write_repo.py
+++ b/packages/core/ww_core/domain/outline_write_repo.py
@@ -0,0 +1,109 @@
+"""大纲**写侧** Repository（逐章 upsert；ARCH §5.4 outliner writes=outline / §7.2 POST /outline）。
+
+读侧（`get`，供 assemble 注入本章 beats）已在 `ww_core.memory.sql_repositories.SqlOutlineRepo`
+ `domain.repositories.OutlineRepo`/`OutlineView` 提供（C5 稳定，不动）。本模块加**写**能力，
+命名加 `Write` 前缀避免与读侧 `OutlineRepo` 同名歧义（同 `DigestAppendRepo`/`ForeshadowLedgerRepo`
+区别于读侧的先例，见 memory/decisions）。
+
+**提交边界**：`upsert_chapter` 只 `flush()` 不 `commit()`——提交交端点事务（T3.5 端点末尾
+一次 `commit()`，与网关 ledger 一并落库；与 M2 验收-side repos 一致，见 memory/gotchas）。
+
+唯一约束 `(project_id, chapter_no)`：用**显式 read-modify-write**（按 chapter_no 查→有则改、
+无则插），不用 PG `ON CONFLICT`（保持与项目其它 upsert 路径一致、可纯 fake 单测，不引 pg 依赖）。
+
+**beats 存储形**：DB 列 `outline.beats` 是 JSONB `dict`，而 outliner 产 `list[str]`——
+故包裹成 `{"beats": [...]}` 落库，读侧 `OutlineView.beats` 取到同形 dict（round-trip 一致）。
+`foreshadow_windows` 是 JSONB list，直接落 `[w.model_dump() for w in chapter.foreshadow_windows]`。
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any, Protocol
+
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from ww_db.models import Outline
+
+
+class OutlineWriteView(BaseModel):
+    """单章大纲写入后的只读快照（snake_case，frozen）。"""
+
+    model_config = {"frozen": True}
+
+    chapter_no: int
+    volume: int
+    beats: list[str] = Field(default_factory=list)
+    foreshadow_windows: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class OutlineWriteRepo(Protocol):
+    """大纲写侧接口（按 project_id 隔离；唯一 `(project_id, chapter_no)`；只 flush 不 commit）。"""
+
+    async def upsert_chapter(
+        self,
+        project_id: uuid.UUID,
+        *,
+        volume: int,
+        chapter_no: int,
+        beats: list[str],
+        foreshadow_windows: list[dict[str, Any]],
+    ) -> OutlineWriteView:
+        """逐章 upsert：同 `(project_id, chapter_no)` 覆盖，否则插入。只 flush。"""
+        ...
+
+
+def _to_view(row: Outline) -> OutlineWriteView:
+    raw = row.beats or {}
+    beats = list(raw.get("beats", [])) if isinstance(raw, dict) else []
+    return OutlineWriteView(
+        chapter_no=row.chapter_no,
+        volume=row.volume,
+        beats=beats,
+        foreshadow_windows=list(row.foreshadow_windows or []),
+    )
+
+
+class SqlOutlineWriteRepo:
+    """SQLAlchemy 实现：显式 read-modify-write upsert（只 flush 不 commit）。"""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._s = session
+
+    async def upsert_chapter(
+        self,
+        project_id: uuid.UUID,
+        *,
+        volume: int,
+        chapter_no: int,
+        beats: list[str],
+        foreshadow_windows: list[dict[str, Any]],
+    ) -> OutlineWriteView:
+        row = (
+            await self._s.execute(
+                select(Outline).where(
+                    Outline.project_id == project_id,
+                    Outline.chapter_no == chapter_no,
+                )
+            )
+        ).scalar_one_or_none()
+        # JSONB 写入建新对象重赋值（避免原地 mutate 不被 ORM 侦测，见 gotcha）。
+        beats_payload = {"beats": list(beats)}
+        windows_payload = [dict(w) for w in foreshadow_windows]
+        if row is None:
+            row = Outline(
+                project_id=project_id,
+                volume=volume,
+                chapter_no=chapter_no,
+                beats=beats_payload,
+                foreshadow_windows=windows_payload,
+            )
+            self._s.add(row)
+        else:
+            row.volume = volume
+            row.beats = beats_payload
+            row.foreshadow_windows = windows_payload
+        await self._s.flush()
+        await self._s.refresh(row)
+        return _to_view(row)
--- a/packages/core/ww_core/orchestrator/init.py
+++ b/packages/core/ww_core/orchestrator/init.py
@@ -10,9 +10,13 @@ from __future__ import annotations

 from .collect import (
    CONTINUITY,
+    FORESHADOW,
+    PACE,
    ReviewRecorder,
    collect_reviews,
    extract_conflicts,
+    extract_foreshadow_sug,
+    extract_pace,
 )
 from .graph import (
    COLLECT_NODE,
@@ -22,6 +26,7 @@ from .graph import (
    build_write_graph,
    setup_checkpointer,
 )
+from .outline_node import build_outline_request, run_outline
 from .review_node import (
    REVIEW_INCOMPLETE,
    REVIEW_OK,
@@ -36,6 +41,8 @@ from .sse import (
    EVENT_CONFLICT,
    EVENT_DONE,
    EVENT_ERROR,
+    EVENT_FORESHADOW,
+    EVENT_PACE,
    EVENT_SECTION,
    EVENT_TOKEN,
    SECTION_DONE,
@@ -45,8 +52,10 @@ from .sse import (
    conflict_event,
    done_event,
    error_event,
+    foreshadow_event,
    normalize_deltas,
    normalize_review,
+    pace_event,
    section_event,
    token_event,
 )
@@ -59,8 +68,12 @@ __all__ = [
    "EVENT_CONFLICT",
    "EVENT_DONE",
    "EVENT_ERROR",
+    "EVENT_FORESHADOW",
+    "EVENT_PACE",
    "EVENT_SECTION",
    "EVENT_TOKEN",
+    "FORESHADOW",
+    "PACE",
    "REVIEW_INCOMPLETE",
    "REVIEW_OK",
    "REVIEW_SPECS",
@@ -74,6 +87,7 @@ __all__ = [
    "GatewayStream",
    "ReviewRecorder",
    "SseEvent",
+    "build_outline_request",
    "build_review_context",
    "build_review_graph",
    "build_review_request",
@@ -84,10 +98,15 @@ __all__ = [
    "done_event",
    "error_event",
    "extract_conflicts",
+    "extract_foreshadow_sug",
+    "extract_pace",
+    "foreshadow_event",
    "make_review_node",
    "merge_reviews",
    "normalize_deltas",
    "normalize_review",
+    "pace_event",
+    "run_outline",
    "run_review",
    "section_event",
    "setup_checkpointer",
--- a/packages/core/ww_core/orchestrator/collect.py
+++ b/packages/core/ww_core/orchestrator/collect.py
@@ -1,17 +1,21 @@
 """collect 节点（C4 扩 / ARCH §5.2 collect 行）——汇总并行审 → 落 `chapter_reviews` 留痕。

-并行四审各把 `{spec.name: {status, result}}` 合并进 `state["reviews"]`（reducer，见 state.py）；
-collect 把它们汇总，调 `review_repo.record(...)` append 一行 `chapter_reviews`（真相源留痕）。
+并行审各把 `{spec.name: {status, result}}` 合并进 `state["reviews"]`（reducer，见 state.py）；
+collect 把它们汇总，**一次** `review_repo.record(...)` append 一行 `chapter_reviews`（真相源留痕），
+按 spec.name 把各审结果映射到对应列。

 不变量 #3：审稿节点只读不写库；**唯一写入点是这里经 `review_repo`**（仍是「留痕」非「生效」，
-AI 产出真正入库经 T2.4 验收事务的裁决）。
+AI 产出真正入库经验收事务的裁决）。

 记账边界（gotcha「网关 ledger 只 flush」）：审稿经网关产 usage（网关内 `flush` 进同一 session）；
-collect 与 review_repo 同样**只 flush 不 commit**——`commit` 归 HTTP 端点（T2.5）/验收事务（T2.4），
-与 draft 端点一致（端点在流耗尽后 `await session.commit()`）。本节点绝不 commit。
+collect 与 review_repo 同样**只 flush 不 commit**——`commit` 归 HTTP 端点 / 验收事务。
+本节点绝不 commit。

-M2 只接 continuity：从 `reviews["continuity"].result.conflicts` 抽冲突落 `conflicts` 列；
-foreshadow/style/pace 列留空（M3/M4 接入后填）。设计成按 spec 名取分项，便于扩展。
+M3 三审齐 → 列映射（一次 record 落齐本章三审）：
+- continuity → `conflicts`（冲突清单 list）；
+- foreshadow → `foreshadow_sug`（planted/resolved 扁平为建议 list，每条带 `kind` 标记）；
+- pace       → `pace`（节奏诊断 dict：water/hook/beat_map）。
+style 第四审留 M4。任一审 `incomplete`（网关失败隔离，§5.2）→ 该列留空/None，不阻塞其余。
 """

 from __future__ import annotations
@@ -27,6 +31,12 @@ from .state import ChapterState
 log = structlog.get_logger(__name__)

 CONTINUITY = "continuity"
+FORESHADOW = "foreshadow"
+PACE = "pace"
+
+# foreshadow 建议扁平化时的来源标记（planted=新埋 / resolved=回收）。
+FORESHADOW_KIND_PLANTED = "planted"
+FORESHADOW_KIND_RESOLVED = "resolved"


 class ReviewRecorder(Protocol):
@@ -49,17 +59,53 @@ class ReviewRecorder(Protocol):
    ) -> Any: ...


+def _ok_result(reviews: dict[str, Any], name: str) -> dict[str, Any] | None:
+    """取某审的结果 dict，仅当 `ok` 且有结果；否则 None（incomplete/缺席）。"""
+    entry = reviews.get(name)
+    if not entry or entry.get("status") != REVIEW_OK:
+        return None
+    result = entry.get("result")
+    return result if isinstance(result, dict) else None
+
+
 def extract_conflicts(reviews: dict[str, Any]) -> list[dict[str, Any]]:
    """从 continuity 分项抽冲突清单（纯函数）。

    仅当该审 `ok` 且有结果时取其 `conflicts`；未完成/缺席 → 空列表（不臆造）。
    """
-    entry = reviews.get(CONTINUITY)
-    if not entry or entry.get("status") != REVIEW_OK:
+    result = _ok_result(reviews, CONTINUITY)
+    if result is None:
        return []
-    result = entry.get("result") or {}
-    conflicts = result.get("conflicts") or []
-    return [dict(c) for c in conflicts]
+    return [dict(c) for c in result.get("conflicts") or []]
+
+
+def extract_foreshadow_sug(reviews: dict[str, Any]) -> list[dict[str, Any]]:
+    """从 foreshadow 分项扁平化建议清单（纯函数）。
+
+    `ForeshadowReview{planted,resolved}` → 单个 list，每条加 `kind` 标记来源
+    （`foreshadow_sug` 列是 JSONB list）。未完成/缺席 → 空列表。
+    """
+    result = _ok_result(reviews, FORESHADOW)
+    if result is None:
+        return []
+    out: list[dict[str, Any]] = []
+    for sug in result.get("planted") or []:
+        out.append({**dict(sug), "kind": FORESHADOW_KIND_PLANTED})
+    for sug in result.get("resolved") or []:
+        out.append({**dict(sug), "kind": FORESHADOW_KIND_RESOLVED})
+    return out
+
+
+def extract_pace(reviews: dict[str, Any]) -> dict[str, Any] | None:
+    """从 pace 分项取节奏诊断 dict（纯函数）。
+
+    `PaceReview{water,hook,beat_map}` 整体入 `pace` 列（JSONB dict）。
+    未完成/缺席 → None（列留空，§5.2 不阻塞其余）。
+    """
+    result = _ok_result(reviews, PACE)
+    if result is None:
+        return None
+    return dict(result)


 async def collect_reviews(
@@ -67,25 +113,32 @@ async def collect_reviews(
    *,
    review_repo: ReviewRecorder,
 ) -> dict[str, Any]:
-    """collect 节点：汇总并行审 → 抽冲突 → `review_repo.record` 留痕（只 flush 不 commit）。
+    """collect 节点：汇总并行审 → 各列映射 → 一次 `review_repo.record` 留痕（只 flush 不 commit）。

    直接调用本函数（注入内存 fake repo）即可单测，无需图运行时。
-    `chapter_version=None`：留痕行不绑具体 version（T2.4 验收时再绑/裁决，R3/R4）。
+    `chapter_version=None`：留痕行不绑具体 version（验收时再绑/裁决，R3/R4）。
+    某审 incomplete → 该列空/None，不阻塞其余审落库（§5.2）。
    返回空增量字典——审稿真相已在表，state 不作真相源（不变量 #5）。
    """
    reviews = state.get("reviews") or {}
    conflicts = extract_conflicts(reviews)
+    foreshadow_sug = extract_foreshadow_sug(reviews)
+    pace = extract_pace(reviews)
    await review_repo.record(
        state["project_id"],
        state["chapter_no"],
        chapter_version=None,
        conflicts=conflicts,
+        foreshadow_sug=foreshadow_sug,
+        pace=pace,
    )
    log.info(
        "collect_reviews_recorded",
        project_id=str(state["project_id"]),
        chapter_no=state["chapter_no"],
        conflict_count=len(conflicts),
+        foreshadow_sug_count=len(foreshadow_sug),
+        has_pace=pace is not None,
        reviews=sorted(reviews.keys()),
    )
    # 留痕已在表（真相源）；不在节点 commit（端点/事务层负责，见模块 docstring）。
--- a/packages/core/ww_core/orchestrator/graph.py
+++ b/packages/core/ww_core/orchestrator/graph.py
@@ -15,7 +15,12 @@ from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any

 from langgraph.graph import END, START, StateGraph
-from ww_agents import AgentSpec, continuity_spec
+from ww_agents import (
+    AgentSpec,
+    continuity_spec,
+    foreshadow_spec,
+    pace_spec,
+)

 from .collect import ReviewRecorder, collect_reviews
 from .review_node import GatewayRun, run_review
@@ -51,7 +56,8 @@ def build_write_graph(
    return builder.compile(checkpointer=checkpointer)


-REVIEW_SPECS: tuple[AgentSpec, ...] = (continuity_spec,)
+# M3：三审齐（continuity + foreshadow + pace）。文风第四审属 M4/T4.2。
+REVIEW_SPECS: tuple[AgentSpec, ...] = (continuity_spec, foreshadow_spec, pace_spec)


 def build_review_graph(
@@ -63,9 +69,10 @@ def build_review_graph(
 ) -> CompiledStateGraph[ChapterState, None, ChapterState, ChapterState]:
    """构建并编译审稿子图：`START →`（各 review spec 并行节点）`→ collect → END`。

-    M2 默认仅 continuity；`review_specs` 可扩（M3/M4 加 foreshadow/style/pace 三审），
-    图按这组 spec 循环加并行分支，全部汇入 collect（任一审失败不阻塞，§5.2——失败隔离
-    在 `run_review` 内、标 incomplete 占位）。
+    M3 默认三审齐（continuity + foreshadow + pace）；`review_specs` 仍可扩（M4 加 style
+    第四审），图按这组 spec 循环加并行分支，全部汇入 collect（任一审失败不阻塞，§5.2——
+    失败隔离在 `run_review` 内、标 incomplete 占位）。`run_review`/`make_review_node` 对 spec
+    泛型，三审天然支持（按 `spec.output_schema` 产 parsed，无 spec-specific 假设）。

    `gateway` / `review_repo` 经闭包绑进节点（langgraph 不收 functools.partial，见 gotcha）。
    **accept 不在本图**：本任务只交付到 collect（审稿留痕落 `chapter_reviews`）+ review SSE；
--- a/packages/core/ww_core/orchestrator/outline_node.py
+++ b/packages/core/ww_core/orchestrator/outline_node.py
@@ -0,0 +1,81 @@
+"""大纲节点（C6 扩 / ARCH §5.4 outliner 行 / §6.2 伏笔窗口）。
+
+大纲是**独立生成**——不在写章/审稿流水线（不接进 review 图）。端点（T3.5）直接调
+`run_outline` 拿结构化 `OutlineResult`，再持久化到 `outline` 表（含 foreshadow_windows）。
+
+确定性边界（CLAUDE.md「LangGraph」纪律）：
+- 节点逻辑里**无 LLM 非确定性**——不确定性藏在网关后；节点只做
+  `AgentSpec + context → LlmRequest` 的纯构造 + 转发 `Gateway.run` 的 `parsed`。
+- 因此注入 mock 网关（产 `parsed`）即可单测，无需图运行时、无需真 Postgres。
+
+不变量 #2：agent 只声明 `tier`，绝不传具体 model（spec.tier 透传）。
+不变量 #3：节点**只读、不写库**——结构化大纲返回调用方，落库经端点/验收（T3.5）。
+不变量 #9：`spec.system_prompt` 进缓存断点前块（cache=True）；注入材料进 `input`（断点后）。
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Protocol
+
+import structlog
+from ww_agents import AgentSpec, OutlineResult
+from ww_llm_gateway.types import Block, LlmRequest, LlmResponse, Scope
+
+log = structlog.get_logger(__name__)
+
+
+class GatewayRun(Protocol):
+    """大纲节点对网关的最小依赖——只需 `run`（注入真网关或 mock）。"""
+
+    async def run(self, req: LlmRequest) -> LlmResponse: ...
+
+
+def build_outline_request(
+    spec: AgentSpec,
+    *,
+    context: str,
+    user_id: uuid.UUID,
+    project_id: uuid.UUID,
+) -> LlmRequest:
+    """据 spec + 注入材料构造大纲请求（纯函数）。
+
+    `spec.system_prompt` → `system` 缓存断点前块（cache=True，不变量 #9）；
+    `context`（设定/伏笔/人物/世界观的序列化文本）→ `input`（断点后）。
+    排大纲用 `run()` 非 `stream()`（要结构化产物，非流式正文）。
+    """
+    return LlmRequest(
+        tier=spec.tier,  # 不变量 #2：只传档位，不传 model
+        system=[Block(text=spec.system_prompt, cache=True)],
+        input=context,
+        output_schema=spec.output_schema,
+        scope=Scope(user_id=user_id, project_id=project_id),
+    )
+
+
+async def run_outline(
+    spec: AgentSpec,
+    *,
+    context: str,
+    gateway: GatewayRun,
+    user_id: uuid.UUID,
+    project_id: uuid.UUID,
+) -> OutlineResult:
+    """跑大纲生成：构请求 → `gateway.run` → 返回结构化 `OutlineResult`。
+
+    裸函数（显式 gateway 关键字），注入 mock 网关即可单测，无需图运行时。
+    大纲是独立生成（非并行审）：网关失败直接上抛（端点/T3.5 处理），不静默吞。
+    **只产结构化大纲、不写库**（持久化属 T3.5，不变量 #3）。
+    """
+    req = build_outline_request(spec, context=context, user_id=user_id, project_id=project_id)
+    resp = await gateway.run(req)
+    parsed = resp.parsed
+    if not isinstance(parsed, OutlineResult):
+        # 带 output_schema 时 parsed 必非 None（C1）；违约则明确报错而非返回空大纲。
+        log.error(
+            "outline_node_missing_parsed",
+            project_id=str(project_id),
+            parsed_type=type(parsed).__name__,
+        )
+        raise ValueError("gateway returned no parsed OutlineResult for outline request")
+    return parsed
--- a/packages/core/ww_core/orchestrator/sse.py
+++ b/packages/core/ww_core/orchestrator/sse.py
@@ -18,6 +18,7 @@ from pydantic import BaseModel
 from ww_llm_gateway.types import Delta
 from ww_shared import AppError, ErrorCode

+from .collect import CONTINUITY, FORESHADOW, PACE
 from .review_node import REVIEW_OK

 log = structlog.get_logger(__name__)
@@ -25,7 +26,9 @@ log = structlog.get_logger(__name__)
 # §7.3 事件名（全集）。完整清单以 ARCHITECTURE §7.3 为准。
 EVENT_TOKEN = "token"  # 正文增量（写章 draft）
 EVENT_SECTION = "section"  # 四审分项开始/完成（审稿 review）
-EVENT_CONFLICT = "conflict"  # 冲突命中（审稿 review）
+EVENT_CONFLICT = "conflict"  # continuity 冲突命中（审稿 review）
+EVENT_FORESHADOW = "foreshadow"  # foreshadow 建议命中（新埋/回收，审稿 review）
+EVENT_PACE = "pace"  # pace 节奏诊断（注水/钩子/节拍图，审稿 review）
 EVENT_DONE = "done"
 EVENT_ERROR = "error"

@@ -72,6 +75,35 @@ def conflict_event(
    )


+def foreshadow_event(
+    *,
+    kind: str,
+    code: str | None,
+    title: str,
+    where: str | None,
+    note: str | None,
+) -> SseEvent:
+    """伏笔建议事件（审稿 review）——前端伏笔看板联动。
+
+    `kind` ∈ planted（新埋）/ resolved（回收）；作者验收裁决后才登记/改状态（只读建议）。
+    """
+    return SseEvent(
+        event=EVENT_FORESHADOW,
+        data={"kind": kind, "code": code, "title": title, "where": where, "note": note},
+    )
+
+
+def pace_event(*, water: list[dict[str, object]], hook: bool, beat_map: list[int]) -> SseEvent:
+    """节奏诊断事件（审稿 review）——前端节奏报告/节拍图 ▁▃▅ 可视化。
+
+    一次审产一条：注水段清单 + 章末钩子有无 + 逐段节拍强度序列。
+    """
+    return SseEvent(
+        event=EVENT_PACE,
+        data={"water": water, "hook": hook, "beat_map": beat_map},
+    )
+
+
 def error_event(*, code: str, message: str, request_id: str | None = None) -> SseEvent:
    """错误事件，形对齐错误信封 §7.1（`code`/`message`），附 `request_id` 便于贯通排查。"""
    return SseEvent(
@@ -122,40 +154,79 @@ async def normalize_deltas(
    yield done_event(length=total)


+def _section_result_events(name: str, result: dict[str, Any]) -> list[SseEvent]:
+    """把某审 `result` dict 按审种 surface 为结构化事件（纯函数，便于单测）。
+
+    畸形 result（非 dict 等）由调用方 try/except 兜成 `error` 事件；这里只按已知形读。
+    """
+    events: list[SseEvent] = []
+    if name == CONTINUITY:
+        for conflict in result.get("conflicts") or []:
+            events.append(
+                conflict_event(
+                    type=conflict.get("type", ""),
+                    where=conflict.get("where", ""),
+                    refs=list(conflict.get("refs") or []),
+                    suggestion=conflict.get("suggestion", ""),
+                )
+            )
+    elif name == FORESHADOW:
+        for kind in ("planted", "resolved"):
+            for sug in result.get(kind) or []:
+                events.append(
+                    foreshadow_event(
+                        kind=kind,
+                        code=sug.get("code"),
+                        title=sug.get("title", ""),
+                        where=sug.get("where"),
+                        note=sug.get("note"),
+                    )
+                )
+    elif name == PACE:
+        events.append(
+            pace_event(
+                water=[dict(w) for w in result.get("water") or []],
+                hook=bool(result.get("hook", False)),
+                beat_map=[int(b) for b in result.get("beat_map") or []],
+            )
+        )
+    return events
+
+
 async def normalize_review(
    reviews: Mapping[str, Any],
    *,
    request_id: str | None = None,
 ) -> AsyncIterator[SseEvent]:
-    """把审稿子图的结构化产出（`state["reviews"]`）归一为 SSE 事件序列。
+    """把审稿子图的结构化产出（`state["reviews"]`）归一为 SSE 事件序列（M3 三审齐）。

-    供 **T2.5** 的 `POST .../review` 端点消费：跑 `build_review_graph` 后拿 `final["reviews"]`
-    （形 `{name: {status, result}}`），本函数产 `section`（每审 done/incomplete）+ `conflict`
-    （命中冲突，形对齐 C6 `Conflict`）+ `done`。HTTP event-stream 编码归 T2.5，不在此处。
+    供 `POST .../review` 端点消费：跑 `build_review_graph` 后拿 `final["reviews"]`
+    （形 `{name: {status, result}}`），本函数每审先发 `section`（done/incomplete），再按审种
+    surface 结构化结果：
+    - continuity → 每冲突一条 `conflict`（形对齐 C6 `Conflict`）；
+    - foreshadow → 每条建议一条 `foreshadow`（planted/resolved，前端看板联动）；
+    - pace       → 一条 `pace`（注水/钩子/节拍图，前端节奏报告/▁▃▅）。
+    末尾 `done{length=审项数}`。HTTP event-stream 编码归端点，不在此处。

    错误处理纪律（同 `normalize_deltas`）：任何意外 → 一条 `error` 事件后正常收尾，不上抛。
-    某审 `incomplete`（网关失败被隔离，§5.2）→ 发 `section{status:incomplete}`，不发其冲突。
-    `reviews` 键按字典序遍历 → 确定性事件顺序（便于单测/前端稳定渲染）。
+    某审 `incomplete`（网关失败被隔离，§5.2）→ 仅发 `section{status:incomplete}`，不发其结果。
+    `reviews` 键按字典序遍历 → 确定性事件顺序（便于单测/前端稳定渲染）。向后兼容：
+    既有 `section`/`conflict` 形不变，仅新增 `foreshadow`/`pace`。
    """
    try:
        section_count = 0
        for name in sorted(reviews.keys()):
            entry = reviews.get(name) or {}
            status = entry.get("status")
-            if status == REVIEW_OK:
-                yield section_event(name=name, status=SECTION_DONE)
-                section_count += 1
-                result = entry.get("result") or {}
-                for conflict in result.get("conflicts") or []:
-                    yield conflict_event(
-                        type=conflict.get("type", ""),
-                        where=conflict.get("where", ""),
-                        refs=list(conflict.get("refs") or []),
-                        suggestion=conflict.get("suggestion", ""),
-                    )
-            else:
+            if status != REVIEW_OK:
                yield section_event(name=name, status=SECTION_INCOMPLETE)
                section_count += 1
+                continue
+            yield section_event(name=name, status=SECTION_DONE)
+            section_count += 1
+            result = entry.get("result") or {}
+            for event in _section_result_events(name, result):
+                yield event
    except Exception as exc:  # noqa: BLE001 — 边界兜底：任何意外都归一为 error 事件，不泄异常
        log.error("sse_review_unexpected_error", error=str(exc), request_id=request_id)
        yield error_event(
--- a/packages/llm_gateway/tests/test_ledger_concurrency.py
+++ b/packages/llm_gateway/tests/test_ledger_concurrency.py
@@ -0,0 +1,112 @@
+"""T3.8 回归：并行审记账并发安全（add-only，无 flush 重入）。
+
+旧 bug（T3.7 真 pg E2E 暴露）：三审作为同一 LangGraph superstep 的并行分支，各自
+`gateway.run()` → 共用**同一请求 session** 的 `SqlAlchemyLedgerSink.record`
+(`session.add` + `await session.flush()`)。`AsyncSession` 非并发安全：`await flush()`
+是并行路径里唯一让出控制权的 DB-IO，第二/三审的 flush 撞上「Session is already
+flushing」（flush 重入）→ 被 `run_review` 失败隔离吞成 `incomplete` → foreshadow/pace
+结果静默丢失。
+
+本测试用一个会**侦测 flush 重入**的假 AsyncSession：`flush` 异步（`await asyncio.sleep`
+让出控制权），若两协程同时在 flush 中即抛 `RuntimeError("Session is already flushing")`
+——精确复刻 SQLAlchemy 的报错。在旧 add-then-flush 实现下，`asyncio.gather` 三并发
+`record` 会触发该重入；修复后 `record` 只 `add`（同步、不让出），不再调 flush，故三行
+全部入 session 且零重入。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+
+import pytest
+from ww_llm_gateway.ledger import SqlAlchemyLedgerSink
+from ww_llm_gateway.types import Scope, Usage
+
+
+class _FlushReentryDetectingSession:
+    """假 AsyncSession：记录 `add` 的行；`flush` 异步且侦测并发重入。
+
+    `add` 同步追加（镜像真 session）。`flush` 进入时若已有协程在 flush 中 → 抛
+    `RuntimeError("Session is already flushing")`（复刻 SQLAlchemy 行为），否则
+    `await asyncio.sleep(0)` 让出控制权模拟真实 DB-IO 让步点。
+    """
+
+    def __init__(self) -> None:
+        self.added: list[object] = []
+        self.flush_calls = 0
+        self._flushing = False
+
+    def add(self, row: object) -> None:
+        self.added.append(row)
+
+    async def flush(self) -> None:
+        self.flush_calls += 1
+        if self._flushing:
+            raise RuntimeError("Session is already flushing")
+        self._flushing = True
+        try:
+            await asyncio.sleep(0)  # 让出控制权——并行协程在此交错
+        finally:
+            self._flushing = False
+
+
+def _usage(input_tokens: int) -> Usage:
+    return Usage(
+        provider="deepseek",
+        model="deepseek-chat",
+        input_tokens=input_tokens,
+        output_tokens=1,
+        cost_minor=0,
+        currency="USD",
+    )
+
+
+def _scope() -> Scope:
+    return Scope(user_id=uuid.UUID(int=1), project_id=uuid.UUID(int=2))
+
+
+async def test_concurrent_record_adds_all_rows_without_flush_reentry() -> None:
+    """三并发 `record`（模拟三审并行）→ 三行全部入 session，绝不抛 flush 重入。
+
+    旧 add-then-flush 实现在此会因 `await flush()` 让出控制权而 flush 重入抛错；
+    add-only 实现下 `record` 不让出控制权，三行确定性入 session。
+    """
+    session = _FlushReentryDetectingSession()
+    sink = SqlAlchemyLedgerSink(session)  # type: ignore[arg-type]
+
+    await asyncio.gather(
+        sink.record(_scope(), _usage(100)),
+        sink.record(_scope(), _usage(200)),
+        sink.record(_scope(), _usage(300)),
+    )
+
+    # 三审用量全部入 session（一次调用一行，记账不丢）
+    assert len(session.added) == 3
+    input_tokens = sorted(row.input_tokens for row in session.added)  # type: ignore[attr-defined]
+    assert input_tokens == [100, 200, 300]
+    # add-only：sink 不在并行路径 flush（持久化靠端点/事务 commit）
+    assert session.flush_calls == 0
+
+
+async def test_record_does_not_flush_so_caller_commit_persists() -> None:
+    """单次 `record` 只 add 不 flush——契约「网关 ledger 只 add、调用方负责 commit」。"""
+    session = _FlushReentryDetectingSession()
+    sink = SqlAlchemyLedgerSink(session)  # type: ignore[arg-type]
+
+    await sink.record(_scope(), _usage(42))
+
+    assert len(session.added) == 1
+    assert session.flush_calls == 0
+
+
+@pytest.mark.parametrize("n_concurrent", [2, 5, 10])
+async def test_concurrent_record_scales_without_loss(n_concurrent: int) -> None:
+    """N 并发 `record` 全部落 session，无丢失、无 flush 重入（不论并发度）。"""
+    session = _FlushReentryDetectingSession()
+    sink = SqlAlchemyLedgerSink(session)  # type: ignore[arg-type]
+
+    await asyncio.gather(*(sink.record(_scope(), _usage(i + 1)) for i in range(n_concurrent)))
+
+    assert len(session.added) == n_concurrent
+    assert session.flush_calls == 0
--- a/packages/llm_gateway/ww_llm_gateway/ledger.py
+++ b/packages/llm_gateway/ww_llm_gateway/ledger.py
@@ -24,6 +24,19 @@ class SqlAlchemyLedgerSink:
        self._session = session

    async def record(self, scope: Scope, usage: Usage) -> None:
+        """把一次调用追加进 session（**只 add，不 flush/commit**）。
+
+        并发安全（T3.8 bugfix）：三审作为同一 LangGraph superstep 的并行分支
+        共用一个请求 session 的本 sink。`AsyncSession` 非并发安全——若在此
+        `await self._session.flush()`，第二/三审的 flush 会撞上「Session is already
+        flushing」（flush 重入），被 `run_review` 失败隔离吞成 `incomplete`，
+        foreshadow/pace 结果静默丢失。`session.add()` 是同步、无 await、不让出
+        控制权，故并行协程不会在 add 中途交错。
+
+        持久化靠调用方：每个端点/验收事务在流耗尽/事务末尾 `commit()`（commit
+        自动 flush 全部待决行），与「网关 ledger 只 add/flush、端点或事务负责
+        commit」的既有边界一致（见 gotcha「网关 ledger 只 flush、调用方必须 commit」）。
+        """
        row = UsageLedger(
            owner_id=scope.user_id,
            project_id=scope.project_id,
@@ -36,4 +49,3 @@ class SqlAlchemyLedgerSink:
            currency=usage.currency,
        )
        self._session.add(row)
-        await self._session.flush()