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

fix(todos): 删除/暂停项目不再产生待办+孤儿待办根因修复(2026-09-15用户指令)——四个poller(submitted/review/qc_review/failed)排除已删项目悬空任务(旧代码只排paused,孤儿failed任务每轮raise fault_report灌给owner.superuser,实测38条堆积);handle_failed_task项目行不存在时跳过(旧'if _p and'守卫在项目已删时整体失效);raise_problem对已删项目拒发(全局tenant_id放行,paused不拦防吞故障通知);create_human_task项目状态门禁(paused/archived/cancelled/completed不发新待办,恢复后对账器补发)

Browse Source
This commit is contained in:
ymq 2026-09-15 16:45:46 +08:00
parent e774cb238b
commit b4bfa9ec54
4 changed files with 45 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
pipeline_service/agent_loop.py
View File

@ -3931,10 +3931,16 @@ async def handle_failed_task(task_id: str, project_id: str) -> dict:
return {"status": "idle", "task_id": task_id}
# paused 项目零消耗（2026-09-02）：暂停 = 完全停摆，失败任务不判定不重跑，
# 直接跳过（本函数后续有 LLM 决策调用，paused 项目每轮白烧 tokens）。
# 已删除项目同样跳过（2026-09-15）：_p 查空 = 项目行已不存在（或 tenant_id 悬空），
# 旧实现 `if _p and ...` 在项目被删时整个守卫被跳过 → 孤儿 failed 任务每轮
# raise fault_report 灌给 owner.superuser（实测 38 条堆积）。全局任务（''/'0'）放行。
_p = await sor.sqlExe(
"SELECT status FROM sd_projects WHERE id=${pid}$", {"pid": project_id})
await sor.sqlExe("COMMIT", {})
if _p and getattr(_p[0], "status", "") == "paused":
if not _p:
if project_id not in ("", "0"):
return {"status": "skipped_deleted_project", "task_id": task_id}
elif getattr(_p[0], "status", "") == "paused":
return {"status": "skipped_paused", "task_id": task_id}
t = recs[0]
title = getattr(t, "title", "") or ""

14
pipeline_service/communication.py
View File

@ -66,6 +66,20 @@ async def raise_problem(problem_type, question, from_role, from_agentid="",
if isinstance(question, (list, tuple)):
question = "\n".join(str(x) for x in question)
async with db.sqlorContext(dbname) as sor:
# 已删除项目门禁（2026-09-15）：项目行已不存在的冒泡问题没有处理载体
# （待办详情/文件链接/任务恢复全悬空），且孤儿任务会反复触发同款问题
# 灌给 owner.superuser（实测 38 条堆积）——直接拒发并告知调用方。
# 全局问题（tenant_id 空或 '0'）不受影响。暂停项目不在此拦：failed 任务
# 报故障的流程是「先 pause 再 raise」，拦 paused 会把故障通知吞掉。
_tid = (tenant_id or '').strip()
if _tid and _tid != '0':
_prec = await sor.sqlExe(
"SELECT id FROM sd_projects WHERE id=${pid}$", {"pid": _tid})
await sor.sqlExe("COMMIT", {})
if not _prec:
logger.info("raise_problem skipped: project %s deleted (type=%s task=%s)",
_tid, problem_type, task_id)
return ""
hr = first_handler_role or "agent.main_agent"
ha = first_handler_agentid or ""
qid = getID()

10
pipeline_service/human_task_capability.py
View File

@ -115,12 +115,16 @@ async def create_human_task(project_id, title, description="", task_type=T_GENER
return False, "须指定 assignee_role 或 assignee_id"
db, dbname = _get_db()
async with db.sqlorContext(dbname) as sor:
# 项目存在性 + 迭代（未显式传时取当前迭代）
# 项目存在性 + 状态门禁（2026-09-15 用户要求：暂停/删除的项目不再产生待办）
proj = await sor.sqlExe(
"SELECT id, org_id FROM sd_projects WHERE id=${pid}$", {"pid": project_id})
"SELECT id, org_id, status FROM sd_projects WHERE id=${pid}$", {"pid": project_id})
await sor.sqlExe("COMMIT", {})
if not proj:
return False, "项目不存在"
return False, "项目不存在（可能已删除，不再产生待办）"
p_status = getattr(proj[0], 'status', '') or ''
if p_status in ('paused', 'archived', 'cancelled', 'completed'):
return False, ("项目当前状态为 %s，不再产生新待办（恢复推进后由对账器按需补发）"
% p_status)
if not iteration_id:
it = await sor.sqlExe(
"SELECT id FROM sd_iterations WHERE project_id=${pid}$ "

20
pipeline_service/init.py
View File

@ -693,6 +693,10 @@ def load_pipeline_service():
"WHERE t.state='submitted' AND t.pipeline_id='role_task' "
"AND t.claimed_by IS NULL "
"AND NOT EXISTS (SELECT 1 FROM sd_projects p WHERE p.id=t.tenant_id AND p.status='paused') "
# 排除已删除项目的悬空任务（tenant_id 非空非全局但 sd_projects 已无此行）——
# 否则孤儿任务永远处理不完、每轮再生产 fault_report 灌给 owner.superuser（2026-09-15 实测 38 条）。
# 全局任务（tenant_id='' 或 '0'）不受影响，照常处理。
"AND (t.tenant_id IN ('','0') OR EXISTS (SELECT 1 FROM sd_projects p2 WHERE p2.id=t.tenant_id)) "
"AND NOT EXISTS ("
" SELECT 1 FROM pipeline_human_tasks ht WHERE ht.iteration_id = ("
" SELECT i.id FROM sd_iterations i WHERE i.project_id=t.tenant_id "
@ -840,6 +844,10 @@ def load_pipeline_service():
"SELECT t.id, t.tenant_id, t.role FROM pipeline_tasks t "
"WHERE t.state='review' AND t.claimed_by IS NULL "
"AND NOT EXISTS (SELECT 1 FROM sd_projects p WHERE p.id=t.tenant_id AND p.status='paused') "
# 排除已删除项目的悬空任务（tenant_id 非空非全局但 sd_projects 已无此行）——
# 否则孤儿任务永远处理不完、每轮再生产 fault_report 灌给 owner.superuser（2026-09-15 实测 38 条）。
# 全局任务（tenant_id='' 或 '0'）不受影响，照常处理。
"AND (t.tenant_id IN ('','0') OR EXISTS (SELECT 1 FROM sd_projects p2 WHERE p2.id=t.tenant_id)) "
"AND NOT EXISTS ("
" SELECT 1 FROM pipeline_human_tasks ht WHERE ht.iteration_id = ("
" SELECT i.id FROM sd_iterations i WHERE i.project_id=t.tenant_id "
@ -922,6 +930,8 @@ def load_pipeline_service():
"SELECT t.id, t.tenant_id, t.role FROM pipeline_tasks t "
"WHERE t.state='qc_review' AND t.claimed_by IS NULL "
"AND NOT EXISTS (SELECT 1 FROM sd_projects p WHERE p.id=t.tenant_id AND p.status='paused') "
# 排除已删除项目的悬空任务（同 submitted/review poller，2026-09-15）
"AND (t.tenant_id IN ('','0') OR EXISTS (SELECT 1 FROM sd_projects p2 WHERE p2.id=t.tenant_id)) "
"ORDER BY t.created_at ASC LIMIT 5",
{})
for rec in (recs or []):
@ -968,9 +978,13 @@ def load_pipeline_service():
try:
async with fd_db.sqlorContext("pipeline") as sor:
recs = await sor.sqlExe(
"SELECT id, tenant_id FROM pipeline_tasks "
"WHERE state='failed' AND pipeline_id='role_task' "
"ORDER BY created_at ASC LIMIT 10",
"SELECT t.id, t.tenant_id FROM pipeline_tasks t "
"WHERE t.state='failed' AND t.pipeline_id='role_task' "
# 排除已删除项目的悬空任务——孤儿 failed 任务反复触发 fault_report，
# 每轮灌给 owner.superuser（admin），是"待办越来越多"的直接源头（2026-09-15 实测）。
# 全局任务（tenant_id='' 或 '0'）照常处理。
"AND (t.tenant_id IN ('','0') OR EXISTS (SELECT 1 FROM sd_projects p2 WHERE p2.id=t.tenant_id)) "
"ORDER BY t.created_at ASC LIMIT 10",
{})
for rec in (recs or []):
tid = getattr(rec, 'id', '')