diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index e579bcf..69cc10e 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -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 "" diff --git a/pipeline_service/communication.py b/pipeline_service/communication.py index 3a3e542..0ab01b2 100644 --- a/pipeline_service/communication.py +++ b/pipeline_service/communication.py @@ -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() diff --git a/pipeline_service/human_task_capability.py b/pipeline_service/human_task_capability.py index 8e66f44..c30b9f2 100644 --- a/pipeline_service/human_task_capability.py +++ b/pipeline_service/human_task_capability.py @@ -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}$ " diff --git a/pipeline_service/init.py b/pipeline_service/init.py index ff693c7..cca668b 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -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', '')