From b4bfa9ec5478794359973334ad484bd71cd12737 Mon Sep 17 00:00:00 2001 From: ymq Date: Tue, 15 Sep 2026 16:45:46 +0800 Subject: [PATCH] =?UTF-8?q?fix(todos):=20=E5=88=A0=E9=99=A4/=E6=9A=82?= =?UTF-8?q?=E5=81=9C=E9=A1=B9=E7=9B=AE=E4=B8=8D=E5=86=8D=E4=BA=A7=E7=94=9F?= =?UTF-8?q?=E5=BE=85=E5=8A=9E+=E5=AD=A4=E5=84=BF=E5=BE=85=E5=8A=9E?= =?UTF-8?q?=E6=A0=B9=E5=9B=A0=E4=BF=AE=E5=A4=8D(2026-09-15=E7=94=A8?= =?UTF-8?q?=E6=88=B7=E6=8C=87=E4=BB=A4)=E2=80=94=E2=80=94=E5=9B=9B?= =?UTF-8?q?=E4=B8=AApoller(submitted/review/qc=5Freview/failed)=E6=8E=92?= =?UTF-8?q?=E9=99=A4=E5=B7=B2=E5=88=A0=E9=A1=B9=E7=9B=AE=E6=82=AC=E7=A9=BA?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1(=E6=97=A7=E4=BB=A3=E7=A0=81=E5=8F=AA?= =?UTF-8?q?=E6=8E=92paused,=E5=AD=A4=E5=84=BFfailed=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E6=AF=8F=E8=BD=AEraise=20fault=5Freport=E7=81=8C=E7=BB=99owner?= =?UTF-8?q?.superuser,=E5=AE=9E=E6=B5=8B38=E6=9D=A1=E5=A0=86=E7=A7=AF);han?= =?UTF-8?q?dle=5Ffailed=5Ftask=E9=A1=B9=E7=9B=AE=E8=A1=8C=E4=B8=8D?= =?UTF-8?q?=E5=AD=98=E5=9C=A8=E6=97=B6=E8=B7=B3=E8=BF=87(=E6=97=A7'if=20?= =?UTF-8?q?=5Fp=20and'=E5=AE=88=E5=8D=AB=E5=9C=A8=E9=A1=B9=E7=9B=AE?= =?UTF-8?q?=E5=B7=B2=E5=88=A0=E6=97=B6=E6=95=B4=E4=BD=93=E5=A4=B1=E6=95=88?= =?UTF-8?q?);raise=5Fproblem=E5=AF=B9=E5=B7=B2=E5=88=A0=E9=A1=B9=E7=9B=AE?= =?UTF-8?q?=E6=8B=92=E5=8F=91(=E5=85=A8=E5=B1=80tenant=5Fid=E6=94=BE?= =?UTF-8?q?=E8=A1=8C,paused=E4=B8=8D=E6=8B=A6=E9=98=B2=E5=90=9E=E6=95=85?= =?UTF-8?q?=E9=9A=9C=E9=80=9A=E7=9F=A5);create=5Fhuman=5Ftask=E9=A1=B9?= =?UTF-8?q?=E7=9B=AE=E7=8A=B6=E6=80=81=E9=97=A8=E7=A6=81(paused/archived/c?= =?UTF-8?q?ancelled/completed=E4=B8=8D=E5=8F=91=E6=96=B0=E5=BE=85=E5=8A=9E?= =?UTF-8?q?,=E6=81=A2=E5=A4=8D=E5=90=8E=E5=AF=B9=E8=B4=A6=E5=99=A8?= =?UTF-8?q?=E8=A1=A5=E5=8F=91)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_service/agent_loop.py | 8 +++++++- pipeline_service/communication.py | 14 ++++++++++++++ pipeline_service/human_task_capability.py | 10 +++++++--- pipeline_service/init.py | 20 +++++++++++++++++--- 4 files changed, 45 insertions(+), 7 deletions(-) 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', '')