diff --git a/apps/api/tests/test_chain.py b/apps/api/tests/test_chain.py index 109c0e8..405805b 100644 --- a/apps/api/tests/test_chain.py +++ b/apps/api/tests/test_chain.py @@ -572,6 +572,62 @@ async def test_run_chain_job_no_conflict_completes_done(_patched_runner: None) - assert [a["chapter_no"] for a in fakes["accepted"]] == [1, 2] +async def test_run_chain_job_sets_recursion_limit_scaled_to_count( + _patched_runner: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """CR-C2:run_chain_job 给 graph.ainvoke 传显式 top-level recursion_limit(防御硬化)。 + + langgraph 默认上界已足够(无实时崩溃),但每链上界应显式 = count*NODES_PER_CHAPTER + + HEADROOM。断言 config 键值 + 仍正常跑完(行为不变)。用 PROXY 包裹真 graph 记录 config + (不 setattr 到可能冻结的 CompiledStateGraph 上)。 + """ + from ww_core.orchestrator.chain import build_chain_graph as real_build + + captured: dict[str, Any] = {} + + class _SpyGraph: + def __init__(self, inner: Any) -> None: + self._inner = inner + + async def ainvoke(self, inp: Any, *, config: Any = None, **kw: Any) -> Any: + captured["config"] = config + return await self._inner.ainvoke(inp, config=config, **kw) + + def _spy_build(*args: Any, **kwargs: Any) -> Any: + return _SpyGraph(real_build(*args, **kwargs)) + + monkeypatch.setattr("ww_api.services.chain_runner.build_chain_graph", _spy_build) + + fakes = _chain_runner_fakes() + gateway = _FakeChainGateway(conflicts=[]) + job_id = uuid.uuid4() + saver = MemorySaver() + + await run_chain_job( + _session_factory, + job_id, + project_id=uuid.uuid4(), + user_id=USER, + chain_key="draft_volume", + start_chapter_no=1, + count=7, + chain_gateway_builder=_async_returning(gateway), + checkpointer_ctx=lambda: _memsaver_ctx(saver), + accept_op=fakes["accept_op"], + chapter_repo_factory=fakes["chapter_repo_factory"], + review_repo_factory=fakes["review_repo_factory"], + ) + + # (1) 显式上界 = count*NODES_PER_CHAPTER + HEADROOM = 7*4+10 = 38。 + assert captured["config"]["recursion_limit"] == 7 * 4 + 10 == 38 + # (2) recursion_limit 是 top-level 键(不在 configurable 下——langgraph 只认 top-level)。 + assert "recursion_limit" not in captured["config"]["configurable"] + # (3) 仍正常跑完(行为不变)。 + rec = _RecordingJobRepo.state[job_id] + assert rec["status"] == STATUS_DONE + assert rec["result"]["written"] == list(range(1, 8)) + + async def test_run_chain_job_conflict_sets_awaiting_then_resume_done( _patched_runner: None, ) -> None: diff --git a/apps/api/ww_api/services/chain_runner.py b/apps/api/ww_api/services/chain_runner.py index eda1dce..7f8cec2 100644 --- a/apps/api/ww_api/services/chain_runner.py +++ b/apps/api/ww_api/services/chain_runner.py @@ -50,6 +50,11 @@ log = structlog.get_logger(__name__) # 链 job 的 kind(jobs.kind 自由 Text 列,零迁移复用;§7)。 JOB_KIND_CHAIN = "chain" +# 每章在链图里的超级步数(write→review→decide→accept),用于算 recursion_limit 上界。 +NODES_PER_CHAPTER = 4 # 每章 4 个超级步:write→review→decide→accept +# START 边 + interrupt/resume 重入的余量(防御性冗余,非精确计数)。 +RECURSION_LIMIT_HEADROOM = 10 # START 边 + interrupt/resume 重入余量 + # 「按 session 建 digest 网关」缝:accept 节点在自建短事务里需 light 档网关跑终稿提炼。 DigestGatewayBuilder = Callable[[Any], Awaitable[Gateway]] @@ -207,7 +212,14 @@ async def run_chain_job( try: async with checkpointer_ctx() as checkpointer: await _set_running(session_factory, job_id) - config: RunnableConfig = {"configurable": {"thread_id": str(job_id)}} + # 显式每链 recursion_limit 上界(防御硬化,CR-C2):count 已被端点上限 MAX_CHAIN_COUNT + # 夹住,故 count*每章步数+余量 是安全且贴合的上界。**必须放 config 顶层**——放到 + # `configurable` 下 langgraph 会静默忽略。单份 config 同时喂初始 run 与 resume 分支。 + recursion_limit = count * NODES_PER_CHAPTER + RECURSION_LIMIT_HEADROOM + config: RunnableConfig = { + "configurable": {"thread_id": str(job_id)}, + "recursion_limit": recursion_limit, + } graph = build_chain_graph( chain_gateway_builder, session_factory=session_factory,