diff --git a/pipeline_service/init.py b/pipeline_service/init.py index ca870bf..3b1dd94 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -581,54 +581,60 @@ def load_pipeline_service(): poll_db = DBPools() _dispatched = set() # 防止重复分发 + async def _poll_once(poll_db): + """执行一轮 agent poll。可能因连接获取/SQL 锁等待卡住，由外层 wait_for 超时保护。""" + async with poll_db.sqlorContext("pipeline") as sor: + # 回收僵尸 running 任务：进程崩溃/协程挂起遗留（心跳超时未更新）。 + # 阈值 20 分钟 > LLM 单轮最坏时长（3 次 × 300s 重试 ≈ 15 分钟），避免误杀慢任务。 + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL, updated_at=NOW() " + "WHERE state='running' AND pipeline_id='role_task' " + "AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {}) + + # 全局并发 agent 数限制（max_concurrent_agents，appbase params 表，默认 3）。 + # 以 DB state='running' 计数为准（权威），超出上限时本轮回合不再派发。 + from .workspace import get_max_concurrent_agents + max_n = await get_max_concurrent_agents(sor) + running_rows = await sor.sqlExe( + "SELECT COUNT(*) as c FROM pipeline_tasks " + "WHERE state='running' AND pipeline_id='role_task'", {}) + cur = getattr(running_rows[0], 'c', 0) if running_rows else 0 + avail = max(0, max_n - cur) + if avail <= 0: + # 达到并发上限，本轮不派发（退出 context 后末尾 sleep 再 poll） + return + + recs = await sor.sqlExe( + "SELECT id, tenant_id, role FROM pipeline_tasks " + "WHERE state='submitted' AND pipeline_id='role_task' " + "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 20", + {}) + for rec in (recs or [])[:avail]: + 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}") + + 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)) + async def _poll_loop(): while True: try: - async with poll_db.sqlorContext("pipeline") as sor: - # 回收僵尸 running 任务：进程崩溃/协程挂起遗留（心跳超时未更新）。 - # 阈值 20 分钟 > LLM 单轮最坏时长（3 次 × 300s 重试 ≈ 15 分钟），避免误杀慢任务。 - await sor.sqlExe( - "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL, updated_at=NOW() " - "WHERE state='running' AND pipeline_id='role_task' " - "AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {}) - - # 全局并发 agent 数限制（max_concurrent_agents，appbase params 表，默认 3）。 - # 以 DB state='running' 计数为准（权威），超出上限时本轮回合不再派发。 - from .workspace import get_max_concurrent_agents - max_n = await get_max_concurrent_agents(sor) - running_rows = await sor.sqlExe( - "SELECT COUNT(*) as c FROM pipeline_tasks " - "WHERE state='running' AND pipeline_id='role_task'", {}) - cur = getattr(running_rows[0], 'c', 0) if running_rows else 0 - avail = max(0, max_n - cur) - if avail <= 0: - # 达到并发上限，本轮不派发（退出 context 后末尾 sleep 再 poll） - continue - - recs = await sor.sqlExe( - "SELECT id, tenant_id, role FROM pipeline_tasks " - "WHERE state='submitted' AND pipeline_id='role_task' " - "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 20", - {}) - for rec in (recs or [])[:avail]: - 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}") - - 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)) + # watchdog：每轮 poll 最多 60 秒，超时则跳过本轮，防止连接池/MDL 锁 + # 卡死导致 poller 永久停摆（2026-08 生产事故：poller 停摆 15.7h） + await asyncio.wait_for(_poll_once(poll_db), timeout=60) except Exception as e: - debug(f"agent_poller error: {e}") + debug(f"agent_poller error/timeout: {e}") await asyncio.sleep(10) asyncio.create_task(_poll_loop()) @@ -643,38 +649,43 @@ def load_pipeline_service(): pm_db = _DBP() _pm_dispatched = set() + async def _pm_poll_once(pm_db): + """执行一轮 PM poll。可能因连接获取/SQL 锁等待卡住，由外层 wait_for 超时保护。""" + async with pm_db.sqlorContext("pipeline") as sor: + # 回收僵尸 review 任务：PM 审核进程崩溃后 claimed_by 残留，重新放回审核队列。 + await sor.sqlExe( + "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)", {}) + 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", + {}) + for rec in (recs or []): + 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}") + + 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)) + async def _pm_poll_loop(): while True: try: - async with pm_db.sqlorContext("pipeline") as sor: - # 回收僵尸 review 任务：PM 审核进程崩溃后 claimed_by 残留，重新放回审核队列。 - await sor.sqlExe( - "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)", {}) - 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", - {}) - for rec in (recs or []): - 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}") - - 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)) + # watchdog：每轮 PM poll 最多 60 秒，超时则跳过本轮，防止 poller 永久停摆 + await asyncio.wait_for(_pm_poll_once(pm_db), timeout=60) except Exception as e: - debug(f"pm_poller error: {e}") + debug(f"pm_poller error/timeout: {e}") await asyncio.sleep(15) asyncio.create_task(_pm_poll_loop())