diff --git a/packages/core/tests/test_chain_graph.py b/packages/core/tests/test_chain_graph.py index 6fd0763..40e4604 100644 --- a/packages/core/tests/test_chain_graph.py +++ b/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 diff --git a/packages/core/ww_core/orchestrator/chain/nodes.py b/packages/core/ww_core/orchestrator/chain/nodes.py index 66b8d12..04fd8c4 100644 --- a/packages/core/ww_core/orchestrator/chain/nodes.py +++ b/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)