董宏凯andClaude Opus 4.8
b1b1ff98f5
深链接会商（taskId+goPath，businessCode）下按所选业务链完整层级强制校正 结构化抽取(flow-json) 与 方案总结报告： - 新增 harness 层 extraction_spec（app+harness 共用，单份领域知识两角色用法）： build_structure_directive / build_summary_directive 按 businessCode 取 · rwfx(6BF)：线性 任务→目的→行为体→预期行为→驱动因素→叙事；预期行为↔驱动因素 1:1、余 1:N · xdfx(7BF)：分叉汇聚 DAG，根=行动；行动→受众目标/关键传播节点(1:N)→脆弱点(1:N)→策略(多对一) · 其余业务通用规范（强约束完整性+命名+取数、不套固定模板）；business_code 空→零副作用 - 结构化抽取：flow_extract._hierarchy_hint + _validate_messages 用 structure_directive 权威覆盖默认层级（根治"无论什么业务都按目的→行为体抽/停在行为体"）；前端 streamFlowExtract 传 business_code（源 ingestBusinessCode） - 方案总结报告：前台 multi_agent._special_run(summary) + 后台 build_summary_prompt 注入 business 完整性核对（StartJobRequest/StartParams 透传） - 报告产物策略：.md 报告文件保持完整、干净、无需修改（不写 待补充/不足/⚠️）； 「核对说明」（意图对齐+业务链完整性+不足/可补充）改为只在对话回复正文、且每次必出 - 种子 roundtable-structure SOUL 层级顺序改 spec-driven、去 rwfx 写死示例 Tests: test_roundtable_extraction_spec / test_roundtable_summary_prompt Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
433 lines
22 KiB
Python
433 lines
22 KiB
Python
"""圆桌后台作业执行器（Phase 3a）。
|
||
|
||
把 Phase 2 的编排引擎（``run_orchestration``）作为**后台 asyncio 任务**跑起来，
|
||
survive 发起请求的生命周期。每个任务：
|
||
|
||
1. ``set_current_user``：用作业的 user_id 设进程内用户上下文，让引擎里的进度落库
|
||
（``JobProgressWriter`` → ``RoundtableJobRepository.update_progress(user_id=AUTO)``）
|
||
正确归属（与 scheduled task 后台执行同一套做法）；
|
||
2. 用 ``gateway_factory`` 构造一个 ``RoundtableGateway``（Phase 3a 默认模拟网关，
|
||
Phase 3b 换成进程内真网关），跑完整编排（循环 → 共识 → 报告）。
|
||
|
||
任务句柄登记在 ``_tasks`` 里，供 ``cancel`` 用；完成后自动摘除。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass, field
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
from deerflow.agents.roundtable_orchestrator import (
|
||
JobProgressWriter,
|
||
SeatRef,
|
||
run_orchestration,
|
||
)
|
||
from deerflow.persistence.roundtable_diagnostics import DiagnosticsRecorder, get_default_store
|
||
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 「后台挂起」的 Step3 自动产出（方案总结报告 md + 结果绘制 HTML）总开关。
|
||
# 历史：2026-06-05 因后台 start_run 路径报告不稳定临时禁用；2026-06-12 定位到根因是
|
||
# ``_enforce_agent_access`` 对进程内虚拟 request 的 ``request.state`` 裸访问（AttributeError
|
||
# 杀死编排，见 tests/test_roundtable_inprocess_request.py），修复后恢复开启。
|
||
# 现 Step2 收口后自动跑：① roundtable-summary 写「方案总结报告.md」② roundtable-report
|
||
# 画「方案总览.html」，两段故障隔离（engine._run_step3），产物按手动同款形状写回草稿。
|
||
ROUNDTABLE_BACKGROUND_REPORT_ENABLED = True
|
||
|
||
|
||
class _JobUser:
|
||
"""最小 CurrentUser：满足 user_context 的 ``.id`` 结构协议。"""
|
||
|
||
def __init__(self, user_id: str) -> None:
|
||
self.id = user_id
|
||
|
||
|
||
# 头像配色：与前端 getAvatarTypeFor 对齐（内置圆桌席位固定色，其余按序轮换）。
|
||
_BUILTIN_AVATAR = {
|
||
"roundtable-intelligence": "purple",
|
||
"roundtable-environment": "emerald",
|
||
"roundtable-solution-design": "human",
|
||
"roundtable-risk-review": "amber",
|
||
"roundtable-execution-plan": "pink",
|
||
"roundtable-summary": "blue",
|
||
}
|
||
_FALLBACK_AVATAR = ["blue", "purple", "emerald", "amber", "pink"]
|
||
|
||
|
||
def _avatar_for(agent_id: str, index: int) -> str:
|
||
return _BUILTIN_AVATAR.get(agent_id) or _FALLBACK_AVATAR[index % len(_FALLBACK_AVATAR)]
|
||
|
||
|
||
# 工具步骤卡的回退文案。前端 getStepDisplay 对 tool 步骤是从 key+args 推图标/文案的,
|
||
# 存的 label 基本不展示——这里给个合理回退即可（与前端默认文案对齐）。
|
||
_STEP_LABEL = {
|
||
"agent_orchestration": "派活给子智能体",
|
||
"write_file": "写入文件",
|
||
"str_replace": "写入文件",
|
||
"read_file": "读取文件",
|
||
"ls": "列出文件夹",
|
||
"bash": "执行命令",
|
||
"web_search": "搜索相关信息",
|
||
"ask_clarification": "需要你的协助",
|
||
"present_files": "展示文件",
|
||
}
|
||
|
||
|
||
def _steps_from_tool_calls(tool_calls: list[dict], dialogue_id: object, ts: str) -> list[dict]:
|
||
"""把一轮 tool_calls 映射成前端 ``Step2Dialogue.steps``（工具步骤卡）。
|
||
|
||
- ``key`` = ``tool_calling:<toolName>``——前端 ``getStepDisplay`` 据此推图标 + 文案;
|
||
- ``id`` = ``<dialogueId>-<i>``——对齐前端 ``${bubbleId}-${stepIndex}``;
|
||
- ``args`` 原样带上（write_file 的 path/content、派活的 agent_name/task 等）。
|
||
"""
|
||
steps: list[dict] = []
|
||
for i, tc in enumerate(tool_calls or []):
|
||
name = tc.get("name") or ""
|
||
if not name:
|
||
continue
|
||
steps.append(
|
||
{
|
||
"id": f"{dialogue_id}-{i}",
|
||
"key": f"tool_calling:{name}",
|
||
"label": _STEP_LABEL.get(name, f"使用 “{name}” 工具"),
|
||
"ts": ts,
|
||
"args": tc.get("args") or {},
|
||
}
|
||
)
|
||
return steps
|
||
|
||
|
||
def _seat_stub_messages(dialogues: list[dict]) -> dict[str, list[dict]]:
|
||
"""从各对话的 write_file 调用重建 ``seatStubMessages``（threadId → [stub ai message]）。
|
||
|
||
对齐前端 ``upsertSeatToolCall`` 的 stub 形状与 id 约定（消息 ``roundtable-stub-<tid>``、
|
||
工具调用 ``tc-<tid>-<i>``），使加载草稿后点开产物的虚拟 URL 能正确解析到内容。
|
||
同一 thread 跨轮写多次按 path 去重（后写覆盖），与前端一致。
|
||
"""
|
||
by_tid: dict[str, dict[str, dict]] = {}
|
||
for d in dialogues:
|
||
tid = d.get("threadId")
|
||
if not tid:
|
||
continue
|
||
for tc in d.get("toolCalls") or []:
|
||
if tc.get("name") != "write_file":
|
||
continue
|
||
path = str((tc.get("args") or {}).get("path") or "")
|
||
by_tid.setdefault(tid, {})[path] = tc # 后写覆盖
|
||
out: dict[str, list[dict]] = {}
|
||
for tid, by_path in by_tid.items():
|
||
tool_calls = [
|
||
{"id": f"tc-{tid}-{i}", "name": "write_file", "args": tc.get("args") or {}}
|
||
for i, tc in enumerate(by_path.values())
|
||
]
|
||
out[tid] = [{"id": f"roundtable-stub-{tid}", "type": "ai", "content": "", "tool_calls": tool_calls}]
|
||
return out
|
||
|
||
|
||
def _to_step2_dialogue(d: dict, idx_by_agent: dict[str, int]) -> dict:
|
||
"""把作业的中性对话记录映射成前端 Step2Dialogue 形状（含 avatarType / steps /
|
||
agentThreadId / time），使写进草稿 step2.runs 后，前端 hydrateStep2Run 能像手动跑
|
||
一样直接渲染——步骤卡、时间戳、点开产物全部可用。"""
|
||
dialogue_id = d.get("id")
|
||
time_str = d.get("time") or ""
|
||
base: dict = {
|
||
"id": dialogue_id,
|
||
"content": d.get("content") or "",
|
||
"time": time_str,
|
||
"streaming": False,
|
||
"steps": _steps_from_tool_calls(d.get("toolCalls") or [], dialogue_id, time_str),
|
||
}
|
||
thread_id = d.get("threadId")
|
||
if thread_id:
|
||
base["agentThreadId"] = thread_id
|
||
if d.get("role") == "leader":
|
||
return {**base, "sender": "总控协调", "avatarType": "blue", "confidence": "派活说明"}
|
||
aid = d.get("agentId") or ""
|
||
return {
|
||
**base,
|
||
"sender": d.get("name") or aid,
|
||
"avatarType": _avatar_for(aid, idx_by_agent.get(aid, 0)),
|
||
"confidence": "子智能体交付",
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class StartParams:
|
||
job_id: str
|
||
user_id: str
|
||
agents: list[SeatRef]
|
||
seed_message: str
|
||
model: str | None = None
|
||
mode: str = "recommend"
|
||
intent_text: str = ""
|
||
thread_ids: dict[str, str] | None = None
|
||
coordinator_name: str = "roundtable-coordinator"
|
||
# 关联草稿：作业完成后把 step3 报告写回该草稿，使「查看报告」可用。
|
||
draft_id: str | None = None
|
||
# 外部任务 id（taskId 深链会商）。非空 = task 作业：draft_id 指向独立的
|
||
# roundtable_task_drafts 表，研讨/报告写回 task-draft 存储（不分权）。
|
||
task_id: str | None = None
|
||
# 是否为「续跑」（resume 澄清）。首启=False → 真网关建全新线程（不复用前端线程，
|
||
# 避免与前端本地研讨的活跃 run 撞 409）；续跑=True → 复用作业自己的线程。
|
||
is_resume: bool = False
|
||
# chain 模式来源链条（写进草稿 run 用，展示）。
|
||
chain: dict[str, Any] | None = None
|
||
# dag 模式分层编排计划（前端 OrchestrationPlan dict）：驱动引擎 _run_dag + 写进草稿 run
|
||
# 供前端流程图渲染。None = 非 dag（引擎按 mode 回退 chain/recommend）。
|
||
orchestration_plan: dict[str, Any] | None = None
|
||
# 是否自动产出 Step3（方案总结报告 + 结果绘制）。默认沿用全局开关（现已恢复 True，
|
||
# 见 ROUNDTABLE_BACKGROUND_REPORT_ENABLED）。测试可显式覆盖。
|
||
enable_report: bool = ROUNDTABLE_BACKGROUND_REPORT_ENABLED
|
||
# 席位执行模式（快速/思考/专业问答/多智能体）派生参数，沿用 Step2 选择。仅 special 席位用：
|
||
# 覆盖 roundtable_run_policy("seat") 的 thinking/reasoning（None → policy 默认）；
|
||
# subagent 仅 ultra 档为真。透传给进程内网关(InProcessRoundtableGateway)的席位 run。
|
||
seat_thinking_enabled: bool | None = None
|
||
seat_reasoning_effort: str | None = None
|
||
seat_subagent_enabled: bool = False
|
||
# 每席位「技能识别强化」开关映射 {agent_id: bool}（沿用 Step2 所选业务链条里每个席位的
|
||
# 配置）。透传给进程内网关，run_seat 按 agent_id 查表，与各席位 agent 自身 config.yaml 的
|
||
# 同名开关取 OR。默认空 dict。
|
||
seat_skill_directives: dict[str, bool] = field(default_factory=dict)
|
||
# 每席位「推理深度覆盖」映射 {agent_id: SeatMode}（值 flash/thinking/pro/ultra，沿用 Step2 所选
|
||
# 业务链条里每个席位单独配的推理深度）。透传给进程内网关，run_seat 按 agent_id 查表：命中 → 用该
|
||
# 档派生的 thinking/reasoning/subagent 覆盖作业级 seatMode；未命中 → 跟随作业级。默认空 dict。
|
||
seat_modes: dict[str, str] = field(default_factory=dict)
|
||
# 是否在研讨前先跑一轮**全席位并行『取数』**（业务链条逐条 opt-in）：各席位并行只调技能取数、
|
||
# 不研讨，产出作为共享资料注入后续研讨轮。透传给 run_orchestration 的 gather_first。默认 False。
|
||
gather_first: bool = False
|
||
# 深链接业务码（rwfx→6BF 等）。透传给 run_orchestration → build_summary_prompt：写实业务额外注入
|
||
# 「业务链完整性核对」，让后台总结报告核对产出相对业务链完不完整。普通会商为空 → 不注入。
|
||
business_code: str | None = None
|
||
|
||
|
||
# 网关工厂：给定一次启动参数，返回一个 RoundtableGateway 实例。
|
||
GatewayFactory = Callable[[StartParams], object]
|
||
|
||
|
||
class RoundtableJobExecutor:
|
||
def __init__(self, store, *, gateway_factory: GatewayFactory, draft_store=None, task_draft_store=None) -> None:
|
||
self._store = store
|
||
self._gateway_factory = gateway_factory
|
||
# 草稿仓储（可选）：作业完成后把 step3 写回草稿，供前端「查看报告」。
|
||
self._draft_store = draft_store
|
||
# task 草稿仓储（独立、不分权）：task 作业(params.task_id 非空)写回这里而非个人草稿。
|
||
self._task_draft_store = task_draft_store
|
||
self._tasks: dict[str, asyncio.Task] = {}
|
||
|
||
def _draft_store_for(self, params: StartParams):
|
||
"""task 作业 → 独立 task-draft 存储；普通作业 → 个人草稿存储。"""
|
||
return self._task_draft_store if params.task_id else self._draft_store
|
||
|
||
def start_job(self, params: StartParams) -> bool:
|
||
"""启动一个后台编排任务。同一 job 已在跑则忽略（幂等）。返回是否新启动。"""
|
||
if params.job_id in self._tasks and not self._tasks[params.job_id].done():
|
||
return False
|
||
task = asyncio.create_task(self._run(params))
|
||
self._tasks[params.job_id] = task
|
||
task.add_done_callback(lambda t, jid=params.job_id: self._tasks.pop(jid, None))
|
||
return True
|
||
|
||
async def _run(self, params: StartParams) -> None:
|
||
token = set_current_user(_JobUser(params.user_id))
|
||
diag = DiagnosticsRecorder(
|
||
get_default_store(),
|
||
scope="background",
|
||
job_id=params.job_id,
|
||
draft_id=params.draft_id,
|
||
task_id=params.task_id,
|
||
user_id=params.user_id,
|
||
)
|
||
try:
|
||
await diag.record(
|
||
stage="job", level="info", event="job_started",
|
||
message=f"后台挂起作业启动（mode={params.mode}{'，续跑' if params.is_resume else ''}）",
|
||
detail={"mode": params.mode, "is_resume": params.is_resume, "agents": [a.agent_id for a in params.agents]},
|
||
)
|
||
gateway = self._gateway_factory(params)
|
||
thread_ids = params.thread_ids or {}
|
||
coordinator_name = params.coordinator_name
|
||
# 真网关需要先进程内建好 leader/seat 线程；模拟网关无 init_threads。
|
||
# 首启（非 resume）始终建**全新线程** —— 绝不复用前端本地研讨的线程
|
||
# （那些线程可能有活跃 run，复用会撞 409）。resume 时复用作业自己的线程。
|
||
init = getattr(gateway, "init_threads", None)
|
||
if init is not None and not params.is_resume:
|
||
thread_ids, coordinator_name = await init()
|
||
try:
|
||
await self._store.update_progress(
|
||
params.job_id, thread_ids=thread_ids, coordinator_name=coordinator_name
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("persist thread_ids for job %s failed", params.job_id, exc_info=True)
|
||
|
||
# 每轮模型返回后，把对话作为一个带 jobId 的 run 写进草稿 step2.runs[]
|
||
# —— 和手动跑同一存取路径（draft_store.update_draft），加载草稿即可显示。
|
||
async def _on_dialogue(dialogues: list[dict], consensus: int) -> None:
|
||
await self._write_run_to_draft(
|
||
params, dialogues=dialogues, status="running", consensus=consensus, thread_ids=thread_ids
|
||
)
|
||
|
||
progress = JobProgressWriter(
|
||
self._store,
|
||
job_id=params.job_id,
|
||
agents=params.agents,
|
||
on_dialogue=_on_dialogue if (self._draft_store_for(params) and params.draft_id) else None,
|
||
recorder=diag,
|
||
)
|
||
result = await run_orchestration(
|
||
gateway=gateway,
|
||
progress=progress,
|
||
agents=params.agents,
|
||
thread_ids=thread_ids,
|
||
coordinator_name=coordinator_name,
|
||
model=params.model,
|
||
seed_message=params.seed_message,
|
||
mode=params.mode,
|
||
intent_text=params.intent_text,
|
||
enable_report=params.enable_report,
|
||
plan=params.orchestration_plan,
|
||
gather_first=params.gather_first,
|
||
business_code=params.business_code,
|
||
)
|
||
# 终态：把最终对话 + step3 + jobStatus 落进同一个 run（含 furthest_step=3）。
|
||
await self._write_run_to_draft(
|
||
params,
|
||
dialogues=result.dialogues,
|
||
status=result.status,
|
||
consensus=100 if result.status == "done" else 95,
|
||
step3=result.step3,
|
||
thread_ids=thread_ids,
|
||
)
|
||
# step3 也挂 job（API / 弹窗用）。
|
||
if result.status == "done" and result.step3:
|
||
try:
|
||
await self._store.update_progress(params.job_id, step3=result.step3)
|
||
except Exception: # noqa: BLE001
|
||
logger.warning("persist step3 to job %s failed", params.job_id, exc_info=True)
|
||
except asyncio.CancelledError:
|
||
# 干净停止即可，**不在被取消的任务里写 DB**：取消态下二次 await 会再抛
|
||
# CancelledError、并可能打断进行中的 sqlite 连接。``cancelled`` 终态由
|
||
# 路由的 cancel 端点（executor.cancel + store.update_progress）权威写入。
|
||
logger.info("roundtable job %s cancelled", params.job_id)
|
||
raise
|
||
except Exception as exc: # noqa: BLE001 — 顶层兜底：run_orchestration 内部已尽量标 error
|
||
logger.exception("roundtable job %s crashed", params.job_id)
|
||
await diag.record(
|
||
stage="job", level="error", event="job_crashed",
|
||
message=f"后台挂起作业崩溃：{exc}",
|
||
detail={"type": type(exc).__name__, "error": str(exc)},
|
||
)
|
||
finally:
|
||
reset_current_user(token)
|
||
|
||
async def _write_run_to_draft(
|
||
self,
|
||
params: StartParams,
|
||
*,
|
||
dialogues: list[dict],
|
||
status: str,
|
||
consensus: int,
|
||
step3: dict | None = None,
|
||
thread_ids: dict[str, str] | None = None,
|
||
) -> None:
|
||
"""把本次后台作业作为一个带 jobId 的 run 写进草稿 step2.runs[]（同手动跑存取）。
|
||
|
||
读现有 step2 → 按 run_id=``job-<jobId>`` upsert（保留其它手动 run）→ 设为 activeRun
|
||
→ update_draft。前端加载草稿即可像手动跑一样 hydrate 出对话。
|
||
"""
|
||
draft_store = self._draft_store_for(params)
|
||
if draft_store is None or not params.draft_id:
|
||
return
|
||
idx_by_agent = {a.agent_id: i for i, a in enumerate(params.agents)}
|
||
s2_dialogues = [_to_step2_dialogue(d, idx_by_agent) for d in dialogues]
|
||
last_leader = next(
|
||
(d.get("content") for d in reversed(dialogues) if d.get("role") == "leader"), ""
|
||
)
|
||
run_id = f"job-{params.job_id}"
|
||
try:
|
||
draft = await draft_store.get_draft(params.draft_id, params.user_id)
|
||
except Exception: # noqa: BLE001
|
||
draft = None
|
||
step2 = dict((draft or {}).get("step2") or {})
|
||
runs = list(step2.get("runs") or [])
|
||
idx = next((i for i, r in enumerate(runs) if r.get("id") == run_id), -1)
|
||
created = runs[idx].get("createdAt") if idx >= 0 else datetime.now(UTC).isoformat()
|
||
prev_step3 = runs[idx].get("step3") if idx >= 0 else None
|
||
# 本次有效 step3：新报告优先，否则保留上次（running 中间态 step3=None 不抹掉已存的）。
|
||
effective_step3 = step3 if step3 is not None else prev_step3
|
||
run = {
|
||
"id": run_id,
|
||
"title": "后台研讨",
|
||
"createdAt": created,
|
||
"source": params.mode,
|
||
"selectedAgents": [{"agent_id": a.agent_id, "name": a.name} for a in params.agents],
|
||
# 后台真建的 leader/seat 线程是真实持久的——存进去，前端加载后能像手动跑一样
|
||
# 按 thread 打开各席位产物（HTTP outputs 兜底）。hydrate 是纯展示、绝不 resume，
|
||
# 不会触发重跑（init 闸门 + isRoundTableRunning=false 双重拦截）。空则存 None。
|
||
"threadIds": (thread_ids or None),
|
||
"coordinatorName": params.coordinator_name,
|
||
"step2RoundtableDialogues": s2_dialogues,
|
||
"lastLeaderContent": last_leader or "",
|
||
"hasConsensus": status == "done",
|
||
"consensusPercentage": int(consensus),
|
||
"budgetLimit": 0,
|
||
# 从各席位 write_file 调用重建 stub，使加载草稿后点开产物的虚拟预览可用。
|
||
"seatStubMessages": _seat_stub_messages(dialogues),
|
||
"orchestrationMode": params.mode,
|
||
"chain": params.chain,
|
||
# dag 模式分层计划：让前端加载草稿后能像手动跑一样按 stage 画流程图。
|
||
"orchestrationPlan": params.orchestration_plan,
|
||
"step3": effective_step3,
|
||
"jobId": params.job_id,
|
||
"jobStatus": status,
|
||
}
|
||
runs = [run if i == idx else r for i, r in enumerate(runs)] if idx >= 0 else [*runs, run]
|
||
step2["runs"] = runs
|
||
step2["activeRunId"] = run_id
|
||
# 顶层 draft.step3 也写一份（同 Step2 把对话写进 run 的同一次落库一并带上）。
|
||
# 前端 hydrateStep3 用**顶层** draft.step3 重建报告气泡（结果绘制的聊天记录 + HTML 预览）；
|
||
# 只存 run 内 step3 时，加载会被 hydrateStep3(draft.step3=None) 覆盖成空 → 报告聊天记录丢失。
|
||
# furthest_step 跟随**是否真有 step3**：done 且有报告→3（落到结果绘制）；
|
||
# 否则停在 2（Step2 研讨）。后台禁用报告时 step3=None → 停在 Step2，用户可在前端手动画报告。
|
||
has_report = effective_step3 is not None
|
||
update_kwargs: dict[str, Any] = {"step2": step2, "furthest_step": 3 if (status == "done" and has_report) else 2}
|
||
if has_report:
|
||
update_kwargs["step3"] = effective_step3
|
||
try:
|
||
await draft_store.update_draft(params.draft_id, params.user_id, **update_kwargs)
|
||
except Exception as exc: # noqa: BLE001 — 写草稿失败不影响作业本身
|
||
logger.warning("write run to draft %s failed", params.draft_id, exc_info=True)
|
||
from deerflow.persistence.roundtable_diagnostics import record_diagnostic
|
||
|
||
await record_diagnostic(
|
||
scope="background", stage="job", level="error", event="draft_write_failed",
|
||
message=f"研讨结果写回草稿失败（status={status}）：{exc}",
|
||
detail={"type": type(exc).__name__, "error": str(exc)},
|
||
job_id=params.job_id, draft_id=params.draft_id, task_id=params.task_id, user_id=params.user_id,
|
||
)
|
||
|
||
def cancel(self, job_id: str) -> bool:
|
||
"""请求取消一个在跑的作业。返回是否命中在跑任务。"""
|
||
task = self._tasks.get(job_id)
|
||
if task is None or task.done():
|
||
return False
|
||
task.cancel()
|
||
return True
|
||
|
||
def is_running(self, job_id: str) -> bool:
|
||
task = self._tasks.get(job_id)
|
||
return task is not None and not task.done()
|
||
|
||
async def await_job(self, job_id: str) -> None:
|
||
"""等待某作业任务结束（测试用；生产不调用）。"""
|
||
task = self._tasks.get(job_id)
|
||
if task is not None:
|
||
try:
|
||
await task
|
||
except (Exception, asyncio.CancelledError): # noqa: BLE001 — 取消是 BaseException
|
||
pass
|