From fa5b5dc8e0f6fcd97c9ac72b364d1632d3d1fc67 Mon Sep 17 00:00:00 2001
From: ymq <ymq@opencomputing.net>
Date: Wed, 19 Aug 2026 19:06:22 +0800
Subject: [PATCH] =?UTF-8?q?feat(task):=20=E6=96=B0=E5=A2=9E=20cancel=5Ftas?=
 =?UTF-8?q?k=20=E5=B7=A5=E5=85=B7=EF=BC=8C=E4=BC=9A=E8=AF=9D=20agent=20?=
 =?UTF-8?q?=E5=8F=AF=E5=8F=96=E6=B6=88=E5=8D=95=E4=B8=AA=E4=BB=BB=E5=8A=A1?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

背景：会话 agent 报告"没有终止运行中任务"的工具，只能靠驳回+附作废意见兜底。
核对：现有工具只有项目级(pause_project)/迭代级(start_next_iteration 的
_cancel_active_tasks、cancel_iteration)的停止能力，task_capability 连 S_CANCELLED
常量都没有，缺"取消单个任务"的任务级工具。

改动：
- task_capability.py: 加 S_CANCELLED 常量 + cancel_task()（任意非终态→cancelled+清
  claimed_by，CAS 只动非终态，附 audit）
- sdlc_ability.py: 加 cancel_task ToolDefinition + _h_cancel_task handler（前缀解析
  截断ID）+ SDL_HANDLERS 注册
---
 pipeline_service/sdlc_ability.py    | 35 +++++++++++++++++++++++++++++
 pipeline_service/task_capability.py | 33 +++++++++++++++++++++++++++
 2 files changed, 68 insertions(+)

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 状态迁移兜底（跨产线自定义状态机用）。"""