762 lines
27 KiB
Python
762 lines
27 KiB
Python
"""C2 多章链端点 + 服务层单测（chain-workflow §6/§10）——内存替身，无 DB/无网络/无真 LLM。
|
||
|
||
覆盖：
|
||
- 端点契约（HTTP 层，`run_chain_job` 被 no-op 替换以隔离调度逻辑）：
|
||
run → 202 `{job_id,...}`；resume → 202；未知 chain_key → 404；count 越界（>50 / 0）→ 422；
|
||
resume 非 awaiting 态 → 409；项目不存在 → 404；resume job 不存在 → 404。
|
||
- 服务层 `run_chain_job`（mock 网关 + MemorySaver + fake session_factory + fake accept_op）：
|
||
无冲突全跑 → job done + result.written 满；有冲突 → interrupt → job awaiting_input；
|
||
resume 续跑 → done；token 绝不入 job result。
|
||
- `build_accept_op`：冲突缺判 → CONFLICT_UNRESOLVED；无冲突章正常落库（经 fake op 桩验证）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from collections.abc import AsyncIterator
|
||
from contextlib import asynccontextmanager
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from fakes_projects import FakeJobRepo, FakeProjectRepo
|
||
from fastapi import FastAPI
|
||
from langgraph.checkpoint.memory import MemorySaver
|
||
from pydantic import BaseModel
|
||
from ww_agents import Conflict, ContinuityReview
|
||
from ww_api.schemas.projects import ConflictDecision
|
||
from ww_api.services.chain_runner import build_accept_op, run_chain_job
|
||
from ww_core.domain.job_repo import STATUS_AWAITING, STATUS_DONE
|
||
from ww_core.domain.project_repo import ProjectCreate
|
||
from ww_llm_gateway.types import LlmRequest, LlmResponse, ServedBy, Usage
|
||
|
||
PROJECT_OWNER = uuid.UUID(int=1)
|
||
USER = uuid.UUID("00000000-0000-0000-0000-0000000000aa")
|
||
|
||
|
||
# ---- 共享替身 ----
|
||
|
||
|
||
def _usage() -> Usage:
|
||
return Usage(
|
||
provider="fake", model="fake", input_tokens=1, output_tokens=1, cost_minor=0, currency="USD"
|
||
)
|
||
|
||
|
||
class _FakeChainGateway:
|
||
"""链网关 mock：write（无 output_schema）回固定正文；审稿按 schema 回 parsed。"""
|
||
|
||
def __init__(self, *, conflicts: list[Conflict]) -> None:
|
||
self._continuity = ContinuityReview(conflicts=conflicts)
|
||
from ww_agents import foreshadow_spec, pace_spec, style_drift_spec
|
||
|
||
self._by_schema: dict[type, BaseModel] = {}
|
||
for spec in (foreshadow_spec, pace_spec, style_drift_spec):
|
||
schema = spec.output_schema
|
||
assert schema is not None
|
||
self._by_schema[schema] = schema()
|
||
|
||
async def run(self, req: LlmRequest) -> LlmResponse:
|
||
schema = req.output_schema
|
||
if schema is None:
|
||
return LlmResponse(
|
||
text="第 N 章正文。",
|
||
parsed=None,
|
||
usage=_usage(),
|
||
served_by=ServedBy(provider="fake", model="fake"),
|
||
)
|
||
parsed = self._continuity if schema is ContinuityReview else self._by_schema[schema]
|
||
return LlmResponse(
|
||
text=parsed.model_dump_json(),
|
||
parsed=parsed,
|
||
usage=_usage(),
|
||
served_by=ServedBy(provider="fake", model="fake"),
|
||
)
|
||
|
||
|
||
class _FakeDraftView:
|
||
def __init__(self, content: str) -> None:
|
||
self.content = content
|
||
|
||
|
||
class _FakeChapterRepo:
|
||
def __init__(self, store: dict[int, str]) -> None:
|
||
self._store = store
|
||
|
||
async def save_draft(
|
||
self, project_id: uuid.UUID, chapter_no: int, *, text: str, volume: int = 1
|
||
) -> Any:
|
||
self._store[chapter_no] = text
|
||
return _FakeDraftView(text)
|
||
|
||
async def get_draft(self, project_id: uuid.UUID, chapter_no: int) -> Any:
|
||
content = self._store.get(chapter_no)
|
||
return _FakeDraftView(content) if content is not None else None
|
||
|
||
|
||
class _FakeReviewRecord:
|
||
def __init__(self, chapter_no: int, conflicts: list[dict[str, Any]]) -> None:
|
||
self.id = uuid.uuid4()
|
||
self.chapter_no = chapter_no
|
||
self.conflicts = conflicts
|
||
|
||
|
||
class _FakeReviewRepo:
|
||
def __init__(self, records: dict[int, list[_FakeReviewRecord]]) -> None:
|
||
self._records = records
|
||
|
||
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:
|
||
rec = _FakeReviewRecord(chapter_no, list(conflicts))
|
||
self._records.setdefault(chapter_no, []).insert(0, rec)
|
||
return rec
|
||
|
||
async def list_for_chapter(self, project_id: uuid.UUID, chapter_no: int) -> list[Any]:
|
||
return list(self._records.get(chapter_no, []))
|
||
|
||
|
||
class _FakeSession:
|
||
def __init__(self) -> None:
|
||
self.commits = 0
|
||
|
||
async def commit(self) -> None:
|
||
self.commits += 1
|
||
|
||
|
||
@asynccontextmanager
|
||
async def _session_cm() -> AsyncIterator[Any]:
|
||
yield _FakeSession()
|
||
|
||
|
||
def _session_factory() -> Any:
|
||
"""fake 独立 session 工厂（满足 `SessionFactory` 形：`()` → async-CM）。
|
||
|
||
返回 Any 避开 AsyncSession 不变型校验（同 test_job_runner 的 `__call__ -> Any` 纪律）。
|
||
"""
|
||
return _session_cm()
|
||
|
||
|
||
@asynccontextmanager
|
||
async def _memsaver_ctx(saver: MemorySaver) -> AsyncIterator[Any]:
|
||
yield saver
|
||
|
||
|
||
def _async_returning(gateway: Any) -> Any:
|
||
"""构造「按 session 建网关」的 builder 替身：忽略 session，恒返给定 mock 网关（绝不联网）。
|
||
|
||
链节点经 `gateway_builder(session)` 现建网关（不变量 #1）；测试里 ledger 不落库（fake
|
||
session），故 builder 只需把 mock 网关交回即可。
|
||
"""
|
||
|
||
async def _build(_session: Any) -> Any:
|
||
return gateway
|
||
|
||
return _build
|
||
|
||
|
||
class _FakeContext:
|
||
stable_core = "## 世界观硬规则\n灵气可凝丹"
|
||
volatile = "## 写作指令\n写第 N 章"
|
||
|
||
|
||
# ---- 端点契约（HTTP 层，run_chain_job no-op）----
|
||
|
||
|
||
def _app(project_repo: FakeProjectRepo, job_repo: FakeJobRepo) -> FastAPI:
|
||
import os
|
||
|
||
from cryptography.fernet import Fernet
|
||
|
||
os.environ.setdefault("CREDENTIAL_ENC_KEY", Fernet.generate_key().decode())
|
||
from ww_api.main import create_app
|
||
from ww_api.services.chain_deps import get_checkpointer_factory
|
||
from ww_api.services.project_deps import (
|
||
get_chain_gateway,
|
||
get_chain_gateway_builder,
|
||
get_digest_gateway_builder,
|
||
get_job_repo,
|
||
get_project_repo,
|
||
get_session_factory,
|
||
)
|
||
from ww_db import get_session
|
||
|
||
app = create_app()
|
||
app.dependency_overrides[get_project_repo] = lambda: project_repo
|
||
app.dependency_overrides[get_job_repo] = lambda: job_repo
|
||
app.dependency_overrides[get_session] = lambda: _FakeSession()
|
||
app.dependency_overrides[get_session_factory] = lambda: _session_factory
|
||
app.dependency_overrides[get_chain_gateway] = lambda: _FakeChainGateway(conflicts=[])
|
||
app.dependency_overrides[get_chain_gateway_builder] = lambda: _async_returning(
|
||
_FakeChainGateway(conflicts=[])
|
||
)
|
||
saver = MemorySaver()
|
||
app.dependency_overrides[get_checkpointer_factory] = lambda: lambda: _memsaver_ctx(saver)
|
||
app.dependency_overrides[get_digest_gateway_builder] = lambda: lambda _s: None
|
||
return app
|
||
|
||
|
||
def _client(app: FastAPI) -> httpx.AsyncClient:
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
return httpx.AsyncClient(transport=transport, base_url="http://test")
|
||
|
||
|
||
async def _seed_project(repo: FakeProjectRepo) -> uuid.UUID:
|
||
view = await repo.create(PROJECT_OWNER, ProjectCreate(title="测试作品", genre="玄幻"))
|
||
return uuid.UUID(str(view.id))
|
||
|
||
|
||
@pytest.fixture
|
||
def _noop_run_chain_job(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
|
||
"""把端点登记的 `run_chain_job` 换成记录调用的 no-op（隔离调度逻辑，不真跑图）。"""
|
||
calls: list[dict[str, Any]] = []
|
||
|
||
async def _fake(*args: Any, **kwargs: Any) -> None:
|
||
calls.append({"args": args, "kwargs": kwargs})
|
||
|
||
monkeypatch.setattr("ww_api.routers.chain.run_chain_job", _fake)
|
||
return calls
|
||
|
||
|
||
async def test_run_chain_returns_202_and_schedules_job(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/draft_volume/run",
|
||
json={"start_chapter_no": 1, "count": 3},
|
||
)
|
||
|
||
assert resp.status_code == 202
|
||
body = resp.json()
|
||
assert body["chain_key"] == "draft_volume"
|
||
assert body["start_chapter_no"] == 1
|
||
assert body["count"] == 3
|
||
assert uuid.UUID(body["job_id"])
|
||
# job 行已落（queued）+ 调度了一次后台任务
|
||
assert len(job_repo.rows) == 1
|
||
assert len(_noop_run_chain_job) == 1
|
||
|
||
|
||
async def test_run_continue_volume_chain_returns_202_and_forwards_chain_key(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
"""F2：continue_volume 链受支持 → 202，且 chain_key 透传给 run_chain_job（续写模式）。"""
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/continue_volume/run",
|
||
json={"start_chapter_no": 2, "count": 2},
|
||
)
|
||
|
||
assert resp.status_code == 202
|
||
body = resp.json()
|
||
assert body["chain_key"] == "continue_volume"
|
||
assert len(_noop_run_chain_job) == 1
|
||
assert _noop_run_chain_job[0]["kwargs"]["chain_key"] == "continue_volume"
|
||
|
||
|
||
async def test_run_chain_unknown_key_returns_404(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/unknown_chain/run",
|
||
json={"start_chapter_no": 1, "count": 2},
|
||
)
|
||
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == "NOT_FOUND"
|
||
assert len(_noop_run_chain_job) == 0 # 未调度
|
||
|
||
|
||
async def test_run_chain_count_over_50_returns_422(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/draft_volume/run",
|
||
json={"start_chapter_no": 1, "count": 51},
|
||
)
|
||
|
||
assert resp.status_code == 422
|
||
|
||
|
||
async def test_run_chain_count_zero_returns_422(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/draft_volume/run",
|
||
json={"start_chapter_no": 1, "count": 0},
|
||
)
|
||
|
||
assert resp.status_code == 422
|
||
|
||
|
||
async def test_run_chain_project_not_found_returns_404(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{uuid.uuid4()}/chains/draft_volume/run",
|
||
json={"start_chapter_no": 1, "count": 2},
|
||
)
|
||
|
||
assert resp.status_code == 404
|
||
|
||
|
||
async def test_resume_chain_awaiting_returns_202(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
# 预置一个 awaiting_input 的链 job
|
||
job = await job_repo.create(pid, "chain")
|
||
await job_repo.set_awaiting(job.id, {"awaiting_chapter": 1})
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/runs/{job.id}/resume",
|
||
json={"decisions": [{"conflict_index": 0, "verdict": "ignore"}]},
|
||
)
|
||
|
||
assert resp.status_code == 202
|
||
assert resp.json()["job_id"] == str(job.id)
|
||
assert len(_noop_run_chain_job) == 1
|
||
|
||
|
||
async def test_resume_continue_volume_reports_true_chain_key(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
"""回归守卫 #1：resume continue_volume run → 回执 chain_key="continue_volume"（读自
|
||
job.result，run 时由 _chain_result 持久化），而非旧 bug 按 job.kind 推断的 "draft_volume"。
|
||
且转发给 run_chain_job 的 chain_key 也是真值（续写模式不被错降为 draft_volume）。"""
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
job = await job_repo.create(pid, "chain") # kind 恒为通用常量 "chain"
|
||
# run 时 interrupt 把真 chain_key 持久进 result（continue_volume）。
|
||
await job_repo.set_awaiting(job.id, {"awaiting_chapter": 1, "chain_key": "continue_volume"})
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/runs/{job.id}/resume",
|
||
json={"decisions": [{"conflict_index": 0, "verdict": "ignore"}]},
|
||
)
|
||
|
||
assert resp.status_code == 202
|
||
assert resp.json()["chain_key"] == "continue_volume"
|
||
assert _noop_run_chain_job[0]["kwargs"]["chain_key"] == "continue_volume"
|
||
|
||
|
||
async def test_resume_chain_non_awaiting_returns_409(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
job = await job_repo.create(pid, "chain") # queued（非 awaiting）
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/runs/{job.id}/resume",
|
||
json={"decisions": []},
|
||
)
|
||
|
||
assert resp.status_code == 409
|
||
assert resp.json()["error"]["code"] == "CONFLICT"
|
||
assert len(_noop_run_chain_job) == 0
|
||
|
||
|
||
async def test_resume_chain_job_not_found_returns_404(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/runs/{uuid.uuid4()}/resume",
|
||
json={"decisions": []},
|
||
)
|
||
|
||
assert resp.status_code == 404
|
||
|
||
|
||
async def test_resume_chain_wrong_project_returns_404(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
"""所有权校验（审评 #4）：job 属项目 A，从项目 B 路径 resume → 404（同案，不枚举）。"""
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid_a = await _seed_project(project_repo)
|
||
pid_b = uuid.UUID(
|
||
str((await project_repo.create(PROJECT_OWNER, ProjectCreate(title="作品B"))).id)
|
||
)
|
||
job = await job_repo.create(pid_a, "chain") # job 属 A
|
||
await job_repo.set_awaiting(job.id, {"awaiting_chapter": 1})
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid_b}/chains/runs/{job.id}/resume", # 从 B 续 A 的 job
|
||
json={"decisions": []},
|
||
)
|
||
|
||
assert resp.status_code == 404
|
||
assert len(_noop_run_chain_job) == 0 # 未调度
|
||
|
||
|
||
async def test_resume_chain_concurrent_claim_returns_409(
|
||
_noop_run_chain_job: list[dict[str, Any]],
|
||
) -> None:
|
||
"""原子抢占（审评 #3）：第一个 resume 已把 awaiting→running，第二个抢不到 → 409。"""
|
||
project_repo = FakeProjectRepo()
|
||
job_repo = FakeJobRepo()
|
||
pid = await _seed_project(project_repo)
|
||
job = await job_repo.create(pid, "chain")
|
||
await job_repo.set_awaiting(job.id, {"awaiting_chapter": 1})
|
||
# 模拟并发：第一个 resume 已抢占 awaiting→running。
|
||
claimed = await job_repo.claim_awaiting_to_running(job.id)
|
||
assert claimed is not None
|
||
app = _app(project_repo, job_repo)
|
||
|
||
async with _client(app) as client:
|
||
resp = await client.post(
|
||
f"/projects/{pid}/chains/runs/{job.id}/resume",
|
||
json={"decisions": []},
|
||
)
|
||
|
||
# job 现态 running（非 awaiting）→ 守卫先在 status 检查处 409（CONFLICT）。
|
||
assert resp.status_code == 409
|
||
assert resp.json()["error"]["code"] == "CONFLICT"
|
||
assert len(_noop_run_chain_job) == 0
|
||
|
||
|
||
# ---- 服务层 run_chain_job（直接 await，注入全 fake；job 状态由 fake SqlJobRepo 替身记录）----
|
||
|
||
|
||
class _RecordingJobRepo:
|
||
"""run_chain_job 内部 `SqlJobRepo(session)` 的替身（monkeypatch）：记录状态流转。"""
|
||
|
||
state: dict[uuid.UUID, dict[str, Any]] = {}
|
||
|
||
def __init__(self, session: Any) -> None:
|
||
pass
|
||
|
||
async def set_running(self, job_id: uuid.UUID) -> Any:
|
||
_RecordingJobRepo.state.setdefault(job_id, {})["status"] = "running"
|
||
return None
|
||
|
||
async def complete(self, job_id: uuid.UUID, result: dict[str, Any]) -> Any:
|
||
_RecordingJobRepo.state[job_id] = {"status": STATUS_DONE, "result": result}
|
||
return None
|
||
|
||
async def set_awaiting(self, job_id: uuid.UUID, result: dict[str, Any]) -> Any:
|
||
_RecordingJobRepo.state[job_id] = {"status": STATUS_AWAITING, "result": result}
|
||
return None
|
||
|
||
async def fail(self, job_id: uuid.UUID, error: str) -> Any:
|
||
_RecordingJobRepo.state[job_id] = {"status": "failed", "error": error}
|
||
return None
|
||
|
||
|
||
def _chain_runner_fakes() -> dict[str, Any]:
|
||
draft_store: dict[int, str] = {}
|
||
review_store: dict[int, list[_FakeReviewRecord]] = {}
|
||
accepted: list[dict[str, Any]] = []
|
||
|
||
async def accept_op(
|
||
*, project_id: uuid.UUID, chapter_no: int, user_id: uuid.UUID, decisions: list[Any]
|
||
) -> None:
|
||
accepted.append({"chapter_no": chapter_no, "decisions": list(decisions)})
|
||
|
||
return {
|
||
"draft_store": draft_store,
|
||
"review_store": review_store,
|
||
"accepted": accepted,
|
||
"accept_op": accept_op,
|
||
"chapter_repo_factory": lambda _s: _FakeChapterRepo(draft_store),
|
||
"review_repo_factory": lambda _s: _FakeReviewRepo(review_store),
|
||
}
|
||
|
||
|
||
@pytest.fixture
|
||
def _patched_runner(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""把 run_chain_job 里的 SqlJobRepo + assemble + sql_memory_repos 换成 fake（不触 DB）。"""
|
||
_RecordingJobRepo.state = {}
|
||
monkeypatch.setattr("ww_api.services.chain_runner.SqlJobRepo", _RecordingJobRepo)
|
||
|
||
async def _fake_assemble(
|
||
repos: Any, project_id: uuid.UUID, chapter_no: int, *a: Any, **k: Any
|
||
) -> _FakeContext:
|
||
return _FakeContext()
|
||
|
||
monkeypatch.setattr("ww_api.services.chain_runner.assemble", _fake_assemble)
|
||
monkeypatch.setattr("ww_api.services.chain_runner.sql_memory_repos", lambda _s: object())
|
||
|
||
|
||
async def test_run_chain_job_no_conflict_completes_done(_patched_runner: None) -> None:
|
||
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=2,
|
||
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"],
|
||
)
|
||
|
||
rec = _RecordingJobRepo.state[job_id]
|
||
assert rec["status"] == STATUS_DONE
|
||
assert rec["result"]["written"] == [1, 2]
|
||
assert rec["result"]["completed"] is True
|
||
assert rec["result"]["awaiting_chapter"] is None
|
||
# token 不入 result：只含章号/标志，无正文/prompt
|
||
assert "第 N 章正文" not in str(rec["result"])
|
||
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)
|
||
|
||
async def aget_state(self, config: Any, **kw: Any) -> Any:
|
||
# runner 现经公开 aget_state 判 interrupt（CR-M3）——代理透传给真 graph。
|
||
return await self._inner.aget_state(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:
|
||
fakes = _chain_runner_fakes()
|
||
conflict = Conflict(type="性格漂移", where="第3段", suggestion="统一主角姓名")
|
||
gateway = _FakeChainGateway(conflicts=[conflict])
|
||
job_id = uuid.uuid4()
|
||
saver = MemorySaver() # 同 saver 跨 run/resume 两次调用
|
||
project_id = uuid.uuid4()
|
||
|
||
# 初始 run：第 1 章四审报冲突 → interrupt → job awaiting_input。
|
||
await run_chain_job(
|
||
_session_factory,
|
||
job_id,
|
||
project_id=project_id,
|
||
user_id=USER,
|
||
chain_key="draft_volume",
|
||
start_chapter_no=1,
|
||
count=1,
|
||
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"],
|
||
)
|
||
|
||
rec = _RecordingJobRepo.state[job_id]
|
||
assert rec["status"] == STATUS_AWAITING
|
||
assert rec["result"]["awaiting_chapter"] == 1
|
||
assert fakes["accepted"] == [] # 尚未验收
|
||
|
||
# resume：作者裁决 → 续跑 accept → done。
|
||
decisions = [ConflictDecision(conflict_index=0, verdict="ignore", note="笔误")]
|
||
await run_chain_job(
|
||
_session_factory,
|
||
job_id,
|
||
project_id=project_id,
|
||
user_id=USER,
|
||
chain_key="draft_volume",
|
||
start_chapter_no=1,
|
||
count=1,
|
||
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"],
|
||
resume_decisions=decisions,
|
||
)
|
||
|
||
rec = _RecordingJobRepo.state[job_id]
|
||
assert rec["status"] == STATUS_DONE
|
||
assert rec["result"]["written"] == [1]
|
||
assert [a["chapter_no"] for a in fakes["accepted"]] == [1]
|
||
assert fakes["accepted"][0]["decisions"] == [
|
||
{"conflict_index": 0, "verdict": "ignore", "note": "笔误"}
|
||
]
|
||
|
||
|
||
async def test_run_chain_job_error_marks_failed_safely(_patched_runner: None) -> None:
|
||
"""work 抛 → job failed + 脱敏文案（不泄 str(exc)），异常不冒泡。"""
|
||
fakes = _chain_runner_fakes()
|
||
job_id = uuid.uuid4()
|
||
|
||
class _BoomGateway:
|
||
async def run(self, req: LlmRequest) -> LlmResponse:
|
||
raise RuntimeError("boom secret=/internal/path")
|
||
|
||
await run_chain_job(
|
||
_session_factory,
|
||
job_id,
|
||
project_id=uuid.uuid4(),
|
||
user_id=USER,
|
||
chain_key="draft_volume",
|
||
start_chapter_no=1,
|
||
count=1,
|
||
chain_gateway_builder=_async_returning(_BoomGateway()),
|
||
checkpointer_ctx=lambda: _memsaver_ctx(MemorySaver()),
|
||
accept_op=fakes["accept_op"],
|
||
chapter_repo_factory=fakes["chapter_repo_factory"],
|
||
review_repo_factory=fakes["review_repo_factory"],
|
||
)
|
||
|
||
rec = _RecordingJobRepo.state[job_id]
|
||
assert rec["status"] == "failed"
|
||
assert rec["error"] == "任务执行失败"
|
||
assert "secret" not in rec["error"]
|
||
|
||
|
||
# ---- build_accept_op：冲突缺判 gate ----
|
||
|
||
|
||
async def test_build_accept_op_missing_decision_raises_conflict_unresolved(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
from ww_shared import AppError, ErrorCode
|
||
|
||
review_store: dict[int, list[_FakeReviewRecord]] = {
|
||
1: [_FakeReviewRecord(1, [{"type": "矛盾", "where": "第1段"}])]
|
||
}
|
||
|
||
class _Session:
|
||
async def commit(self) -> None: ...
|
||
|
||
@asynccontextmanager
|
||
async def _sf_cm() -> AsyncIterator[Any]:
|
||
yield _Session()
|
||
|
||
def _sf() -> Any:
|
||
return _sf_cm()
|
||
|
||
# 桩出 accept_op 依赖：链 accept 节点重读审稿、跑 gate（缺判应在 gate 拦下，不触 digest/事务）。
|
||
monkeypatch.setattr(
|
||
"ww_api.services.chain_runner.SqlChapterRepo", lambda _s: _FakeChapterRepo({1: "正文"})
|
||
)
|
||
monkeypatch.setattr(
|
||
"ww_api.services.chain_runner.SqlReviewRepo", lambda _s: _FakeReviewRepo(review_store)
|
||
)
|
||
monkeypatch.setattr("ww_api.services.chain_runner.SqlDigestAppendRepo", lambda _s: object())
|
||
|
||
async def _digest_builder(_s: Any) -> Any:
|
||
return object()
|
||
|
||
accept_op = build_accept_op(session_factory=_sf, digest_gateway_builder=_digest_builder)
|
||
|
||
with pytest.raises(AppError) as exc:
|
||
await accept_op(project_id=uuid.uuid4(), chapter_no=1, user_id=USER, decisions=[])
|
||
assert exc.value.code == ErrorCode.CONFLICT_UNRESOLVED
|