perf(chain): review 节点并行扇出五审——复用主图 add-only 记账并发范式（CR-M3-6）

packages/core/tests/test_chain_graph.py

@@ -8,6 +8,7 @@ asyncio_mode=auto。
 
 from __future__ import annotations
 
+import asyncio
 import uuid
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
@@ -522,3 +523,137 @@ async def test_continue_volume_chain_second_chapter_sees_first_chapter_text() ->
     # 第 1 章请求无前文（前一章 0 不存在）→ 降级占位，不报错。
     first_write_input = h["gateway"].write_inputs[0]
     assert "前文正文" in first_write_input
+
+
+# ---- CR-M3-6 review 节点并行扇出 ----
+
+
+class _ConcurrencyReviewGateway:
+    """记录审稿并发峰值的 mock 网关：各审 run 之间 `await` 让出——并行扇出时 active 峰值 > 1。
+
+    串行 for 循环（旧实现）每审跑完才起下一审 → active 恒为 1；并行 gather → 峰值 = 审稿数。
+    审稿按 `output_schema` 回各自空实例 parsed（`by_schema` 覆盖全部审 spec 的 schema）。
+    """
+
+    def __init__(self, *, by_schema: dict[type, BaseModel]) -> None:
+        self._by_schema = by_schema
+        self._active = 0
+        self.peak = 0
+        self.review_calls = 0
+
+    async def run(self, req: LlmRequest) -> LlmResponse:
+        schema = req.output_schema
+        assert schema is not None  # review 节点仅发结构化审稿请求
+        self.review_calls += 1
+        self._active += 1
+        self.peak = max(self.peak, self._active)
+        await asyncio.sleep(0)  # 让出：并行扇出时其余审得以在此同时进入（峰值 > 1）
+        self._active -= 1
+        parsed = self._by_schema[schema]
+        return LlmResponse(
+            text=parsed.model_dump_json(),
+            parsed=parsed,
+            usage=_usage(),
+            served_by=ServedBy(provider="fake", model="fake"),
+        )
+
+
+class _CapturingReviewRepo:
+    """捕获 collect 落库的各列 kwargs（断言五审全部落齐一条留痕）。"""
+
+    def __init__(self) -> None:
+        self.recorded: dict[str, Any] | None = None
+
+    async def record(
+        self,
+        project_id: uuid.UUID,
+        chapter_no: int,
+        *,
+        chapter_version: int | None = None,
+        conflicts: list[dict[str, Any]],
+        foreshadow_sug: list[dict[str, Any]] | None = None,
+        style: dict[str, Any] | None = None,
+        pace: dict[str, Any] | None = None,
+        characterization: dict[str, Any] | None = None,
+        health_score: int | None = None,
+    ) -> Any:
+        self.recorded = {
+            "conflicts": conflicts,
+            "foreshadow_sug": foreshadow_sug,
+            "style": style,
+            "pace": pace,
+            "characterization": characterization,
+        }
+        return object()
+
+    async def list_for_chapter(self, project_id: uuid.UUID, chapter_no: int) -> list[Any]:
+        return []
+
+
+async def test_review_chapter_fans_out_reviews_in_parallel_and_records_all() -> None:
+    """review 节点并行扇出五审（复用主图 add-only 记账并发范式，T3.8）→ 一次 collect 落齐各列。
+
+    并发证据：mock 网关审稿并发峰值 > 1（串行恒 1）。正确性：五审各列全部落一条留痕。
+    """
+    from ww_agents import (
+        characterization_spec,
+        continuity_spec,
+        foreshadow_spec,
+        pace_spec,
+        style_drift_spec,
+    )
+    from ww_core.orchestrator.chain import review_chapter
+
+    specs = (
+        continuity_spec,
+        foreshadow_spec,
+        pace_spec,
+        style_drift_spec,
+        characterization_spec,
+    )
+    by_schema: dict[type, BaseModel] = {}
+    for spec in specs:
+        assert spec.output_schema is not None
+        by_schema[spec.output_schema] = spec.output_schema()
+    gateway = _ConcurrencyReviewGateway(by_schema=by_schema)
+    review_repo = _CapturingReviewRepo()
+    draft_store = {5: "第 5 章草稿正文。"}
+
+    async def gateway_builder(session: Any) -> Any:
+        return gateway
+
+    def memory_repos_factory(session: Any) -> Any:
+        return object()
+
+    def chapter_repo_factory(session: Any) -> Any:
+        return FakeChapterRepo(draft_store)
+
+    def review_repo_factory(session: Any) -> Any:
+        return review_repo
+
+    async def assemble(repos: Any, project_id: uuid.UUID, chapter_no: int) -> Any:
+        return FakeContext()
+
+    state = initial_chain_state(project_id=PROJECT, user_id=USER, start_chapter_no=5, count=1)
+
+    await review_chapter(
+        state,
+        gateway_builder=gateway_builder,
+        session_factory=_session_factory,
+        memory_repos_factory=memory_repos_factory,
+        chapter_repo_factory=chapter_repo_factory,
+        review_repo_factory=review_repo_factory,
+        assemble=assemble,
+        review_specs=specs,
+    )
+
+    # 并发：五审全部起跑且并发峰值 > 1（串行实现恒为 1 → 本断言 RED）。
+    assert gateway.review_calls == 5
+    assert gateway.peak >= 2
+    # 正确性：五审各列均落一条留痕（continuity→conflicts、foreshadow→sug、pace/style/角色塑造）。
+    assert review_repo.recorded is not None
+    assert review_repo.recorded["conflicts"] == []
+    assert review_repo.recorded["foreshadow_sug"] == []
+    assert review_repo.recorded["pace"] is not None
+    assert review_repo.recorded["style"] is not None
+    assert review_repo.recorded["characterization"] is not None

packages/core/ww_core/orchestrator/chain/nodes.py

@@ -10,6 +10,7 @@ checkpoint 只存控制流，正文/审稿落领域表——resume 时重读领
 
 from __future__ import annotations
 
+import asyncio
 import uuid
 from collections.abc import Awaitable, Callable, Sequence
 from contextlib import AbstractAsyncContextManager
@@ -210,9 +211,15 @@ async def review_chapter(
             "user_id": user_id,
             "review_context": review_context,
         }
+        # 并行扇出五审（复用主图 write→review 的 add-only 记账并发范式，T3.8）：各 run_review
+        # 共用本节点 session 的网关，其 SqlAlchemyLedgerSink.record 只 add（同步、不 flush），故
+        # 共享 session 并发安全；run_review 自带失败隔离（不抛、返回 incomplete 占位），gather
+        # 无需 return_exceptions。collect_reviews 在 gather 之后**一次**落库（不引入新并发写）。
+        parts = await asyncio.gather(
+            *(run_review(spec, review_state, gateway=gateway) for spec in review_specs)
+        )
         reviews: dict[str, Any] = {}
-        for spec in review_specs:
-            part = await run_review(spec, review_state, gateway=gateway)
+        for part in parts:
             reviews.update(part.get("reviews", {}))
         review_state["reviews"] = reviews
         review_repo = review_repo_factory(session)