diff --git a/PROGRESS.md b/PROGRESS.md index d0c7f86..e39702b 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -111,6 +111,7 @@ T0.1 monorepo 骨架 ✅ @devops · T0.2 16 MVP 表迁移（无漂移，users st > 格式：`- [YYYY-MM-DD] @skill 完成/进展 Txx — 一句话结果 + 影响的契约/文件` +- [2026-06-23] @backend — **✅ Chain Workflow C2（多章链 服务+端点+schema+checkpointer 接线，分支 `feat/chain-workflow`）**。承 C1(@llm)链图（`build_chain_graph` §3.3），落地服务/端点壳：① **3 端点**`routers/chain.py`（`POST .../chains/{key}/run`→202 `ChainRunAccepted`；`POST .../chains/runs/{job_id}/resume`→202；`GET /jobs/{id}` 复用）。校验：count 1..50→422、未知 chain_key→404、项目不存在→404、resume job 不存在→404、resume 非 awaiting→**409 新码 `CONFLICT`**、无凭据→503。② **schema**`schemas/chain.py`（`ChainRunRequest/ChainRunAccepted/ChainResumeRequest`，`ConflictDecision` 复用）。③ **服务**`services/chain_runner.py`（`run_chain_job` 仿 `run_job` 壳自建独立 session 驱动链图：set_running→ainvoke→据 `__interrupt__` 置 `awaiting_input` 或 `done`/`failed`；`build_accept_op` 在 apps/api 装配验收事务闭包注入图节点，守 #3/#4；token 不入 result/日志）+ `services/chain_deps.py`（`get_checkpointer_factory`：运行时 `AsyncPostgresSaver` 上下文 / 测试 MemorySaver）。④ **零迁移**（§7）：复用 `jobs`，新增 `status="awaiting_input"` + `JobRepo.set_awaiting`，awaiting 章经 `result.awaiting_chapter`；新错误码 `ErrorCode.CONFLICT`(409)。⑤ **新网关缝**`project_deps.build_chain_gateway`/`get_chain_gateway`（按请求 tier writer/analyst/light 分派——单档网关恒返该档会错路由 review/digest，见 gotcha）+ `get_digest_gateway_builder`。**单测**`apps/api/tests/test_chain.py` 12 用例（mock 网关 + MemorySaver + fake session/accept_op，无 DB/无网络/无真 LLM）：run→202、resume→202、未知 key 404、count 越界(>50/0) 422、resume 非 awaiting 409、项目/ job 不存在 404、run_chain_job 无冲突→done+written 满、冲突→awaiting→resume→done、错误脱敏、accept_op 冲突缺判→CONFLICT_UNRESOLVED。**全仓门禁绿**：ruff/format 干净、mypy **193 Success**、alembic **无漂移**（无 ORM 变更）、pytest **600 passed**。守不变量 #1/#3/#4/#5/#9。记 contracts C-Chain 稳定 + 2 decision + 3 gotcha。**唯一新增 DDL（langgraph 检查点表）= C3 迁移；前端链 UI = 契约稳定后 @frontend follow-up（本期非目标）。** - [2026-06-22] @orchestrator — **🎉 T6 创作工具箱（通用生成器框架）全工具箱交付 + 独立复核（分支 `feat/t6-creation-toolbox`）**。多 agent 并行波次：Wave 0（@llm T6.5 ‖ @backend 描述符类型 ‖ P2 收尾）→ Wave A（@backend T6.2+T6.3 TOOLBOX+端点合龙，契约稳定点）→ Wave B（@frontend T6.6 ‖ @qa T6.4）→ Wave C 收口，目录单写者零写冲突、契约先行、零返工。**@llm T6.5**：`ww_agents` +7 输出 schema + 7 spec（book-title/blurb/name/golden-finger/glossary/opening/fine-outline，只声明 tier #2）+ `build_outline_chapter_context`（with_outline_chapter 策略）。**@frontend T6.6**：`gen:api` 纳入 3 端点；`app/projects/[id]/toolbox/page.tsx`(RSC) + `components/toolbox/{ToolboxPage,ToolCard,GeneratorRunner}` + `lib/toolbox/*`（声明驱动纯函数，按 `input_fields` 渲染表单、按 `output_kind` 渲染预览、可入库者复用 `ConflictAdjudication` 走 409→ack）+ LeftNav「工具箱」+ ⌘K `nav-toolbox`/`action-gen-*`；legacy 3 跳现页不回归。**@qa T6.4**：`tests/test_t6_toolbox_e2e.py` 5 用例真 pg + mock 网关零 token，**未发现端点 bug**。**P2 收尾**：限流→`decisions.md` 记延后（单用户原型）；noopener/Committable 经查本分支早已修；`provider_deps.py` 残留 `type:ignore` 为 SDK 私有属性探活、load-bearing 保留。**最终独立复跑全仓门禁绿**：后端 ruff/format 干净、mypy **195 Success**、alembic **无漂移**（descriptor 在代码、复用既有表，零建表迁移）、pytest **583 passed**；前端 gen:api/lint/tsc/vitest **279**/build OK。守不变量 #2（只声明 tier）/#3（预览不写库、入库经验收 gate）/#9（system_prompt 进缓存前块）。**"加生成器=加一份声明" 经实证（8 个新生成器复用同一通用执行器 + 同一前端组件）。** **实景 browse 复验（工具箱卡片栅格→脑洞预览→词条入库）待做。** 后续可选 P5：封面图像轨道 / 自定义 skill 运行期 schema / 收敛已有 3 竖井。 - [2026-06-20] UX计划 B0(可控版)+F1(控件) — **✅ 本章注入透明 可控版（分支 `feat/ux-b0-controllable-injection`，提交 `0d473e7` 后端 + `6449c31` 前端，未合 develop）**。在 B0 读端点之上加作者微调，守不变量 #6「确定性选择」的作者兜底。**后端**(@backend/@db)：`PUT /projects/{id}/chapters/{no}/injection` ← `InjectionOverrideRequest{pinned,excluded,recent_n(1..20→422)}` → `InjectionResponse`（新增回显 `pinned`/`excluded`，`selected[].reasons` 可含新值 **`author_pin`**）；`select_relevant_entities` 加 `pinned/excluded` frozenset 入参（pin 强制纳入加 author_pin / excluded 强制剔除且**优先于 pin**）；`assemble` 加 `override` 关键字参（recent_n 覆盖回看章数）；**GET injection + PUT + draft 流式端点同读同一覆盖**（看到的=写章用的）。持久化：**新表 `chapter_injection`**（迁移 `ad2c4c663daf`，唯一 `(project_id,chapter_no)`；复用 outline 行污染 beats 已否决）+ 新 `domain/injection_repo.py`（`InjectionOverride`/`EntityRef`/`SqlInjectionOverrideRepo`，upsert 只 flush 端点 commit）。**前端**(@frontend)：`gen:api` 纳入 PUT + 新 schema；`lib/workbench/injection.ts` override 纯函数（`currentOverride`/`withPinToggled`/`withExcluded`/`withRestored`/`withRecentN`/`isPinned`，钳 1..20）；`useInjection` 加 `saving` 态 + `togglePin/exclude/restore/setRecentN`（PUT 后以服务端确定结果为准、写失败可读文案不改本地态）；`ChapterAssistant` 每实体 📌置顶/✕排除 + 「近 N 章」步进器 + 「已排除」恢复区。**全仓门禁绿**：后端 ruff/format/mypy **163**/alembic 无漂移/pytest **476**；前端 lint/tsc/vitest **183**/build。记 contracts C3 再扩。**实景 browse 复验待做。** UX 计划余项：R1/R2/R3 审稿人体学 + Tier4。 - [2026-06-20] UX计划 M1+M2 — **✅ 移动端可用（纯前端，分支 `feat/ux-m1-m2-mobile` 提交 9b893a6）**。救活手机端（390px）写→审主路径。**M1 导航响应式抽屉**：抽出纯逻辑 `lib/nav/items.ts`（`GLOBAL_NAV_ITEMS`/`projectNavItems`/`isNavItemActive` + `ActiveNav` 类型，6 vitest），桌面静态侧栏与移动抽屉共用避免漂移；新增通用 `Drawer`（遮罩点击/Esc 关闭/打开聚焦面板，复用 `CommandPalette` a11y 模式，`lg:hidden`）+ `NavDrawer`（汉堡 `aria-label=打开导航` + 左抽屉）+ `NavItems`；`LeftNav` 改 `hidden lg:block`，`AppShell` 顶栏加汉堡。**M2 工具条不溢出 + 移动可达**：`Workbench` 工具条改 `flex-wrap`（390px 不再横切「尚未保存」）；`ChapterList`/`ChapterAssistant` 抽出 `ChapterListContent`/`AssistantContent`，`Workbench` 经左/右 `Drawer` 在 ` JobView: + view = self.rows[job_id].model_copy( + update={"status": STATUS_AWAITING, "result": dict(result)} + ) + self.rows[job_id] = view + return view + async def fail(self, job_id: uuid.UUID, error: str) -> JobView: view = self.rows[job_id].model_copy(update={"status": STATUS_FAILED, "error": error}) self.rows[job_id] = view diff --git a/apps/api/tests/test_chain.py b/apps/api/tests/test_chain.py new file mode 100644 index 0000000..9d334c0 --- /dev/null +++ b/apps/api/tests/test_chain.py @@ -0,0 +1,586 @@ +"""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, + 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 + + +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_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=[]) + 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_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_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 + + +# ---- 服务层 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, + gateway=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_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, + gateway=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, + gateway=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, + gateway=_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 diff --git a/apps/api/ww_api/main.py b/apps/api/ww_api/main.py index 56f34c4..e64b5f6 100644 --- a/apps/api/ww_api/main.py +++ b/apps/api/ww_api/main.py @@ -16,6 +16,7 @@ from ww_shared import AppError, ErrorBody, ErrorCode, ErrorEnvelope from ww_api.logging_config import configure_logging, get_logger from ww_api.middleware import REQUEST_ID_HEADER, request_id_middleware from ww_api.routers import ( + chain, foreshadow, generation, health, @@ -117,6 +118,7 @@ def create_app() -> FastAPI: app.include_router(generation.router) app.include_router(generation.skills_router) app.include_router(toolbox.router) + app.include_router(chain.router) app.include_router(settings_providers.router) app.include_router(kimi_oauth.router) return app diff --git a/apps/api/ww_api/routers/chain.py b/apps/api/ww_api/routers/chain.py new file mode 100644 index 0000000..738cdf2 --- /dev/null +++ b/apps/api/ww_api/routers/chain.py @@ -0,0 +1,249 @@ +"""多章工作流链端点（chain-workflow 设计 §6）——发起 / 续跑，进度走 `GET /jobs/{id}`。 + +三端点： +- `POST /projects/{pid}/chains/{chain_key}/run` → 202 `ChainRunAccepted{job_id,...}`： + 写一行 jobs(queued, kind=chain) 立即返 202，链经 BackgroundTask `run_chain_job` 跑 + （**自建独立 session**，请求 session 已关闭）。无凭据 → 流前 `LLM_UNAVAILABLE`（503， + 调度 job 前拦下）。未知 chain_key → 404；`count` 1..50 越界 → 422（schema Field 约束）。 +- `GET /jobs/{job_id}`：复用现有 jobs 端点轮询进度/awaiting（不在本 router）。 +- `POST /projects/{pid}/chains/runs/{job_id}/resume` → 202：仅当 job=awaiting_input； + 带裁决经 BackgroundTask resume 续跑。非 awaiting 态 → 409。 + +不变量：链是**编排**新增、非业务重写——复用同一 `assemble`/审稿 spec/`run_accept_transaction` +（经 `accept_op` 闭包注入），守 #1/#3/#4/#5/#9。token 绝不入 job result/状态/日志（§5）。 +""" + +from __future__ import annotations + +import uuid +from typing import Annotated, Any + +from fastapi import APIRouter, BackgroundTasks, Depends, Request, Response +from sqlalchemy.ext.asyncio import AsyncSession +from ww_core.domain import JobRepo, ProjectRepo +from ww_core.domain.job_repo import STATUS_AWAITING +from ww_db import get_session +from ww_llm_gateway import Gateway +from ww_shared import AppError, ErrorCode, ErrorEnvelope + +from ww_api.logging_config import get_logger +from ww_api.schemas.chain import ( + ChainResumeRequest, + ChainRunAccepted, + ChainRunRequest, +) +from ww_api.services.chain_deps import ( + CheckpointerFactory, + get_checkpointer_factory, +) +from ww_api.services.chain_runner import ( + JOB_KIND_CHAIN, + build_accept_op, + run_chain_job, +) +from ww_api.services.credentials import STUB_OWNER_ID +from ww_api.services.foreshadow_scan import SessionFactory +from ww_api.services.project_deps import ( + GatewayDigestBuilder, + get_chain_gateway, + get_digest_gateway_builder, + get_job_repo, + get_project_repo, + get_session_factory, +) + +log = get_logger("ww.api.chain") + +router = APIRouter(prefix="/projects", tags=["chain"]) + +# 本期唯一内置链（§1.3/§3）。未知 key → 404（系统边界，fail fast）。 +SUPPORTED_CHAINS = frozenset({"draft_volume"}) + +ProjectRepoDep = Annotated[ProjectRepo, Depends(get_project_repo)] +JobRepoDep = Annotated[JobRepo, Depends(get_job_repo)] +ChainGatewayDep = Annotated[Gateway, Depends(get_chain_gateway)] +SessionDep = Annotated[AsyncSession, Depends(get_session)] +SessionFactoryDep = Annotated[SessionFactory, Depends(get_session_factory)] +CheckpointerFactoryDep = Annotated[CheckpointerFactory, Depends(get_checkpointer_factory)] +DigestBuilderDep = Annotated[GatewayDigestBuilder, Depends(get_digest_gateway_builder)] + +_RUN_ERRORS: dict[int | str, dict[str, Any]] = { + 404: {"model": ErrorEnvelope, "description": "项目或链 key 不存在"}, + 422: {"model": ErrorEnvelope, "description": "请求参数非法（count 越界等）"}, + 503: {"model": ErrorEnvelope, "description": "无可用 LLM 凭据"}, +} +_RESUME_ERRORS: dict[int | str, dict[str, Any]] = { + 404: {"model": ErrorEnvelope, "description": "项目或 job 不存在"}, + 409: {"model": ErrorEnvelope, "description": "job 非 awaiting_input 态，不可续跑"}, +} + + +async def _require_project(project_repo: ProjectRepo, project_id: uuid.UUID) -> None: + if await project_repo.get(STUB_OWNER_ID, project_id) is None: + raise AppError(ErrorCode.NOT_FOUND, f"project not found: {project_id}") + + +@router.post( + "/{project_id}/chains/{chain_key}/run", + status_code=202, + responses=_RUN_ERRORS, +) +async def run_chain( + project_id: uuid.UUID, + chain_key: str, + body: ChainRunRequest, + request: Request, + response: Response, + background_tasks: BackgroundTasks, + project_repo: ProjectRepoDep, + job_repo: JobRepoDep, + gateway: ChainGatewayDep, # 凭据探测（无凭据 → dep 解析阶段 503） + session: SessionDep, + session_factory: SessionFactoryDep, + checkpointer_factory: CheckpointerFactoryDep, + digest_gateway_builder: DigestBuilderDep, +) -> ChainRunAccepted: + """发起多章链：写一行 job 返 202，链经 BackgroundTask 异步跑。 + + 项目不存在 → 404；未知 chain_key → 404；无凭据 → 503（dep 解析阶段拦下）； + `count` 越界由 schema Field 在解析阶段 → 422。`gateway` 仅做请求阶段凭据探测—— + 链本体在 BackgroundTask 里用独立 session 重建网关跑（请求 session 即将关闭）。 + """ + request_id = getattr(request.state, "request_id", None) + + await _require_project(project_repo, project_id) + if chain_key not in SUPPORTED_CHAINS: + raise AppError(ErrorCode.NOT_FOUND, f"unknown chain key: {chain_key}") + + job = await job_repo.create(project_id, JOB_KIND_CHAIN) + await session.commit() # job 行须在 202 返回前持久化（供前端立即轮询）。 + + accept_op = build_accept_op( + session_factory=session_factory, + digest_gateway_builder=digest_gateway_builder, + request_id=request_id, + ) + background_tasks.add_task( + run_chain_job, + session_factory, + job.id, + project_id=project_id, + user_id=STUB_OWNER_ID, + chain_key=chain_key, + start_chapter_no=body.start_chapter_no, + count=body.count, + gateway=gateway, + checkpointer_ctx=checkpointer_factory, + accept_op=accept_op, + chapter_repo_factory=_chapter_repo_factory, + review_repo_factory=_review_repo_factory, + request_id=request_id, + ) + + log.info( + "chain_run_scheduled", + project_id=str(project_id), + request_id=request_id, + job_id=str(job.id), + chain_key=chain_key, + start_chapter_no=body.start_chapter_no, + count=body.count, + ) + response.status_code = 202 + return ChainRunAccepted( + job_id=job.id, + chain_key=chain_key, + start_chapter_no=body.start_chapter_no, + count=body.count, + ) + + +@router.post( + "/{project_id}/chains/runs/{job_id}/resume", + status_code=202, + responses=_RESUME_ERRORS, +) +async def resume_chain( + project_id: uuid.UUID, + job_id: uuid.UUID, + body: ChainResumeRequest, + request: Request, + response: Response, + background_tasks: BackgroundTasks, + project_repo: ProjectRepoDep, + job_repo: JobRepoDep, + gateway: ChainGatewayDep, + session: SessionDep, + session_factory: SessionFactoryDep, + checkpointer_factory: CheckpointerFactoryDep, + digest_gateway_builder: DigestBuilderDep, +) -> ChainRunAccepted: + """裁决续跑：仅当 job=awaiting_input，带裁决经 BackgroundTask resume。 + + 项目不存在 → 404；job 不存在 → 404；非 awaiting 态 → 409（CONFLICT）。 + resume 经 `Command(resume=decisions)` 从 interrupt 续跑（同 thread_id=job_id）。 + """ + request_id = getattr(request.state, "request_id", None) + + await _require_project(project_repo, project_id) + job = await job_repo.get(job_id) + if job is None: + raise AppError(ErrorCode.NOT_FOUND, f"job not found: {job_id}") + if job.status != STATUS_AWAITING: + raise AppError( + ErrorCode.CONFLICT, + f"job {job_id} 非 awaiting_input 态（当前 {job.status}），不可续跑", + {"job_status": job.status}, + ) + + accept_op = build_accept_op( + session_factory=session_factory, + digest_gateway_builder=digest_gateway_builder, + request_id=request_id, + ) + background_tasks.add_task( + run_chain_job, + session_factory, + job_id, + project_id=project_id, + user_id=STUB_OWNER_ID, + chain_key=job.kind if job.kind in SUPPORTED_CHAINS else "draft_volume", + start_chapter_no=1, # resume 不重置区间：图从检查点续，初值不再使用。 + count=1, + gateway=gateway, + checkpointer_ctx=checkpointer_factory, + accept_op=accept_op, + chapter_repo_factory=_chapter_repo_factory, + review_repo_factory=_review_repo_factory, + resume_decisions=body.decisions, + request_id=request_id, + ) + + log.info( + "chain_resume_scheduled", + project_id=str(project_id), + request_id=request_id, + job_id=str(job_id), + decision_count=len(body.decisions), + ) + response.status_code = 202 + return ChainRunAccepted( + job_id=job_id, + chain_key="draft_volume", + start_chapter_no=0, + count=0, + ) + + +def _chapter_repo_factory(session: AsyncSession) -> object: + """链节点（review 重读草稿）按 session 建 chapter draft repo。""" + from ww_core.domain.chapter_repo import SqlChapterRepo + + return SqlChapterRepo(session) + + +def _review_repo_factory(session: AsyncSession) -> object: + """链节点（review 落留痕 / decide 重读）按 session 建审稿 repo。""" + from ww_core.domain.review_repo import SqlReviewRepo + + return SqlReviewRepo(session) diff --git a/apps/api/ww_api/schemas/chain.py b/apps/api/ww_api/schemas/chain.py new file mode 100644 index 0000000..f3d047c --- /dev/null +++ b/apps/api/ww_api/schemas/chain.py @@ -0,0 +1,54 @@ +"""多章工作流链端点的请求/响应 schema（chain-workflow 设计 §6）。 + +snake_case（命名契约，见 memory/gotchas）。前端经 OpenAPI→TS 客户端消费——改字段 +须 `pnpm gen:api` 重生成。 + +- `ChainRunRequest`：发起一条链（起始章 + 章数）。`count` 1..50（系统边界校验，fail fast）。 +- `ChainRunAccepted`：202 受理回执（job_id + 回显请求参数，供前端轮询 `GET /jobs/{id}`）。 +- `ChainResumeRequest`：裁决续跑（复用 `ConflictDecision`，对 awaiting 章的冲突逐条裁决）。 +""" + +from __future__ import annotations + +import uuid + +from pydantic import BaseModel, Field + +from ww_api.schemas.projects import ConflictDecision + +# 链一次最多写 K 章（系统边界，避免一条 run 无界占用；§6 校验）。 +MAX_CHAIN_COUNT = 50 +MIN_CHAIN_COUNT = 1 + + +class ChainRunRequest(BaseModel): + """发起多章链：从 `start_chapter_no` 起循环写 `count` 章（区间含两端）。 + + `count` 1..50（>50 → 422 VALIDATION，由 Field 约束在边界拦下，不进 BackgroundTask）； + `start_chapter_no >= 1`。 + """ + + start_chapter_no: int = Field(ge=1, description="起始章号（含），>=1") + count: int = Field( + ge=MIN_CHAIN_COUNT, + le=MAX_CHAIN_COUNT, + description="本 run 连续写的章数（1..50，含起始章）", + ) + + +class ChainRunAccepted(BaseModel): + """链受理回执（202）：job_id + 回显请求参数（前端走 `GET /jobs/{id}` 轮询进度/awaiting）。""" + + job_id: uuid.UUID + chain_key: str + start_chapter_no: int + count: int + + +class ChainResumeRequest(BaseModel): + """裁决续跑：对 awaiting 章最近审稿的每个冲突逐条裁决（复用 `ConflictDecision`）。""" + + decisions: list[ConflictDecision] = Field( + default_factory=list, + description="对 awaiting 章冲突的裁决清单（采纳/忽略/手改）", + ) diff --git a/apps/api/ww_api/services/chain_deps.py b/apps/api/ww_api/services/chain_deps.py new file mode 100644 index 0000000..e354466 --- /dev/null +++ b/apps/api/ww_api/services/chain_deps.py @@ -0,0 +1,44 @@ +"""多章链的 checkpointer 注入缝（chain-workflow 设计 §4）——激活休眠资产。 + +`get_checkpointer()` 是可注入缝： +- **运行时**：返 `AsyncPostgresSaver`（`langgraph.checkpoint.postgres.aio`），连 `DATABASE_URL`， + 持久控制流位置 + pending handle（跨小时可恢复）。检查点表由迁移/CI 调 `setup_checkpointer` + 建（C3），本缝**绝不**触碰 DDL（CLAUDE.md「LangGraph」：`setup()` 只在 migrations/CI）。 +- **测试**：经 `app.dependency_overrides[get_checkpointer]` 注 `MemorySaver`（同进程跨两次 + invoke 可恢复 interrupt），绝不连真 Postgres、绝不真 LLM。 + +注：`AsyncPostgresSaver.from_conn_string` 是 async 上下文管理器——其连接生命周期须横跨整条 +链 run（含 interrupt 暂停后的 resume）。故 `run_chain_job` 在自身 task 内打开/关闭 saver +上下文（见 `chain_runner`），本缝只提供「按需建一个 saver 上下文管理器」的工厂。 +""" + +from __future__ import annotations + +from collections.abc import Callable +from contextlib import AbstractAsyncContextManager +from typing import TYPE_CHECKING, Any + +from ww_config import get_settings + +if TYPE_CHECKING: + from langgraph.checkpoint.base import BaseCheckpointSaver + +# checkpointer 工厂：`()` → async 上下文管理器，进入得一个 `BaseCheckpointSaver`。 +# 运行时 = Postgres saver 上下文（连接随上下文开/关）；测试注一个产 MemorySaver 的工厂。 +CheckpointerFactory = Callable[[], AbstractAsyncContextManager[Any]] + + +def get_checkpointer_factory() -> CheckpointerFactory: + """运行时 checkpointer 工厂：建连 `DATABASE_URL` 的 `AsyncPostgresSaver` 上下文。 + + `from_conn_string` 返回的 async-CM 在 `run_chain_job` 内 `async with` 打开——其连接 + 存活整条链（含 interrupt→resume）；run 结束关闭。测试经 dep 覆盖注 MemorySaver 工厂。 + """ + + def _factory() -> AbstractAsyncContextManager[BaseCheckpointSaver[Any]]: + from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver + + # langgraph 的 Postgres saver 走 psycopg(同步驱动串)；用 sync 串建连接。 + return AsyncPostgresSaver.from_conn_string(get_settings().database_url_sync) + + return _factory diff --git a/apps/api/ww_api/services/chain_runner.py b/apps/api/ww_api/services/chain_runner.py new file mode 100644 index 0000000..c04cadc --- /dev/null +++ b/apps/api/ww_api/services/chain_runner.py @@ -0,0 +1,294 @@ +"""多章链长任务 runner（chain-workflow 设计 §5）——`run_chain_job` 在 jobs 壳里跑链图。 + +复用 `run_job` 的生命周期心智（set_running → work → complete/fail），但链多一态： +**interrupt 命中 → `awaiting_input`**（等作者裁决续跑）。故不直接复用 `run_job`，而是 +仿其壳自建独立 session 写 job 状态，work 本体 = 驱动 LangGraph 链图。 + +纪律（守不变量）： +- **#1/#5**：图节点各自短事务落领域表（`chapters`/`chapter_reviews`/`chapter_digests`）； + checkpointer 只存控制流；resume 重读领域表。本 runner 不持业务事务。 +- **#3/#4**：`accept_op`（apps/api 注入）跑冲突 gate → 终稿提炼 digest → 单原子验收事务 → + 伏笔到期扫描，唯一正文写经此。 +- **token 不泄漏**：job result/status/日志只记章号/计数/标志，绝不含 prompt/正文/token。 +- **错误脱敏**：复用 `_classify_job_error`（AppError 透传安全文案，其余通用文案，P0-3）。 + +可测性：`session_factory`/`gateway`/`checkpointer`/`accept_op` 全可注入——单测直接 `await` +它，注 fake session 工厂 + mock 网关 + MemorySaver + fake accept_op（绝不联网/真 DB/真 LLM）。 +""" + +from __future__ import annotations + +import uuid +from collections.abc import Awaitable, Callable, Sequence +from contextlib import AbstractAsyncContextManager +from typing import Any + +import structlog +from langchain_core.runnables import RunnableConfig +from langgraph.types import Command +from ww_core.domain.chapter_repo import ChapterRepo, SqlChapterRepo +from ww_core.domain.digest_repo import SqlDigestAppendRepo +from ww_core.domain.job_repo import SqlJobRepo +from ww_core.domain.review_repo import SqlReviewRepo +from ww_core.memory import assemble +from ww_core.memory.sql_repositories import sql_memory_repos +from ww_core.orchestrator import REVIEW_SPECS, GatewayRun +from ww_core.orchestrator.chain import build_chain_graph, initial_chain_state +from ww_llm_gateway import Gateway + +from ww_api.schemas.projects import ConflictDecision +from ww_api.services.accept_service import ( + assert_conflicts_resolved, + run_accept_transaction, +) +from ww_api.services.digest_extraction import extract_digest_facts +from ww_api.services.foreshadow_scan import SessionFactory, run_overdue_scan +from ww_api.services.job_runner import _classify_job_error + +log = structlog.get_logger(__name__) + +# 链 job 的 kind（jobs.kind 自由 Text 列，零迁移复用；§7）。 +JOB_KIND_CHAIN = "chain" + + +# 「按 session 建 digest 网关」缝：accept 节点在自建短事务里需 light 档网关跑终稿提炼。 +DigestGatewayBuilder = Callable[[Any], Awaitable[Gateway]] + + +def build_accept_op( + *, + session_factory: SessionFactory, + digest_gateway_builder: DigestGatewayBuilder, + request_id: str | None = None, +) -> AcceptOp: + """组装单章验收落库闭包（apps/api，守不变量 #3/#4）。 + + 链 `accept_chapter` 节点经此闭包把作者裁决落库——每章**独立短事务**（自建 session）： + 重读最近审稿 → 冲突 gate（缺判 → `CONFLICT_UNRESOLVED`，链失败）→ 终稿(=已存草稿)提炼 + digest（事务外）→ `run_accept_transaction` 单原子提交 → 提交后伏笔到期扫描（自建 session）。 + + 自动链无作者改稿：`final_text` = `chapters` 里该章草稿正文（write 节点所落）。 + `decisions` 为 dict 清单（interrupt 恢复值 / 无冲突 []），此处校验为 `ConflictDecision`。 + """ + + async def accept_op( + *, + project_id: uuid.UUID, + chapter_no: int, + user_id: uuid.UUID, + decisions: list[Any], + ) -> None: + parsed_decisions = [ + d if isinstance(d, ConflictDecision) else ConflictDecision.model_validate(d) + for d in (decisions or []) + ] + async with session_factory() as session: + chapter_repo: ChapterRepo = SqlChapterRepo(session) + review_repo = SqlReviewRepo(session) + digest_repo = SqlDigestAppendRepo(session) + + history = await review_repo.list_for_chapter(project_id, chapter_no) + latest_review = history[0] if history else None + # 冲突 gate（R5，事务前拦截，不写库；缺判 → CONFLICT_UNRESOLVED 抛，链失败）。 + assert_conflicts_resolved(latest_review, parsed_decisions) + + draft = await chapter_repo.get_draft(project_id, chapter_no) + final_text = draft.content if draft is not None else "" + + gateway = await digest_gateway_builder(session) + # R2：终稿 digest 提炼在开原子事务之前（别在持开事务里跨网络调 LLM）。 + digest_facts = await extract_digest_facts( + gateway, + final_text=final_text, + user_id=user_id, + project_id=project_id, + chapter_no=chapter_no, + ) + await run_accept_transaction( + session=session, + chapter_repo=chapter_repo, + digest_repo=digest_repo, + review_repo=review_repo, + project_id=project_id, + chapter_no=chapter_no, + final_text=final_text, + digest_facts=digest_facts, + latest_review=latest_review, + decisions=parsed_decisions, + ) + # 验收提交后伏笔到期扫描（自建独立 session，§6.2；与单章端点同纪律）。 + await run_overdue_scan( + session_factory, + project_id=project_id, + chapter_no=chapter_no, + request_id=request_id, + ) + + return accept_op + + +# `()` → async-CM，进入得一个 langgraph checkpointer（Postgres 运行时 / MemorySaver 测试）。 +CheckpointerCtx = Callable[[], AbstractAsyncContextManager[Any]] + +#: apps/api 注入的「按 session 建 chapter draft repo」工厂（review 节点重读草稿用）。 +ChapterRepoFactory = Callable[[Any], Any] +#: apps/api 注入的「按 session 建审稿留痕 repo」工厂（review/decide 节点用）。 +ReviewRepoFactory = Callable[[Any], Any] +#: 单章验收落库闭包（apps/api 组装：冲突 gate + digest + 验收事务 + 到期扫描）。 +AcceptOp = Callable[..., Awaitable[None]] + + +def _chain_result(state: dict[str, Any], *, awaiting_chapter: int | None) -> dict[str, Any]: + """从图 state 取非密进度摘要（绝不含正文/token；§5）。 + + `awaiting_chapter` 非 None = interrupt 命中该章待裁决；None 且 completed = 全跑完。 + """ + written = list(state.get("written") or []) + return { + "chain_key": state.get("chain_key"), + "written": written, + "completed": awaiting_chapter is None, + "awaiting_chapter": awaiting_chapter, + } + + +def _extract_awaiting_chapter(final: dict[str, Any]) -> int | None: + """从图返回值判 interrupt：有 `__interrupt__` → 取暂停章号；否则 None（跑完）。 + + interrupt 载荷 = `{"chapter_no": n}`（见链图 `_accept` 节点）。LangGraph 把它放在 + `final["__interrupt__"]`（Interrupt 对象列表）。容错读 `.value`/dict 两种形。 + """ + interrupts = final.get("__interrupt__") + if not interrupts: + return None + first = interrupts[0] + payload = getattr(first, "value", first) + if isinstance(payload, dict): + chapter_no = payload.get("chapter_no") + return int(chapter_no) if chapter_no is not None else None + return None + + +async def run_chain_job( + session_factory: SessionFactory, + job_id: uuid.UUID, + *, + project_id: uuid.UUID, + user_id: uuid.UUID, + chain_key: str, + start_chapter_no: int, + count: int, + gateway: GatewayRun, + checkpointer_ctx: CheckpointerCtx, + accept_op: AcceptOp, + chapter_repo_factory: ChapterRepoFactory, + review_repo_factory: ReviewRepoFactory, + resume_decisions: list[ConflictDecision] | None = None, + review_specs: Sequence[Any] = REVIEW_SPECS, + request_id: str | None = None, +) -> None: + """跑/续一条多章链：set_running → 驱动链图 → 据返回置 done/awaiting_input/failed。 + + `resume_decisions is None` → 初始 run（`graph.ainvoke(initial_state, config)`）； + 否则 resume（`graph.ainvoke(Command(resume=decisions), config)`），从 interrupt 续跑。 + `thread_id = str(job_id)`：同 job 的初始/续跑落同一检查点 thread。 + + checkpointer 上下文（Postgres 连接 / MemorySaver）横跨整次 invoke；本 runner 在 task + 内打开/关闭它。异常一律被吞（后台任务边界），脱敏后置 job failed，不冒泡崩进程。 + """ + try: + async with checkpointer_ctx() as checkpointer: + await _set_running(session_factory, job_id) + config: RunnableConfig = {"configurable": {"thread_id": str(job_id)}} + graph = build_chain_graph( + gateway, + session_factory=session_factory, + memory_repos_factory=sql_memory_repos, + chapter_repo_factory=chapter_repo_factory, + review_repo_factory=review_repo_factory, + assemble=assemble, + accept_op=accept_op, + review_specs=review_specs, + checkpointer=checkpointer, + ) + if resume_decisions is None: + initial = initial_chain_state( + project_id=project_id, + user_id=user_id, + start_chapter_no=start_chapter_no, + count=count, + chain_key=chain_key, + ) + raw_final = await graph.ainvoke(initial, config=config) + else: + resume_value = [d.model_dump() for d in resume_decisions] + raw_final = await graph.ainvoke(Command(resume=resume_value), config=config) + + final: dict[str, Any] = dict(raw_final) + awaiting_chapter = _extract_awaiting_chapter(final) + result = _chain_result(final, awaiting_chapter=awaiting_chapter) + await _finish(session_factory, job_id, result, awaiting=awaiting_chapter is not None) + log.info( + "chain_job_settled", + job_id=str(job_id), + request_id=request_id, + awaiting_chapter=awaiting_chapter, + written_count=len(result["written"]), + ) + except Exception as exc: # noqa: BLE001 — 后台任务边界：记错误 + 置 job failed，不冒泡。 + log.error("chain_job_failed", job_id=str(job_id), request_id=request_id, error=str(exc)) + await _mark_failed(session_factory, job_id, _classify_job_error(exc), request_id) + + +async def _set_running(session_factory: SessionFactory, job_id: uuid.UUID) -> None: + """独立短事务把 job 置 running（图驱动开始前）。""" + async with session_factory() as session: + await SqlJobRepo(session).set_running(job_id) + await session.commit() + + +async def _finish( + session_factory: SessionFactory, + job_id: uuid.UUID, + result: dict[str, Any], + *, + awaiting: bool, +) -> None: + """据是否 interrupt 把 job 置 awaiting_input（待裁决）或 done（跑完），独立短事务。""" + async with session_factory() as session: + repo = SqlJobRepo(session) + if awaiting: + await repo.set_awaiting(job_id, result) + else: + await repo.complete(job_id, result) + await session.commit() + + +async def _mark_failed( + session_factory: SessionFactory, + job_id: uuid.UUID, + error: str, + request_id: str | None, +) -> None: + """在**全新** session 里把 job 置 failed（前一事务已因异常作废）。""" + try: + async with session_factory() as session: + await SqlJobRepo(session).fail(job_id, error) + await session.commit() + except Exception as exc: # noqa: BLE001 — 置失败态本身再炸只记日志，不冒泡。 + log.error( + "chain_job_fail_mark_failed", + job_id=str(job_id), + request_id=request_id, + error=str(exc), + ) + + +__all__ = [ + "JOB_KIND_CHAIN", + "AcceptOp", + "CheckpointerCtx", + "DigestGatewayBuilder", + "build_accept_op", + "run_chain_job", +] diff --git a/apps/api/ww_api/services/project_deps.py b/apps/api/ww_api/services/project_deps.py index 8637f10..befc0eb 100644 --- a/apps/api/ww_api/services/project_deps.py +++ b/apps/api/ww_api/services/project_deps.py @@ -10,7 +10,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable -from typing import Annotated +from typing import Annotated, Any import httpx from fastapi import Depends @@ -399,6 +399,53 @@ async def build_writer_gateway(session: AsyncSession, store: CredentialStore) -> return await build_gateway_for_tier(session, store, "writer") +# 多章链一条 run 内跨 writer(write)/analyst(review)/light(digest) 三档——单档网关的 +# chain_resolver 恒返该档链、忽略请求 tier，会把 review/digest 错路由到 writer。故链需一个 +# **按请求 tier 分派**的网关：并 union 三档所有 provider 的适配器 + 逐 tier 解析其链。 +_CHAIN_TIERS: tuple[Tier, ...] = ("writer", "analyst", "light") + + +def _chain_for_tier(stored_by_tier: dict[Tier, Any], tier: Tier) -> list[Route]: + """据该 tier 的 DB 路由行构链；无行则退回全局默认（单 provider，M1 兼容）。""" + stored = stored_by_tier.get(tier) + if stored is None: + return [resolve_route(tier)] + primary_spec = f"{stored.provider}:{stored.model}" + return chain_from_routing(tier, primary_spec, list(stored.fallback)) + + +async def build_chain_gateway(session: AsyncSession, store: CredentialStore) -> Gateway: + """装配多章链网关：按请求 tier（writer/analyst/light）分派，union 三档适配器 + 回退链。 + + 与 `build_gateway_for_tier` 同凭据/适配器装配，但 `chain_resolver` 据 `req.tier` 返回 + 对应档的链（不锁死单档）。至少一个可用适配器，否则 `LLM_UNAVAILABLE`（无任何凭据）。 + """ + ledger = SqlAlchemyLedgerSink(session) + routing_rows = await store.list_routing() + stored_by_tier: dict[Tier, Any] = {r.tier: r for r in routing_rows if r.tier in _CHAIN_TIERS} + + # union 三档链里所有 provider，逐个建可用适配器（未配凭据/未知 base_url 的跳过）。 + adapters: dict[str, ProviderAdapter] = {} + for tier in _CHAIN_TIERS: + for route in _chain_for_tier(stored_by_tier, tier): + if route.provider in adapters: + continue + built = await _build_provider_adapter(store, route.provider) + if built is not None: + adapters[route.provider] = built + if not adapters: + raise AppError( + ErrorCode.LLM_UNAVAILABLE, + "多章链无任何已配置凭据的 provider，请先在设置中配置", + {"tiers": list(_CHAIN_TIERS)}, + ) + + def _resolver(tier: Tier) -> list[Route]: + return _chain_for_tier(stored_by_tier, tier) + + return Gateway(adapters=adapters, ledger=ledger, chain_resolver=_resolver) + + async def get_writer_gateway( session: Annotated[AsyncSession, Depends(get_session)], ) -> Gateway: @@ -451,3 +498,35 @@ async def get_refine_gateway( """回炉（writer 档位）的可注入网关缝。测试经 override 注 mock（产纯文本重写段）。""" store = SqlCredentialStore(session) return await build_gateway_for_tier(session, store, "writer") + + +async def get_chain_gateway( + session: Annotated[AsyncSession, Depends(get_session)], +) -> Gateway: + """多章链（write writer + review analyst + digest light，按请求 tier 分派）的可注入网关缝。 + + `POST .../chains/{key}/run` 在 dep 解析阶段构建网关 → 无凭据时这里抛 `LLM_UNAVAILABLE` + （503，调度 job 之前拦下，避免凭空写注定失败的 job）。链本体在 BackgroundTask 里用 + `run_chain_job` 自建独立 session 重新构网关跑（请求 session 已关闭）；本依赖只做请求阶段 + 凭据探测。测试经 `app.dependency_overrides[get_chain_gateway]` 注 mock（绝不联网）。 + """ + store = SqlCredentialStore(session) + return await build_chain_gateway(session, store) + + +# 「按 session 建 digest 网关」缝类型（chain accept_op 终稿提炼按节点自建短事务的 session 建）。 +GatewayDigestBuilder = Callable[[AsyncSession], Awaitable[Gateway]] + + +def _digest_gateway_builder(session: AsyncSession) -> Awaitable[Gateway]: + """链 accept 节点在自建短事务里建 light 档 digest 网关（终稿提炼）。""" + return build_gateway_for_tier(session, SqlCredentialStore(session), "light") + + +def get_digest_gateway_builder() -> GatewayDigestBuilder: + """返回「按 session 建 digest 网关」的缝（chain accept_op 终稿提炼用）。 + + BackgroundTask 自建短事务 → 节点内据该 session 现建 light 档网关。测试经 + `app.dependency_overrides[get_digest_gateway_builder]` 注返回 mock 的 builder（绝不联网）。 + """ + return _digest_gateway_builder diff --git a/memory/contracts.md b/memory/contracts.md index df1198e..b552439 100644 --- a/memory/contracts.md +++ b/memory/contracts.md @@ -313,9 +313,24 @@ --- +## C-Chain · 多章工作流链端点 + 服务层　owner @backend / @llm(图)　状态: 稳定（C2, 2026-06-23） +> 设计真源：`docs/design/chain-workflow.md`。C1(@llm) 图签名 §3.3；C2(@backend) 端点/schema/服务接线。 +- **3 端点**（`routers/chain.py`，挂 `/projects` 前缀）： + - `POST /projects/{pid}/chains/{chain_key}/run` ← `ChainRunRequest{start_chapter_no:int>=1, count:int 1..50}` → 202 `ChainRunAccepted{job_id, chain_key, start_chapter_no, count}`。写一行 `jobs(queued, kind="chain")` 返 202，链经 BackgroundTask `run_chain_job` 跑（自建独立 session）。未知 chain_key→404（仅 `draft_volume`）；count 越界→422（schema Field）；项目不存在→404；无凭据→503。 + - `GET /jobs/{job_id}`（**复用**现有）：轮询进度/awaiting。链 job `result = {chain_key, written:[int], completed:bool, awaiting_chapter:int|null}`；interrupt 命中时 `status="awaiting_input"` + `result.awaiting_chapter`。 + - `POST /projects/{pid}/chains/runs/{job_id}/resume` ← `ChainResumeRequest{decisions:[ConflictDecision]}` → 202 `ChainRunAccepted`。仅当 job=`awaiting_input`；非该态→**409 `CONFLICT`**；job 不存在→404。带裁决 `Command(resume=...)` 续跑。 +- **schema**（`schemas/chain.py`）：`ChainRunRequest`/`ChainRunAccepted`/`ChainResumeRequest`；`ConflictDecision` 复用 `schemas/projects.py`（conflict_index/verdict/note）。 +- **零迁移**（§7）：复用 `jobs`（kind/status/progress/result 全已存）；新增 `status` 值 `"awaiting_input"`（jobs.status 自由 Text 列，无 DDL）；awaiting 章经 `result.awaiting_chapter` 表达。`JobRepo` 加 `set_awaiting(job_id, result)`。**唯一新增 DDL = langgraph 检查点表（C3 迁移建，本任务未引入）。** +- **新增错误码**（`ww_shared/errors.py`）：`ErrorCode.CONFLICT`→409（资源状态冲突，如对非 awaiting 链 job 续跑；与 `CONFLICT_UNRESOLVED` 区分）。 +- **新网关缝**（`project_deps.py`）：`get_chain_gateway`（`build_chain_gateway`：按请求 tier writer/analyst/light 分派、union 三档适配器+回退链——单档网关 chain_resolver 恒返该档会错路由 review/digest，故链需多档分派）；`get_digest_gateway_builder`（accept 节点自建短事务建 light 档 digest 网关）。`get_checkpointer_factory`（`services/chain_deps.py`：运行时 `AsyncPostgresSaver.from_conn_string(database_url_sync)` 上下文；测试注 MemorySaver 工厂）。 +- **token 纪律**：job result/status/日志只记章号/计数/标志，绝不含 prompt/正文/token（§5，已断言）。 +- → 影响 @frontend（链发起页/进度/裁决续跑面板，**非目标本期不做**，契约稳定后 follow-up `pnpm gen:api`）；@db/@devops C3（检查点建表迁移 + 可选 jobs 注释）。 + ## 契约变更日志（append-only） > 格式：`- [date] @skill 改 Cx：<改了什么> → 影响 <依赖方/任务>` +- [2026-06-23] @backend 立 C-Chain（C2）：3 端点 + `schemas/chain.py` + `services/{chain_runner,chain_deps}.py` + `jobs` 零迁移复用（加 `status="awaiting_input"` + `JobRepo.set_awaiting`）+ 新 `ErrorCode.CONFLICT`(409) + `project_deps` 加 `build_chain_gateway`/`get_chain_gateway`/`get_digest_gateway_builder`/`get_checkpointer_factory`。OpenAPI 含 2 新 POST（GET /jobs 复用）。门禁绿：ruff/format/mypy(193) 干净 + pytest 600 passed + alembic 无漂移。→ 影响 @frontend（follow-up `pnpm gen:api`）、@db/@devops C3（检查点迁移）。 + - [2026-06-20] @backend 扩 C3：新增 **`GET /projects/{project_id}/chapters/{chapter_no}/injection`**（本章注入透明，B0 读端点）→ `InjectionResponse{project_id, chapter_no, selected:[InjectionEntity{kind,name,reasons[]}], recent_n}`（`schemas/injection.py`，snake_case）。实现仅调既有 `assemble()` 回放确定性 `SelectionTrace`——**无 LLM、无 commit、无 DDL**；项目不存在→404，无大纲→`selected:[]`。reasons 取值同 `SelectionReason`（explicit_beat/main_character/recent_digest/foreshadow_window）。→ 影响 @frontend（已 `pnpm gen:api` + `ChapterAssistant` 消费）。**B0 可控版（PUT override + `select_relevant_entities` 加 pinned/excluded/recent_n + draft 端点同读 override + 持久化）尚未实现，届时再扩本契约。** - [2026-06-20] @backend 再扩 C3（B0 **可控版**，接上条）：新增 **`PUT /projects/{project_id}/chapters/{chapter_no}/injection`** ← `InjectionOverrideRequest{pinned:[{kind,name}], excluded:[{kind,name}], recent_n:int|null(1..20)}` → 返回 `InjectionResponse`（同上，**新增回显字段 `pinned`/`excluded`**，`selected[].reasons` 可含新值 **`author_pin`**）。语义：pin 强制纳入并加 `author_pin` 理由 / excluded 强制剔除（**优先于 pin**）/ recent_n 覆盖近况回看章数。GET injection 与 **draft 流式端点**均先读同一覆盖再 `assemble(override=...)`，故「看到的=写章用的」（不变量 #6 作者兜底）。持久化：**新表 `chapter_injection`**（迁移 `ad2c4c663daf`，唯一 `(project_id,chapter_no)`，复用 outline 行已否决）；`select_relevant_entities` 加 `pinned/excluded` frozenset 入参、`assemble` 加 `override` 关键字参；新 `domain/injection_repo.py`（`InjectionOverride`/`EntityRef`/`SqlInjectionOverrideRepo`，upsert 只 flush 端点 commit）。→ 影响 @frontend（已 `pnpm gen:api`；F1 加 pin/排除控件 + recent_n 步进器）。 diff --git a/memory/decisions.md b/memory/decisions.md index 4f72207..54d6c13 100644 --- a/memory/decisions.md +++ b/memory/decisions.md @@ -149,3 +149,9 @@ - 选择：把实现落在已是 workspace member、且承载「编排/写库 apply 层」的 `ww_core.domain`（`skill_registry.py` + `skill_permissions.py`，经 `ww_core.domain` 导出）。`ARCHITECTURE §5.6` 明确「强制点在编排/写库层」=ww_core，语义上也吻合。`packages/skills/ww_skills/__init__.py` 留**再导出门面**（import 自 ww_core），保留稳定「skills」导入名。 - 理由 / 取舍：非阻塞、门禁可干净覆盖（ruff/mypy/pytest 全跑）、不越权改根配置；代价是物理位置与任务文档措辞略偏，但功能/契约（C8）完整。待 @devops 把 `packages/skills` 纳入 workspace + mypy_path 后，可平滑迁移实现到 `ww_skills` 而不动调用方（只改门面方向）。 - 影响：消费方 import `ww_core.domain` 的 `SkillRegistry`/`filter_reads`/`partition_writes`/`validate_declaration`（或经 `ww_skills` 门面）。若 @devops 后续纳入 workspace，迁移由届时的 @backend/@llm 执行。 + +## [2026-06-23] @backend C2（多章链）服务/端点接线：零迁移表达 awaiting + 多档分派网关 +- 决策 1（**零迁移表达 awaiting**）：链 interrupt 暂停态不新建 jobs 列/枚举——`jobs.status` 是自由 Text 列（无 DB CHECK），直接新增值 `"awaiting_input"` + `JobRepo.set_awaiting`；待裁决章号经 `result.awaiting_chapter` 表达。`jobs.kind="chain"` 同理零迁移。唯一新增 DDL（langgraph 检查点表）归 C3 迁移。理由：设计 §7 优先零迁移；result 已是 JSONB 足以承载非密进度。 +- 决策 2（**多档分派网关 `build_chain_gateway`**）：一条链 run 跨 writer(write)/analyst(review)/light(digest) 三档，但 `build_gateway_for_tier(tier)` 装出的网关 `chain_resolver` 恒返该单档链、忽略 `req.tier`——会把 review/digest 错路由到 writer。故链需一个**按请求 tier 分派**的网关：union 三档所有 provider 适配器 + `chain_resolver` 据 `req.tier` 返回对应档链。digest 在 accept 节点自建短事务内按 session 现建 light 档网关（`get_digest_gateway_builder`）。 +- 决策 3（**accept_op 在 apps/api 装配**）：链编排（图/节点）在 `ww_core`，但具体验收事务（`run_accept_transaction`/冲突 gate/digest 提炼/伏笔到期扫描）属 apps/api——经 `build_accept_op` 闭包注入图节点，守目录所有权 + 不变量 #3/#4。自动链无作者改稿，`final_text` = `chapters` 该章草稿正文（write 节点所落）。 +- 决策 4（**新错误码 `CONFLICT`**）：resume 非 awaiting 态须 409，但既有 409 码 `CONFLICT_UNRESOLVED` 语义专指「未决冲突禁验收」，不宜复用。新增通用 `ErrorCode.CONFLICT`→409（资源状态冲突）。 diff --git a/memory/gotchas.md b/memory/gotchas.md index 16932f4..c691a03 100644 --- a/memory/gotchas.md +++ b/memory/gotchas.md @@ -84,3 +84,6 @@ - [2026-06-19] @llm **Kimi Code coding API 伪造头真源 = `ooojustin/opencode-kimi`（`src/headers.ts`+`src/constants.ts`，1:1 镜像 kimi-cli v1.37.0），发完整 7 头、UA `KimiCLI/1.37.0`**（K1.2 校正，推翻上方 pi-kimi-coder UA-only 那条）：`picassio/pi-kimi-coder` 是分歧/错误参考（只发 UA），照它裁成 UA-only 是误修，已回退。`kimi_code_headers()` 发 `User-Agent=KimiCLI/1.37.0` + `X-Msh-Platform=kimi_cli`（字面常量，非 OS 名）+ `X-Msh-Version=1.37.0`（==UA 版本）+ `X-Msh-Device-Name`（`socket.gethostname()` ASCII 化）+ `X-Msh-Device-Model`（macOS=`f"macOS {platform.mac_ver()[0]} {platform.machine()}"`，`machine()` 原样 `arm64`/`x86_64` 不归一化）+ `X-Msh-Os-Version`（`platform.version()`）+ `X-Msh-Device-Id`（稳定 32 位无连字符 `uuid4().hex`：env `KIMI_DEVICE_ID` 优先→否则读/建 `~/.kimi/device_id` 复用，**绝不每次随机**，否则 Kimi 侧每次当新设备）。源码若与此有出入以源码为准（本次核对 master 一致）。⚠️ HTTP 头值含非 ASCII 会被底层 fetch/httpx 拒，host 派生值（Device-Name/Model/Os-Version）须 ASCII 化（`asciiHeaderValue`：裁 `\x20-\x7e` 之外 + trim，空回退 `unknown`）。单测用 env `KIMI_DEVICE_ID` 注入固定 id 保 CI 确定（不触真 ~/.kimi）。ruff E501 按显示宽度算（中文占 2 列），含中文 docstring/注释的行别贴边 100。 - [2026-06-19] @backend `assemble`（C5）此前只从 world_entities/characters/style/rules + cards/foreshadow/digests/outline.beats 组装，**从不含项目 premise/logline/theme/title，也无「写章」指令** → 全新项目（只有 premise、无世界观/角色/大纲）产出 `stable_core=""`+`volatile=""`，writer 的 user message 为空，LLM 回 `400 "the message at position 0 with role 'user' must not be empty"`。修复：`MemoryRepos` 加第 8 个 repo `project:ProjectSpecRepo`（`spec(project_id)->ProjectSpecView{title,logline?,premise?,theme?}`，按 project_id 读 `projects`，无 owner 过滤——owner 隔离归路由层）；`assemble` 把「作品蓝本」放进 `stable_core` 首段（书级 spec=定型→缓存前缀，不违 #9），并让 `volatile` 始终含 `请创作第 {chapter_no} 章的正文。`（易变指令）。⚠️ 任何手搓 `MemoryRepos(...)` 的地方（含测试 fake）现在**必须**补 `project=` 字段，否则 dataclass 缺参报错。无 DDL 变更（只读既有 projects 列），无迁移。 - [2026-06-22] @frontend/@orchestrator T6 创作工具箱通用端点的两处契约边界（加新生成器时注意）：① `ToolGenerateRequest` 是**固定字段并集** `{brief, chapter_no?, count?, kind?}`——`GeneratorRunner` 只映射落在此集合内的 `input_field` 名；新生成器若声明了集合外的 `input_field`，前端会**静默忽略**该字段。新工具的 `input_fields` 须落在 `brief/chapter_no/count/kind` 内，否则需先扩 `ToolGenerateRequest` 契约（+`pnpm gen:api`）。② `ToolInputFieldView.type` 是自由字符串，后端可声明 `"select"` 但 descriptor **无 `options` 字段** → 前端把 select 当 text 输入渲染。若某工具需要真下拉选项，须先给 `InputField`/`ToolInputFieldView` 加 `options` 再改前端 `GeneratorRunner`。 +- [2026-06-23] @backend 多章链网关（C2）：**单档 `build_gateway_for_tier(tier)` 网关不能驱动跨档链**。其 `chain_resolver`/`_resolver` 闭包恒返该 tier 的链、**忽略入参 `req.tier`**，故拿它跑 review(analyst)/digest(light) 会全部错路由到 writer 档的 provider/model。链/任何跨档编排须用 `build_chain_gateway`（union 三档适配器 + `chain_resolver` 据 `req.tier` 分派）。测试用 mock 网关按 `req.output_schema` 路由（无 schema=write，有=review），不受此影响。 +- [2026-06-23] @backend langgraph `CompiledStateGraph.ainvoke` 的 mypy 重载：`config` 必须是 `RunnableConfig`（`from langchain_core.runnables import RunnableConfig`），裸 `dict[str,dict[str,str]]` 不匹配任何重载 → call-overload 错。返回值是 `dict[str,Any] | Any`，传给取 `__interrupt__`/`written` 的 helper 前先 `dict(raw_final)` 收敛为 `dict[str,Any]`。interrupt 命中时返回值含 `__interrupt__`（Interrupt 对象列表），载荷读 `getattr(first,"value",first)`（兼容 dict）。 +- [2026-06-23] @backend 改 `JobRepo` Protocol 加方法（如 `set_awaiting`）= 契约变更：**所有 fake 实现都要同步补**，否则 mypy 报 Protocol 缺成员（本次漏了 `packages/core/tests/test_job_repo.py::FakeJobRepo` + `apps/api/tests/fakes_projects.py::FakeJobRepo`）。grep `class Fake.*JobRepo` 全仓再补。 diff --git a/packages/core/tests/test_job_repo.py b/packages/core/tests/test_job_repo.py index 3cc41e6..1c32866 100644 --- a/packages/core/tests/test_job_repo.py +++ b/packages/core/tests/test_job_repo.py @@ -15,6 +15,7 @@ import pytest from pydantic import ValidationError from ww_core.domain.job_repo import ( PROGRESS_COMPLETE, + STATUS_AWAITING, STATUS_DONE, STATUS_FAILED, STATUS_QUEUED, @@ -93,6 +94,12 @@ class FakeJobRepo: row.result = dict(result) return _view(row) + async def set_awaiting(self, job_id: uuid.UUID, result: dict[str, Any]) -> JobView: + row = self._require(job_id) + row.status = STATUS_AWAITING + row.result = dict(result) + return _view(row) + async def fail(self, job_id: uuid.UUID, error: str) -> JobView: row = self._require(job_id) row.status = STATUS_FAILED diff --git a/packages/core/ww_core/domain/job_repo.py b/packages/core/ww_core/domain/job_repo.py index 820c165..bdfcfb3 100644 --- a/packages/core/ww_core/domain/job_repo.py +++ b/packages/core/ww_core/domain/job_repo.py @@ -28,6 +28,8 @@ STATUS_QUEUED = "queued" STATUS_RUNNING = "running" STATUS_DONE = "done" STATUS_FAILED = "failed" +# 多章链 interrupt 命中：暂停等作者裁决（chain-workflow §7，零迁移复用 jobs.status 自由 Text 列）。 +STATUS_AWAITING = "awaiting_input" PROGRESS_COMPLETE = 100 @@ -63,6 +65,10 @@ class JobRepo(Protocol): """置 status=done, progress=100, result=。""" ... + async def set_awaiting(self, job_id: uuid.UUID, result: dict[str, Any]) -> JobView: + """置 status=awaiting_input, result=（多章链 interrupt 暂停等裁决；chain §5/§7）。""" + ... + async def fail(self, job_id: uuid.UUID, error: str) -> JobView: """置 status=failed, error=。""" ... @@ -144,6 +150,14 @@ class SqlJobRepo: await self._s.refresh(row) return _to_view(row) + async def set_awaiting(self, job_id: uuid.UUID, result: dict[str, Any]) -> JobView: + row = await self._require(job_id) + row.status = STATUS_AWAITING + row.result = dict(result) + await self._s.flush() + await self._s.refresh(row) + return _to_view(row) + async def fail(self, job_id: uuid.UUID, error: str) -> JobView: row = await self._require(job_id) row.status = STATUS_FAILED diff --git a/packages/shared/ww_shared/errors.py b/packages/shared/ww_shared/errors.py index d243b8e..1d7eb8f 100644 --- a/packages/shared/ww_shared/errors.py +++ b/packages/shared/ww_shared/errors.py @@ -8,6 +8,7 @@ from enum import StrEnum class ErrorCode(StrEnum): NOT_FOUND = "NOT_FOUND" VALIDATION = "VALIDATION" + CONFLICT = "CONFLICT" # 资源状态冲突（如对非 awaiting 链 job 续跑） CONFLICT_UNRESOLVED = "CONFLICT_UNRESOLVED" # 未决冲突禁验收 LLM_UNAVAILABLE = "LLM_UNAVAILABLE" # 回退耗尽 RATE_LIMITED = "RATE_LIMITED" @@ -18,6 +19,7 @@ class ErrorCode(StrEnum): HTTP_STATUS: dict[ErrorCode, int] = { ErrorCode.NOT_FOUND: 404, ErrorCode.VALIDATION: 422, + ErrorCode.CONFLICT: 409, ErrorCode.CONFLICT_UNRESOLVED: 409, ErrorCode.LLM_UNAVAILABLE: 503, ErrorCode.RATE_LIMITED: 429,