From f17f745ac649747f8f5b56b54eb40be6f97e3322 Mon Sep 17 00:00:00 2001
From: ymq <ymq@opencomputing.net>
Date: Fri, 21 Aug 2026 13:20:32 +0800
Subject: [PATCH] =?UTF-8?q?fix(sdlc):=20pm=5Fpoller=20=E9=98=9F=E5=88=97?=
 =?UTF-8?q?=E9=A5=A5=E9=A5=BF=E2=80=94=E2=80=94paused=20=E9=A1=B9=E7=9B=AE?=
 =?UTF-8?q?=E5=83=B5=E5=B0=B8=20review=20=E4=BB=BB=E5=8A=A1=E5=8D=A1?=
 =?UTF-8?q?=E9=98=9F=E5=A4=B4=E9=A5=BF=E6=AD=BB=20active=20=E9=A1=B9?=
 =?UTF-8?q?=E7=9B=AE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

根因：pm_poller SELECT state='review' AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 5，
只取最老 5 个 review 任务，但暂停门控在 pm_review_run 认领后才检查。paused 项目（人事系统/人事系统1）
的 8 个 develop/deploy_test 僵尸任务永远占住队头，每 15s dispatch 后暂停门控 return idle、claimed_by 恒 NULL，
active 项目（人事项目2）的 requirement/design 任务排在后面对、永远进不了 LIMIT 5 窗口 → PM 无法推进项目。

修复（对齐 agent_poller 队列饥饿修复三件套）：
1. SELECT 加 NOT EXISTS paused 项目过滤（暂停门控提前到筛选阶段）
2. LIMIT 5 → 200 容纳僵尸任务
3. 项目轮询 round-robin 保证项目间公平
---
 pipeline_service/init.py | 52 +++++++++++++++++++++++++++++-----------
 1 file changed, 38 insertions(+), 14 deletions(-)

diff --git a/pipeline_service/init.py b/pipeline_service/init.py
index cac8dba..5457ef9 100644
--- a/pipeline_service/init.py
+++ b/pipeline_service/init.py
@@ -739,27 +739,51 @@ def load_pipeline_service():
                     "UPDATE pipeline_tasks SET claimed_by=NULL, updated_at=NOW() "
                     "WHERE state='review' AND claimed_by IS NOT NULL "
                     "AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {})
+                # 候选任务：排除 paused 项目的任务（暂停门控提前到筛选，避免 paused 项目僵尸任务
+                # 卡死队列头、饿死 active 项目的审核）。LIMIT 放大容纳「paused 任务排前面」的僵尸任务。
                 recs = await sor.sqlExe(
-                    "SELECT id, tenant_id, role FROM pipeline_tasks "
-                    "WHERE state='review' AND claimed_by IS NULL "
-                    "ORDER BY created_at ASC LIMIT 5",
+                    "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') "
+                    "ORDER BY t.created_at ASC LIMIT 200",
                     {})
+                await sor.sqlExe("COMMIT", {})
+
+                # 项目轮询（round-robin）：每个项目轮流取一个任务，保证项目间公平，
+                # 避免单项目占满窗口、饿死其他项目的审核（与 agent_poller 队列饥饿修复同款）。
+                by_project = {}
                 for rec in (recs or []):
+                    by_project.setdefault(getattr(rec, 'tenant_id', ''), []).append(rec)
+                selected = []
+                while by_project:
+                    progressed = False
+                    for pid in list(by_project.keys()):
+                        tasks = by_project[pid]
+                        if tasks:
+                            selected.append(tasks.pop(0))
+                            progressed = True
+                        if not tasks:
+                            del by_project[pid]
+                    if not progressed:
+                        break
+
+                for rec in selected:
                     tid = getattr(rec, 'id', '')
                     pid = getattr(rec, 'tenant_id', '')
-                    if tid and pid and tid not in _pm_dispatched:
-                        _pm_dispatched.add(tid)
-                        debug(f"pm_poller dispatching review task={tid}")
+                    if not (tid and pid) or tid in _pm_dispatched:
+                        continue
+                    _pm_dispatched.add(tid)
+                    debug(f"pm_poller dispatching review task={tid}")
 
-                        async def _pm_dispatch(tid, pid):
-                            try:
-                                await pm_review_run(pid, agent_id="pm-poller")
-                            except Exception as e:
-                                debug(f"pm_poller dispatch error: task={tid} err={e}")
-                            finally:
-                                _pm_dispatched.discard(tid)
+                    async def _pm_dispatch(tid, pid):
+                        try:
+                            await pm_review_run(pid, agent_id="pm-poller")
+                        except Exception as e:
+                            debug(f"pm_poller dispatch error: task={tid} err={e}")
+                        finally:
+                            _pm_dispatched.discard(tid)
 
-                        asyncio.ensure_future(_pm_dispatch(tid, pid))
+                    asyncio.ensure_future(_pm_dispatch(tid, pid))
 
         async def _pm_poll_loop():
             while True: