fix(poller): 依赖门控提前到候选筛选 + 项目轮询，修复队列饥饿

agent_poller 原来 SELECT 后 [:avail] 截断只取队列头 avail 个任务，依赖门控却
在认领阶段(_claim_task)才检查——依赖未满足的僵尸 submitted 任务(created_at 最早)
永久占住队头，导致其他项目/新任务饿死(一项目卡死全平台瘫痪)。

改动:
1. 候选 SELECT 带上 depends_on，批量查依赖状态，排除依赖未满足的任务(终态=completed/approved)
2. 项目轮询(round-robin): 按 tenant_id 分组轮流取就绪任务，保证项目间公平
3. SELECT LIMIT 20->200 容纳僵尸任务；批量查依赖替代 N+1

pipeline_service/init.py

@@ -610,28 +610,82 @@ def load_pipeline_service():
                     # 达到并发上限，本轮不派发（退出 context 后末尾 sleep 再 poll）
                     return
 
+                def _deps_of(depends_on_raw):
+                    """解析 depends_on（JSON 数组字符串或 list）→ 依赖任务 ID 列表。空/非法 → []"""
+                    if not depends_on_raw:
+                        return []
+                    try:
+                        d = json.loads(depends_on_raw) if isinstance(depends_on_raw, str) else depends_on_raw
+                        return [str(x) for x in d if x] if isinstance(d, list) else []
+                    except (json.JSONDecodeError, TypeError, ValueError):
+                        return []
+
+                # 候选任务（含 depends_on，LIMIT 放大以容纳「依赖未满足排在前面」的僵尸任务）
                 recs = await sor.sqlExe(
-                    "SELECT id, tenant_id, role FROM pipeline_tasks "
+                    "SELECT id, tenant_id, role, depends_on FROM pipeline_tasks "
                     "WHERE state='submitted' AND pipeline_id='role_task' "
-                    "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 20",
+                    "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 200",
                     {})
-                for rec in (recs or [])[:avail]:
+
+                # 依赖门控提前到候选筛选：批量查 depends_on 里所有依赖任务的状态，
+                # 排除依赖未满足的「僵尸 submitted」任务——否则它们占住队列头、饿死其他就绪任务
+                # （一项目卡死 → 全平台瘫痪）。终态 = completed/approved，与 _task_deps_satisfied 一致。
+                all_dep_ids = set()
+                for rec in (recs or []):
+                    all_dep_ids.update(_deps_of(getattr(rec, 'depends_on', '') or ''))
+                dep_state = {}
+                if all_dep_ids:
+                    dep_ids = ",".join("'" + x.replace("'", "''") + "'" for x in all_dep_ids)
+                    dep_rows = await sor.sqlExe(
+                        "SELECT id, state FROM pipeline_tasks WHERE id IN (" + dep_ids + ")", {})
+                    dep_state = {getattr(r_, 'id', ''): getattr(r_, 'state', '') for r_ in (dep_rows or [])}
+                await sor.sqlExe("COMMIT", {})
+
+                ready = []
+                for rec in (recs or []):
+                    deps = _deps_of(getattr(rec, 'depends_on', '') or '')
+                    # 依赖 id 查不到（任务已删）→ 视为未满足，保守不放行
+                    if all(dep_state.get(d, '') in ('completed', 'approved') for d in deps):
+                        ready.append(rec)
+
+                # 项目轮询（round-robin）：每个项目轮流取一个就绪任务，保证项目间公平，
+                # 避免单项目大量就绪任务占满全局并发名额、饿死其他项目。（dict 保持插入序）
+                by_project = {}
+                for rec in ready:
+                    by_project.setdefault(getattr(rec, 'tenant_id', ''), []).append(rec)
+                selected = []
+                while len(selected) < avail and by_project:
+                    progressed = False
+                    for pid in list(by_project.keys()):
+                        if len(selected) >= avail:
+                            break
+                        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', '')
                     r = getattr(rec, 'role', '')
-                    if tid and pid and r and tid not in _dispatched:
-                        _dispatched.add(tid)
-                        debug(f"agent_poller dispatching task={tid} role={r}")
+                    if not (tid and pid and r) or tid in _dispatched:
+                        continue
+                    _dispatched.add(tid)
+                    debug(f"agent_poller dispatching task={tid} role={r}")
 
-                        async def _dispatch(tid, pid, r):
-                            try:
-                                await role_agent_loop(pid, r, agent_id=f"poller-{r}")
-                            except Exception as e:
-                                debug(f"agent_poller dispatch error: task={tid} err={e}")
-                            finally:
-                                _dispatched.discard(tid)
+                    async def _dispatch(tid, pid, r):
+                        try:
+                            await role_agent_loop(pid, r, agent_id=f"poller-{r}")
+                        except Exception as e:
+                            debug(f"agent_poller dispatch error: task={tid} err={e}")
+                        finally:
+                            _dispatched.discard(tid)
 
-                        asyncio.ensure_future(_dispatch(tid, pid, r))
+                    asyncio.ensure_future(_dispatch(tid, pid, r))
 
         async def _poll_loop():
             while True: