From 4c0b34971806f4ec981388e3489ed6d83556c451 Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Wed, 5 Aug 2026 17:20:26 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E5=90=8E=E5=8F=B0=E8=BD=AE=E8=AF=A2?=
 =?UTF-8?q?=E5=99=A8=20=E2=80=94=20add=5Fstartup=20=E8=87=AA=E5=8A=A8?=
 =?UTF-8?q?=E6=89=A7=E8=A1=8C=20role=5Ftask?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- _role_poller 每10秒扫描 pipeline_tasks 中 submitted 的 role_task
- 逐条 dispatch 给对应角色 agent_loop 异步执行
- 修复 agent_loop 中 agent_ask 导入为相对导入
---
 pipeline_service/agent_loop.py |  2 +-
 pipeline_service/init.py       | 32 ++++++++++++++++++++++++++++++++
 2 files changed, 33 insertions(+), 1 deletion(-)

diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py
index e3582e0..c3f652d 100644
--- a/pipeline_service/agent_loop.py
+++ b/pipeline_service/agent_loop.py
@@ -166,7 +166,7 @@ async def role_agent_run(project_id: str, role: str, agent_id: str = None, model
 
         # 缺信息 → 提问回路：任务置 waiting，等主agent/客户回答
         if parsed.get("status") == "need_info" and parsed.get("question"):
-            from pipeline_service.questions import agent_ask
+            from .questions import agent_ask
             qid = await agent_ask(project_id, task_id, role, parsed["question"],
                                   context={"partial": parsed.get("partial", ""), "title": title})
             return {"status": "need_info", "task_id": task_id,
diff --git a/pipeline_service/init.py b/pipeline_service/init.py
index 2cd6bc4..4bc29b6 100644
--- a/pipeline_service/init.py
+++ b/pipeline_service/init.py
@@ -8,6 +8,8 @@
 """
 
 import json
+import asyncio
+import logging
 from ahserver.serverenv import ServerEnv
 from appPublic.uniqueID import getID
 from appPublic.log import debug
@@ -427,5 +429,35 @@ def load_pipeline_service():
     env.question_list = list_questions
     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():
+        """轮询 pipeline_tasks 中待执行的 role_task，交给对应角色agent执行。"""
+        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)
+
+    from ahserver.configuredServer import add_startup
+    add_startup(_role_poller())
+
     debug(f"[{MODULE_NAME}] v{MODULE_VERSION} loaded — pipeline engine with role-agent + question loop")
     return True