diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py index f77c637..7bd9783 100644 --- a/pipeline_service/sdlc_ability.py +++ b/pipeline_service/sdlc_ability.py @@ -56,6 +56,12 @@ SDL_TOOLS = [ parameters={"task_id": "任务ID"}, category="task", ), + ToolDefinition( + name="cancel_task", + description="取消单个任务(任意非终态→cancelled,清claimed_by)。用于作废跑偏/重复的旧任务,保留其他任务继续推进", + parameters={"task_id": "任务ID", "comment": "取消原因(可选)"}, + category="task", + ), ToolDefinition( name="start_agents", description="启动角色agent执行已提交的任务", @@ -500,6 +506,34 @@ async def _h_reset_task_retry(sor, p, ctx): return f"OK: 任务重复计数已归零并重新执行{resume_note}" +async def _h_cancel_task(sor, p, ctx): + """取消单个任务:任意非终态 → cancelled(清 claimed_by)。作废跑偏/重复的旧任务用。""" + task_id = (p.get("task_id") or "").strip() + if not task_id: + return "需要任务ID" + pid = ctx.get("project_id", "") + if not pid: + return "请先切换到项目" + # 前缀解析(LLM 可能传截断 ID) + recs = await sor.sqlExe( + "SELECT id FROM pipeline_tasks WHERE id=${tid}$ AND tenant_id=${pid}$", + {"tid": task_id, "pid": pid}) + if not recs and len(task_id) >= 6: + recs = await sor.sqlExe( + "SELECT id FROM pipeline_tasks WHERE id LIKE ${prefix}$ AND tenant_id=${pid}$ LIMIT 1", + {"prefix": task_id + "%", "pid": pid}) + if not recs: + return f"任务不存在: {task_id}" + task_id = getattr(recs[0], 'id', '') or task_id + + from .task_capability import cancel_task + comment = (p.get("comment") or "").strip() + ok, msg = await cancel_task(task_id, pid, who="agent.main_agent", comment=comment) + if not ok: + return f"ERROR: {msg}" + return f"OK: {msg}" + + async def _h_start_agents(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: @@ -1556,6 +1590,7 @@ SDL_HANDLERS = { "list_tasks": _h_list_tasks, "task_detail": _h_task_detail, "reset_task_retry": _h_reset_task_retry, + "cancel_task": _h_cancel_task, "start_agents": _h_start_agents, "diagnose_project": _h_diagnose_project, "list_deliverables": _h_list_deliverables, diff --git a/pipeline_service/task_capability.py b/pipeline_service/task_capability.py index 68eea6c..30a8713 100644 --- a/pipeline_service/task_capability.py +++ b/pipeline_service/task_capability.py @@ -25,6 +25,7 @@ S_APPROVED = "approved" # 审核通过 S_COMPLETED = "completed" # 全流程完成 S_WAITING = "waiting" # 挂起等回答 S_FAILED = "failed" # 失败 +S_CANCELLED = "cancelled" # 已取消 def _get_db(): @@ -261,6 +262,38 @@ async def revive_task(task_id, tenant_id, who=None, agent_id=None): who=who, agent_id=agent_id) +async def cancel_task(task_id, tenant_id, who=None, agent_id=None, comment=None): + """取消任务:任意非终态 → cancelled,清 claimed_by(poller 不再认领)。 + + 用途:人工/会话 agent 作废跑偏的单个任务(如重复派发的旧任务、跑错方向的任务), + 保留其他任务继续推进。区别于迭代级 _cancel_active_tasks(批量作废整个迭代)。 + 返回 (ok, message)。 + """ + if not task_id or not tenant_id: + return False, "缺少 task_id 或 tenant_id" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT state FROM pipeline_tasks WHERE id=${tid}$ AND tenant_id=${tn}$", + {"tid": task_id, "tn": tenant_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return False, "任务不存在" + cur = getattr(recs[0], 'state', '') or '' + if cur in (S_COMPLETED, S_APPROVED, S_CANCELLED): + return False, f"任务已终态({cur}),无需取消" + # 非终态 → cancelled + 清 claimed_by(CAS 只动非终态,避免并发覆盖) + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='cancelled', claimed_by=NULL, updated_at=NOW() " + "WHERE id=${tid}$ AND state NOT IN ('completed','approved','cancelled')", + {"tid": task_id}) + await record_audit(tenant_id, 'pipeline_tasks', task_id, 'cancel', + from_state=cur, to_state=S_CANCELLED, + who=who, agent_id=agent_id, detail=comment or '取消任务', sor=sor) + logger.info("cancel_task: task=%s %s -> cancelled", task_id, cur) + return True, f"任务已取消({cur} → cancelled)" + + async def set_task_state(task_id, tenant_id, from_state, to_state, who=None, agent_id=None, detail=None): """通用 CAS 状态迁移兜底(跨产线自定义状态机用)。"""