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)