d1ea83b191
- CRITICAL #1：链 write/review 节点经 gateway_builder 按节点自建 session 现建网关，
usage_ledger sink 绑活 session，随节点 commit 持久化；run_chain_job 不再转发请求网关
（其 session 在 BackgroundTask 跑时已关闭，记账行被静默丢弃）。新增 get_chain_gateway_builder
缝（仿 digest builder），get_chain_gateway 退化为纯 503 凭据预检。守不变量 #1。
- HIGH #2：chain_runner 失败日志不再记 str(exc)（可能含 key/连接串/LLM 输出），改记
_classify_job_error 脱敏文案 + exc_type（设计 §5）。
- HIGH #3：resume 端点原子抢占 awaiting→running（JobRepo.claim_awaiting_to_running 条件
UPDATE），抢不到 → 409，防并发 resume 双 Command(resume) 损坏图。
- HIGH #4：resume 校验 job.project_id == project_id（JobView 新增 project_id），不匹配 → 404。
- HIGH #5：resume 返回新 ChainResumeAccepted{job_id,chain_key}，去掉无意义哨兵 start/count=0。
- HIGH #6：删 nodes.py 死导入 extract_conflicts（import + __all__）。
- 测试 #7：e2e 断言链跑后 usage_ledger 有行（#1 回归守卫）+ chapter_reviews 每章一行；
新增 claim 原子抢占单测 + resume 跨项目 404 / 并发 409 端点测。
265 lines
10 KiB
Python
265 lines
10 KiB
Python
"""多章工作流链端点（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 (
|
||
ChainResumeAccepted,
|
||
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 (
|
||
GatewayChainBuilder,
|
||
GatewayDigestBuilder,
|
||
get_chain_gateway,
|
||
get_chain_gateway_builder,
|
||
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)]
|
||
ChainBuilderDep = Annotated[GatewayChainBuilder, Depends(get_chain_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）；不用于实际 LLM 调用
|
||
session: SessionDep,
|
||
session_factory: SessionFactoryDep,
|
||
checkpointer_factory: CheckpointerFactoryDep,
|
||
digest_gateway_builder: DigestBuilderDep,
|
||
chain_gateway_builder: ChainBuilderDep,
|
||
) -> 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,
|
||
chain_gateway_builder=chain_gateway_builder,
|
||
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, # 凭据探测（无凭据 → dep 解析阶段 503）；不用于实际 LLM 调用
|
||
session: SessionDep,
|
||
session_factory: SessionFactoryDep,
|
||
checkpointer_factory: CheckpointerFactoryDep,
|
||
digest_gateway_builder: DigestBuilderDep,
|
||
chain_gateway_builder: ChainBuilderDep,
|
||
) -> ChainResumeAccepted:
|
||
"""裁决续跑：仅当 job=awaiting_input，带裁决经 BackgroundTask resume。
|
||
|
||
项目不存在 → 404；job 不存在 / 不属于该项目 → 404（同案，不枚举）；非 awaiting 态 →
|
||
409（CONFLICT）。`awaiting→running` 在 HTTP 处理器内**原子抢占**（条件 UPDATE），避免两个
|
||
并发 resume 都过守卫后双 `Command(resume=...)` 损坏图（审评 #3）；抢不到（已被并发抢走/非
|
||
awaiting）→ 409。resume 经 `Command(resume=decisions)` 从 interrupt 续跑（同 thread_id）。
|
||
"""
|
||
request_id = getattr(request.state, "request_id", None)
|
||
|
||
await _require_project(project_repo, project_id)
|
||
job = await job_repo.get(job_id)
|
||
# 所有权校验（审评 #4）：job 不存在 **或** 不属于该项目 → 404（同案，避免跨项目枚举 job）。
|
||
if job is None or job.project_id != project_id:
|
||
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},
|
||
)
|
||
|
||
# 原子抢占 awaiting→running（审评 #3）：抢不到（并发竞态/已非 awaiting）→ 409，绝不调度。
|
||
claimed = await job_repo.claim_awaiting_to_running(job_id)
|
||
await session.commit() # 抢占须在 202 返回前持久化（防并发 resume 双调度）。
|
||
if claimed is None:
|
||
raise AppError(
|
||
ErrorCode.CONFLICT,
|
||
f"job {job_id} 已被并发续跑抢占或非 awaiting_input 态，不可重复续跑",
|
||
)
|
||
|
||
chain_key = claimed.kind if claimed.kind in SUPPORTED_CHAINS else "draft_volume"
|
||
|
||
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=1, # resume 不重置区间：图从检查点续，初值不再使用。
|
||
count=1,
|
||
chain_gateway_builder=chain_gateway_builder,
|
||
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 ChainResumeAccepted(job_id=job_id, chain_key=chain_key)
|
||
|
||
|
||
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)
|