fix(sdlc): pm_poller 队列饥饿——paused 项目僵尸 review 任务卡队头饿死 active 项目

根因：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

@@ -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: