yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "feat(task): 新增 cancel_task 工具，会话 agent 可取消单个任务"

Browse Source 
This reverts commit fa5b5dc8e0f6fcd97c9ac72b364d1632d3d1fc67.
This commit is contained in:
ymq 2026-08-19 19:53:07 +08:00
parent fa5b5dc8e0
commit b341a43ad6
2 changed files with 0 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
pipeline_service/sdlc_ability.py
View File

@ -56,12 +56,6 @@ 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执行已提交的任务",
@ -506,34 +500,6 @@ 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:
@ -1590,7 +1556,6 @@ 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,

33
pipeline_service/task_capability.py
View File

@ -25,7 +25,6 @@ S_APPROVED = "approved" # 审核通过
S_COMPLETED = "completed" # 全流程完成
S_WAITING = "waiting" # 挂起等回答
S_FAILED = "failed" # 失败
S_CANCELLED = "cancelled" # 已取消
def _get_db():
@ -262,38 +261,6 @@ 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 状态迁移兜底（跨产线自定义状态机用）。"""