499 lines
22 KiB
Python
499 lines
22 KiB
Python
"""圆桌后台作业控制 API（Phase 3a）。
|
||
|
||
把 Step 2「后台挂起」的任务交给后端编排引擎跑，并暴露进度订阅 / 续跑 / 取消。
|
||
|
||
Routes（prefix ``/api/roundtable-jobs``）::
|
||
POST "" 启动一个后台作业（创建 job 行 + 拉起执行器任务）
|
||
GET ""?draft_id= 查某草稿最近一个作业
|
||
GET "/{job_id}" 作业快照
|
||
GET "/{job_id}/stream" SSE 进度流（轮询 job 表，多 worker 安全）
|
||
POST "/{job_id}/resume" 回答澄清后续跑
|
||
POST "/{job_id}/cancel" 取消
|
||
|
||
进度数据源是 ``roundtable_jobs`` 表（Phase 1），SSE 走 DB 轮询而非内存 pub/sub —— 这样
|
||
刷新 / 换 worker / 重连都能继续收到进度，规避 ai_writing 踩过的「per-worker 内存 dict
|
||
丢会话」坑。响应统一 camelCase（对齐前端 §2.5）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import time
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from fastapi import APIRouter, HTTPException, Request, Response
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app.gateway.roundtable_job_executor import build_start_snapshot
|
||
from deerflow.agents.roundtable_orchestrator import SeatRef
|
||
from deerflow.runtime.user_context import get_effective_user_id
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter(prefix="/api/roundtable-jobs", tags=["roundtable-jobs"])
|
||
|
||
_TERMINAL = {"done", "error", "cancelled"}
|
||
# SSE 轮询间隔与最长存活（对齐 ai_writing：长连接最多挂 30 分钟）。
|
||
_POLL_INTERVAL_S = 0.7
|
||
_MAX_STREAM_TICKS = int(30 * 60 / _POLL_INTERVAL_S)
|
||
|
||
# 续跑澄清时拼给 leader 的 prompt 前缀。
|
||
_RESUME_PREFIX = "用户已回复你刚才的澄清问题，回答如下，请据此继续派活或给出结论：\n"
|
||
|
||
|
||
# ── schemas ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class JobAgent(BaseModel):
|
||
agentId: str
|
||
name: str = ""
|
||
|
||
|
||
class StartJobRequest(BaseModel):
|
||
# Phase 3 客户端幂等键：前端为每次「后台挂起」生成一个 requestId，网络重试**复用同一
|
||
# requestId**。服务端据此审计 / 排查；「同草稿不双跑」本身由 active_dedupe_key 唯一索引
|
||
# 保证（requestId 只是锦上添花的请求级指纹）。缺省（旧前端）→ 服务端生成一个。
|
||
requestId: str | None = Field(default=None, max_length=64)
|
||
draftId: str | None = Field(default=None, max_length=64)
|
||
# 外部任务 id（taskId 深链会商）。非空 = task 作业：按 task_id 共享、不按 user 分权，
|
||
# 其 draft_id 指向独立的 roundtable_task_drafts 表（最终报告写回那里）。
|
||
taskId: str | None = Field(default=None, max_length=128)
|
||
intent: str = ""
|
||
agents: list[JobAgent] = Field(default_factory=list)
|
||
model: str | None = None
|
||
orchestrationMode: str = "recommend"
|
||
chain: dict[str, Any] | None = None
|
||
# dag 模式分层编排计划（前端 OrchestrationPlan：{mode,stages,finalSynthesis,coordinatorPrompt?}）。
|
||
# 仅 orchestrationMode=="dag" 时携带；驱动后端 _run_dag。缺失 → 引擎回退 recommend。
|
||
orchestrationPlan: dict[str, Any] | None = None
|
||
# 第一轮 leader prompt；缺省用 intent。
|
||
seedMessage: str | None = None
|
||
# 预建好的 thread 映射（Phase 3b 真网关用；模拟网关忽略）。
|
||
threadIds: dict[str, str] | None = None
|
||
coordinatorName: str = "roundtable-coordinator"
|
||
# 席位执行模式（快速/思考/专业问答/多智能体）派生参数，由前端 toStartBody 下发。
|
||
# 仅 special 席位用：覆盖 roundtable_run_policy("seat") 的 thinking/reasoning；
|
||
# None/缺省 → 后端按 policy 默认(≈pro)，向后兼容旧前端。subagent 仅 ultra 档为真。
|
||
thinking_enabled: bool | None = None
|
||
reasoning_effort: str | None = None
|
||
subagent_enabled: bool = False
|
||
# 每席位「技能识别强化」开关映射 {agent_id: bool}（前端读所选业务链条里**每个席位**的
|
||
# 同名配置后下发）。后台席位 run 据此（与各席位 agent 自身 config.yaml 的开关取 OR）决定
|
||
# 是否注入。缺省空 dict（全部按 False/关，仅各 agent 自身开关生效）。
|
||
seat_skill_directives: dict[str, bool] = Field(default_factory=dict)
|
||
# 每席位「推理深度覆盖」映射 {agent_id: SeatMode}（前端读所选业务链条里**每个席位**单独配的
|
||
# 推理深度后下发；值为 flash/thinking/pro/ultra）。后台席位 run 按 agent_id 查表：命中则用该档
|
||
# 派生的 thinking/reasoning/subagent 覆盖作业级 seatMode；未命中 → 跟随作业级。缺省空 dict。
|
||
seat_modes: dict[str, str] = Field(default_factory=dict)
|
||
# 是否研讨前先并行『取数』（业务链条逐条 opt-in；前端读所选链条的开关后下发）。默认 False。
|
||
gather_first: bool = False
|
||
# 深链接业务码（rwfx→6BF 等）。透传到后台总结报告的「业务链完整性核对」。普通会商为空。
|
||
business_code: str | None = None
|
||
|
||
|
||
class ResumeJobRequest(BaseModel):
|
||
answer: str = ""
|
||
|
||
|
||
class CleanupRequest(BaseModel):
|
||
retention_days: int = Field(default=7, ge=0, description="保留多少天内的作业；0 = 不保留")
|
||
only_finished: bool = Field(default=True, description="仅清终态（done/error/cancelled），保护在跑作业")
|
||
batch_size: int = Field(default=500, ge=1, le=5000)
|
||
|
||
|
||
class CleanupResponse(BaseModel):
|
||
deleted: int
|
||
elapsed_seconds: float
|
||
params: dict[str, Any]
|
||
|
||
|
||
class RoundtableJobResponse(BaseModel):
|
||
id: str
|
||
draftId: str | None = None
|
||
status: str = "queued"
|
||
phase: str = "initializing"
|
||
cycle: int = 0
|
||
dispatchChain: list[dict[str, Any]] = Field(default_factory=list)
|
||
activeAgentId: str | None = None
|
||
consensusPercentage: int = 0
|
||
pendingClarification: str | None = None
|
||
orchestrationMode: str = "recommend"
|
||
chain: dict[str, Any] | None = None
|
||
orchestrationPlan: dict[str, Any] | None = None
|
||
error: str | None = None
|
||
dialogues: list[dict[str, Any]] = Field(default_factory=list)
|
||
step3: dict[str, Any] | None = None
|
||
createdAt: datetime | str | None = None
|
||
updatedAt: datetime | str | None = None
|
||
|
||
|
||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _current_user_id(request: Request) -> str:
|
||
user = getattr(request.state, "user", None)
|
||
if user is not None:
|
||
return str(user.id)
|
||
return get_effective_user_id()
|
||
|
||
|
||
def _get_store(request: Request):
|
||
store = getattr(request.app.state, "roundtable_job_store", None)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Roundtable job store not available")
|
||
return store
|
||
|
||
|
||
def _get_executor(request: Request):
|
||
executor = getattr(request.app.state, "roundtable_job_executor", None)
|
||
if executor is None:
|
||
raise HTTPException(status_code=503, detail="Roundtable job executor not available")
|
||
return executor
|
||
|
||
|
||
def _get_dispatcher(request: Request):
|
||
"""Phase 3 dispatcher（可能为 None：store 未配置时）。nudge 前先判空。"""
|
||
return getattr(request.app.state, "roundtable_job_dispatcher", None)
|
||
|
||
|
||
def _request_hash(body: StartJobRequest) -> str:
|
||
"""启动入参指纹（sha256）——审计/排查「同 requestId 但入参变了」的异常重试。"""
|
||
canonical = json.dumps(
|
||
body.model_dump(mode="json", exclude={"requestId"}),
|
||
ensure_ascii=False, sort_keys=True, default=str,
|
||
)
|
||
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _pending_nodes(agents: list[JobAgent]) -> list[dict[str, Any]]:
|
||
"""启动即落一份「全 pending」的派活链路，使 GET / 首帧 SSE 立刻能渲染流程图。"""
|
||
return [
|
||
{"agentId": a.agentId, "name": a.name or a.agentId, "order": i + 1, "state": "pending"}
|
||
for i, a in enumerate(agents)
|
||
]
|
||
|
||
|
||
def _to_response(row: dict[str, Any]) -> RoundtableJobResponse:
|
||
return RoundtableJobResponse(
|
||
id=row["id"],
|
||
draftId=row.get("draft_id"),
|
||
status=row.get("status") or "queued",
|
||
phase=row.get("phase") or "initializing",
|
||
cycle=int(row.get("cycle") or 0),
|
||
dispatchChain=row.get("dispatch_chain") or [],
|
||
activeAgentId=row.get("active_agent_id"),
|
||
consensusPercentage=int(row.get("consensus_percentage") or 0),
|
||
pendingClarification=row.get("pending_clarification"),
|
||
orchestrationMode=row.get("orchestration_mode") or "recommend",
|
||
chain=row.get("chain"),
|
||
orchestrationPlan=row.get("orchestration_plan"),
|
||
error=row.get("error"),
|
||
dialogues=row.get("dialogues") or [],
|
||
step3=row.get("step3"),
|
||
createdAt=row.get("created_at"),
|
||
updatedAt=row.get("updated_at"),
|
||
)
|
||
|
||
|
||
def _seat_refs_from_nodes(nodes: list[dict[str, Any]] | None) -> list[SeatRef]:
|
||
out: list[SeatRef] = []
|
||
for n in nodes or []:
|
||
aid = str(n.get("agentId") or "")
|
||
if aid:
|
||
out.append(SeatRef(aid, str(n.get("name") or aid)))
|
||
return out
|
||
|
||
|
||
# ── routes ───────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@router.post("", response_model=RoundtableJobResponse, status_code=201)
|
||
async def start_job(
|
||
request: Request, body: StartJobRequest, response: Response
|
||
) -> RoundtableJobResponse:
|
||
"""启动一个后台作业（Phase 3：唯一活跃键去重 + dispatcher 领取执行）。
|
||
|
||
流程：
|
||
1. 把整份启动入参序列化成 ``input_snapshot`` 落库（跨 worker 重建 StartParams 的前提）；
|
||
2. ``try_create_or_get_active`` 直接 INSERT 并占用 ``active_dedupe_key`` 唯一索引 ——
|
||
冲突（同草稿已有活跃作业）即幂等返回它（``created=False`` → HTTP 200），不再新建；
|
||
3. 新建成功（HTTP 201，status=queued）后 ``nudge`` dispatcher，由它 ``claim_job``
|
||
原子领取并在本（或其它）worker 执行 —— **不再由路由直接拉起进程内任务**。
|
||
|
||
「同一草稿至多一个活跃作业」由 DB 唯一索引强制，跨 worker / 并发 / 重试天然互斥，
|
||
不依赖 GET_LOCK 命名锁或进程内 asyncio。
|
||
"""
|
||
store = _get_store(request)
|
||
user_id = _current_user_id(request)
|
||
|
||
if not body.agents:
|
||
raise HTTPException(status_code=400, detail="agents is required")
|
||
|
||
task_id = (body.taskId or "").strip() or None
|
||
draft_id = (body.draftId or "").strip() or None
|
||
agents = [SeatRef(a.agentId, a.name or a.agentId) for a in body.agents]
|
||
seed_message = (body.seedMessage or body.intent or "").strip()
|
||
|
||
# 整份启动入参快照 —— dispatcher 在任意 worker 领取后据此无损重建 StartParams。
|
||
snapshot = build_start_snapshot(
|
||
user_id=user_id,
|
||
agents=agents,
|
||
seed_message=seed_message,
|
||
model=body.model,
|
||
mode=body.orchestrationMode,
|
||
intent_text=body.intent,
|
||
thread_ids=body.threadIds,
|
||
coordinator_name=body.coordinatorName,
|
||
draft_id=draft_id,
|
||
task_id=task_id,
|
||
is_resume=False,
|
||
chain=body.chain,
|
||
orchestration_plan=body.orchestrationPlan,
|
||
seat_thinking_enabled=body.thinking_enabled,
|
||
seat_reasoning_effort=body.reasoning_effort,
|
||
seat_subagent_enabled=body.subagent_enabled,
|
||
seat_skill_directives=dict(body.seat_skill_directives or {}),
|
||
seat_modes=dict(body.seat_modes or {}),
|
||
gather_first=bool(body.gather_first),
|
||
business_code=(body.business_code or None),
|
||
)
|
||
request_id = (body.requestId or "").strip() or uuid4().hex
|
||
|
||
job_id = uuid4().hex
|
||
row, created = await store.try_create_or_get_active(
|
||
draft_id=draft_id,
|
||
user_id=user_id,
|
||
task_id=task_id,
|
||
id=job_id,
|
||
orchestration_mode=body.orchestrationMode,
|
||
coordinator_name=body.coordinatorName,
|
||
thread_ids=body.threadIds,
|
||
chain=body.chain,
|
||
orchestration_plan=body.orchestrationPlan,
|
||
dispatch_chain=_pending_nodes(body.agents),
|
||
status="queued",
|
||
request_id=request_id,
|
||
request_hash=_request_hash(body),
|
||
input_snapshot=snapshot,
|
||
)
|
||
if created:
|
||
# 新建（queued）：交给 dispatcher 领取执行。
|
||
response.status_code = 201
|
||
else:
|
||
# 命中已有活跃作业：幂等返回（不重复建 / 不重复跑）。
|
||
response.status_code = 200
|
||
# 任一分支都唤醒 dispatcher：新建需被领取；幂等命中时若既有作业还在 queued、未被任何
|
||
# worker 领走（如 dispatcher 正在轮询间隙），nudge 确保它被立刻 claim，无感延迟。
|
||
dispatcher = _get_dispatcher(request)
|
||
if dispatcher is not None:
|
||
dispatcher.nudge()
|
||
return _to_response(row)
|
||
|
||
|
||
@router.get("", response_model=RoundtableJobResponse | None)
|
||
async def get_job_by_draft(request: Request, draft_id: str, task_id: str | None = None) -> RoundtableJobResponse | None:
|
||
# task_id 非空 = task 作业查询（按 task 共享、不分权）；否则按 user 分权。
|
||
store = _get_store(request)
|
||
tid = (task_id or "").strip() or None
|
||
user_id = _current_user_id(request)
|
||
row = await store.get_by_draft(draft_id, user_id=user_id, task_id=tid)
|
||
return _to_response(row) if row is not None else None
|
||
|
||
|
||
@router.get("/{job_id}", response_model=RoundtableJobResponse)
|
||
async def get_job(request: Request, job_id: str, task_id: str | None = None) -> RoundtableJobResponse:
|
||
store = _get_store(request)
|
||
tid = (task_id or "").strip() or None
|
||
user_id = _current_user_id(request)
|
||
row = await store.get(job_id, user_id=user_id, task_id=tid)
|
||
if row is None:
|
||
raise HTTPException(status_code=404, detail="Job not found")
|
||
return _to_response(row)
|
||
|
||
|
||
@router.get("/{job_id}/stream")
|
||
async def stream_job(request: Request, job_id: str, task_id: str | None = None) -> StreamingResponse:
|
||
store = _get_store(request)
|
||
tid = (task_id or "").strip() or None
|
||
user_id = _current_user_id(request)
|
||
|
||
def _frame(payload: dict[str, Any]) -> str:
|
||
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
||
|
||
async def gen():
|
||
for _ in range(_MAX_STREAM_TICKS):
|
||
if await request.is_disconnected():
|
||
return
|
||
row = await store.get(job_id, user_id=user_id, task_id=tid)
|
||
if row is None:
|
||
yield _frame({"event": "error", "detail": "Job not found"})
|
||
return
|
||
yield _frame({"event": "progress", **_to_response(row).model_dump(mode="json")})
|
||
if (row.get("status") or "") in _TERMINAL:
|
||
return
|
||
await asyncio.sleep(_POLL_INTERVAL_S)
|
||
|
||
return StreamingResponse(
|
||
gen(),
|
||
media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"},
|
||
)
|
||
|
||
|
||
@router.post("/{job_id}/resume", response_model=RoundtableJobResponse)
|
||
async def resume_job(
|
||
request: Request, job_id: str, body: ResumeJobRequest, task_id: str | None = None
|
||
) -> RoundtableJobResponse:
|
||
"""回答澄清后续跑（Phase 3：条件 UPDATE 串行 + dispatcher 领取）。
|
||
|
||
``queue_for_resume`` 用**原子条件 UPDATE**（WHERE status='awaiting_input'）把作业翻成
|
||
queued —— 并发 / 重复的 resume 只有一个能成功（其余返回 None → 409），杜绝「同一澄清
|
||
被续跑两次」的双跑。成功后清租约 + 写含续跑语义的新 snapshot，``nudge`` dispatcher 由
|
||
它在任意 worker 领取执行；**路由不再直接拉起进程内任务**。
|
||
"""
|
||
store = _get_store(request)
|
||
tid = (task_id or "").strip() or None
|
||
user_id = _current_user_id(request)
|
||
|
||
row = await store.get(job_id, user_id=user_id, task_id=tid)
|
||
if row is None:
|
||
raise HTTPException(status_code=404, detail="Job not found")
|
||
|
||
answer = body.answer.strip()
|
||
if not answer:
|
||
raise HTTPException(status_code=400, detail="answer is required")
|
||
|
||
# task 作业不分权：放开写入归属校验（user_id=None）。
|
||
write_user_id = None if tid else user_id
|
||
run_user_id = row.get("user_id") or user_id
|
||
|
||
# 续跑 snapshot：优先复用原 input_snapshot（含席位/模式/链条等完整入参），覆盖续跑三件套
|
||
# （is_resume / seed_message / intent_text），并用作业行里**最新**的 thread_ids /
|
||
# coordinator_name（首跑建线程后由执行器写回，比原 snapshot 里的更新）。旧作业无 snapshot
|
||
# 时从作业行字段兜底重建。
|
||
base_snapshot = row.get("input_snapshot") or {}
|
||
if not isinstance(base_snapshot, dict):
|
||
base_snapshot = {}
|
||
resume_snapshot: dict[str, Any] = {
|
||
"v": base_snapshot.get("v", 1),
|
||
"user_id": run_user_id,
|
||
"agents": base_snapshot.get("agents")
|
||
or [{"agent_id": n.get("agentId"), "name": n.get("name")} for n in (row.get("dispatch_chain") or [])],
|
||
"seed_message": _RESUME_PREFIX + answer,
|
||
"model": base_snapshot.get("model"),
|
||
"mode": row.get("orchestration_mode") or base_snapshot.get("mode") or "recommend",
|
||
"intent_text": answer,
|
||
"thread_ids": row.get("thread_ids"),
|
||
"coordinator_name": row.get("coordinator_name") or "roundtable-coordinator",
|
||
"draft_id": row.get("draft_id"),
|
||
"task_id": tid or row.get("task_id"),
|
||
"is_resume": True,
|
||
"chain": row.get("chain"),
|
||
"orchestration_plan": row.get("orchestration_plan"),
|
||
"enable_report": base_snapshot.get("enable_report", True),
|
||
"seat_thinking_enabled": base_snapshot.get("seat_thinking_enabled"),
|
||
"seat_reasoning_effort": base_snapshot.get("seat_reasoning_effort"),
|
||
"seat_subagent_enabled": base_snapshot.get("seat_subagent_enabled", False),
|
||
"seat_skill_directives": base_snapshot.get("seat_skill_directives") or {},
|
||
"seat_modes": base_snapshot.get("seat_modes") or {},
|
||
"gather_first": base_snapshot.get("gather_first", False),
|
||
"business_code": base_snapshot.get("business_code"),
|
||
}
|
||
|
||
updated = await store.queue_for_resume(
|
||
job_id, input_snapshot=resume_snapshot, user_id=write_user_id, task_id=tid
|
||
)
|
||
if updated is None:
|
||
# 条件 UPDATE 未命中：作业已不在 awaiting_input（被并发 resume 抢先 / 已续跑 / 已终态）。
|
||
raise HTTPException(status_code=409, detail="Job is not awaiting input")
|
||
|
||
dispatcher = _get_dispatcher(request)
|
||
if dispatcher is not None:
|
||
dispatcher.nudge()
|
||
return _to_response(updated)
|
||
|
||
|
||
async def _require_admin(request: Request) -> None:
|
||
"""放行 admin / auth-disabled，其余 403（与 ai_writing / scheduled_tasks 同套语义）。"""
|
||
from app.gateway.deps import get_optional_user_from_request
|
||
|
||
user = await get_optional_user_from_request(request)
|
||
if user is None:
|
||
return
|
||
if getattr(user, "system_role", None) != "admin":
|
||
raise HTTPException(status_code=403, detail="圆桌作业清理管理接口仅限管理员")
|
||
|
||
|
||
@router.post("/admin/cleanup", response_model=CleanupResponse)
|
||
async def admin_cleanup(request: Request, body: CleanupRequest) -> CleanupResponse:
|
||
"""清理过期作业行（活跃作业受 only_finished 保护，永不误删）。
|
||
|
||
分批 ``list_older_than`` → ``delete_by_ids``，避免大事务。``retention_days=0``
|
||
一刷清空所有终态作业（调试用）。
|
||
"""
|
||
await _require_admin(request)
|
||
store = _get_store(request)
|
||
cutoff = datetime.now(UTC) - timedelta(days=body.retention_days)
|
||
|
||
t0 = time.perf_counter()
|
||
deleted = 0
|
||
while True:
|
||
ids = await store.list_older_than(cutoff, only_finished=body.only_finished, limit=body.batch_size)
|
||
if not ids:
|
||
break
|
||
deleted += await store.delete_by_ids(ids)
|
||
if len(ids) < body.batch_size:
|
||
break
|
||
|
||
return CleanupResponse(
|
||
deleted=deleted,
|
||
elapsed_seconds=round(time.perf_counter() - t0, 3),
|
||
params={"retention_days": body.retention_days, "only_finished": body.only_finished, "batch_size": body.batch_size},
|
||
)
|
||
|
||
|
||
@router.post("/{job_id}/cancel", response_model=RoundtableJobResponse)
|
||
async def cancel_job(request: Request, job_id: str, task_id: str | None = None) -> RoundtableJobResponse:
|
||
"""取消作业（Phase 3：两段式 cancel）。
|
||
|
||
1. ``request_cancel``：原子条件 UPDATE（可取消态 → ``cancel_requested``）—— 权威意图落库，
|
||
重复 cancel 幂等（已 cancel_requested/cancelled/终态时返回 None，直接回现状）。
|
||
2. **无持租者**（queued / awaiting_input，执行器未在跑）→ 路由立刻 ``finalize_cancel``
|
||
收口为终态 ``cancelled``（清唯一活跃键 + 租约）。
|
||
3. **有持租者**（running）→ 保留 ``cancel_requested``，由持租者的 DB 看门狗收口
|
||
``cancelled``；本接口先回 ``cancel_requested``（SSE 随后见翻转）。租约过期前未能
|
||
收口（例如 worker 崩溃）时，dispatcher 会回收该取消请求。
|
||
|
||
「已取消」终态永远由持租者或 finalize 收口，避免与执行器的进度写竞态。
|
||
"""
|
||
store = _get_store(request)
|
||
tid = (task_id or "").strip() or None
|
||
user_id = _current_user_id(request)
|
||
|
||
row = await store.get(job_id, user_id=user_id, task_id=tid)
|
||
if row is None:
|
||
raise HTTPException(status_code=404, detail="Job not found")
|
||
|
||
# task 作业不分权：放开写入归属校验。
|
||
write_user_id = None if tid else user_id
|
||
requested = await store.request_cancel(job_id, user_id=write_user_id, task_id=tid)
|
||
if requested is None:
|
||
# 已不可取消（终态 / 已 cancel_requested）：幂等返回现状，不当作错误。
|
||
fresh = await store.get(job_id, user_id=user_id, task_id=tid)
|
||
return _to_response(fresh or row)
|
||
|
||
if not requested.get("lease_owner"):
|
||
# 无持租者：路由直接收口（queued / awaiting_input）。
|
||
finalized = await store.finalize_cancel(job_id)
|
||
return _to_response(finalized or requested)
|
||
|
||
# 有持租者（running）：不要直接 task.cancel()。直接取消会中断当前 worker 的看门狗，
|
||
# 使 DB 永久遗留在 cancel_requested；让 DB 看门狗以持租者身份收口，崩溃时由 dispatcher
|
||
# 在租约到期后回收，才能保证跨进程取消最终完成。
|
||
return _to_response(requested)
|