From eb889d8694ecaa6a7fb57eb0dbc9402b20f0e538 Mon Sep 17 00:00:00 2001
From: ymq <ymq@opencomputing.net>
Date: Sat, 8 Aug 2026 13:43:19 +0800
Subject: [PATCH] fix: poller dedup dispatched tasks + error handling to
 prevent lock pile-up

---
 pipeline_service/init.py | 34 +++++++++++++++++++++++++++-------
 1 file changed, 27 insertions(+), 7 deletions(-)

diff --git a/pipeline_service/init.py b/pipeline_service/init.py
index 3dc4952..e3b54f4 100644
--- a/pipeline_service/init.py
+++ b/pipeline_service/init.py
@@ -531,9 +531,10 @@ def load_pipeline_service():
     # Background poller: auto-dispatch submitted role_tasks to role agents
 
     async def _role_poller(app):
-        """启动后台轮询器，spawn 后立即返回（不阻塞 aiohttp 启动）。"""
+        "启动后台轮询器，spawn 后立即返回（不阻塞 aiohttp 启动）。"
         from sqlor.dbpools import DBPools
         poll_db = DBPools()
+        _dispatched = set()  # 防止重复分发
 
         async def _poll_loop():
             while True:
@@ -548,10 +549,19 @@ def load_pipeline_service():
                             tid = getattr(rec, 'id', '')
                             pid = getattr(rec, 'tenant_id', '')
                             r = getattr(rec, 'role', '')
-                            if tid and pid and r:
+                            if tid and pid and r and tid not in _dispatched:
+                                _dispatched.add(tid)
                                 debug(f"agent_poller dispatching task={tid} role={r}")
-                                asyncio.ensure_future(
-                                    role_agent_loop(pid, r, agent_id=f"poller-{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))
                 except Exception as e:
                     debug(f"agent_poller error: {e}")
                 await asyncio.sleep(10)
@@ -566,6 +576,7 @@ def load_pipeline_service():
     async def _pm_poller(app):
         from sqlor.dbpools import DBPools as _DBP
         pm_db = _DBP()
+        _pm_dispatched = set()
 
         async def _pm_poll_loop():
             while True:
@@ -579,10 +590,19 @@ def load_pipeline_service():
                         for rec in (recs or []):
                             tid = getattr(rec, 'id', '')
                             pid = getattr(rec, 'tenant_id', '')
-                            if tid and pid:
+                            if tid and pid and tid not in _pm_dispatched:
+                                _pm_dispatched.add(tid)
                                 debug(f"pm_poller dispatching review task={tid}")
-                                asyncio.ensure_future(
-                                    pm_review_run(pid, agent_id="pm-poller"))
+
+                                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))
                 except Exception as e:
                     debug(f"pm_poller error: {e}")
                 await asyncio.sleep(15)