From ee9ab066fcfd949cfae6618c46c33070edd13551 Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Wed, 5 Aug 2026 17:31:27 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20on=5Fstartup=20handler=20spawn=20?=
 =?UTF-8?q?=E5=90=8E=E5=8F=B0=E4=BB=BB=E5=8A=A1=E5=90=8E=E7=AB=8B=E5=8D=B3?=
 =?UTF-8?q?=E8=BF=94=E5=9B=9E=EF=BC=88=E4=B8=8D=E9=98=BB=E5=A1=9E=20aiohtt?=
 =?UTF-8?q?p=20=E5=90=AF=E5=8A=A8=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pipeline_service/init.py | 47 +++++++++++++++++++++-------------------
 1 file changed, 25 insertions(+), 22 deletions(-)

diff --git a/pipeline_service/init.py b/pipeline_service/init.py
index 2e63bd3..4ba6d15 100644
--- a/pipeline_service/init.py
+++ b/pipeline_service/init.py
@@ -9,7 +9,6 @@
 
 import json
 import asyncio
-import logging
 from ahserver.serverenv import ServerEnv
 from appPublic.uniqueID import getID
 from appPublic.log import debug
@@ -430,31 +429,35 @@ def load_pipeline_service():
     env.question_qna = get_task_qna
 
     # Background poller: auto-dispatch submitted role_tasks to role agents
-    _logger = logging.getLogger("pipeline.agent_poller")
 
     async def _role_poller(app):
-        """轮询 pipeline_tasks 中待执行的 role_task，交给对应角色agent执行。"""
+        """启动后台轮询器，spawn 后立即返回（不阻塞 aiohttp 启动）。"""
         from sqlor.dbpools import DBPools
         poll_db = DBPools()
-        while True:
-            try:
-                async with poll_db.sqlorContext("pipeline") as sor:
-                    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 5",
-                        {})
-                    for rec in (recs or []):
-                        tid = getattr(rec, 'id', '')
-                        pid = getattr(rec, 'tenant_id', '')
-                        r = getattr(rec, 'role', '')
-                        if tid and pid and r:
-                            _logger.info(f"dispatching task={tid} role={r}")
-                            asyncio.ensure_future(
-                                role_agent_loop(pid, r, agent_id=f"poller-{r}"))
-            except Exception as e:
-                _logger.error(f"error: {e}")
-            await asyncio.sleep(10)
+
+        async def _poll_loop():
+            while True:
+                try:
+                    async with poll_db.sqlorContext("pipeline") as sor:
+                        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 5",
+                            {})
+                        for rec in (recs or []):
+                            tid = getattr(rec, 'id', '')
+                            pid = getattr(rec, 'tenant_id', '')
+                            r = getattr(rec, 'role', '')
+                            if tid and pid and r:
+                                debug(f"agent_poller dispatching task={tid} role={r}")
+                                asyncio.ensure_future(
+                                    role_agent_loop(pid, r, agent_id=f"poller-{r}"))
+                except Exception as e:
+                    debug(f"agent_poller error: {e}")
+                await asyncio.sleep(10)
+
+        asyncio.create_task(_poll_loop())
+        debug("agent poller started")
 
     from ahserver.configuredServer import add_startup
     add_startup(_role_poller)