feat: gateway服务层(统一消息入口+会话生命周期+通道注册)+AgentIO适配走gateway

pipeline_service/gateway.py

@@ -0,0 +1,154 @@
+"""
+pipeline-service: gateway — agent 网关（统一消息入口 + 会话生命周期 + 通道注册）
+
+gateway 的核心价值：通道无关的统一消息入口。
+  Web(AgentIO) / 微信 等通道都通过 gateway.run_message() 触达 agent，
+  共享同一套「解析项目上下文 → 加载产线能力 → AgentExecutor 执行」逻辑。
+
+标准化事件流（channel-neutral NDJSON），所有通道可消费：
+  {"type":"progress","message":...}
+  {"type":"tool_call","tool":...,"params":...}
+  {"type":"tool_result","result":...}
+  {"type":"reply","message":...}
+  {"type":"ask_user","message":...}
+  {"type":"error","message":...}
+
+安全模型（微信通道）：机构级公众号 + 用户 openid 绑定 + 匿名拒绝，
+  通道身份（openid）→ 系统 user_id 的映射由各通道适配器负责，gateway 只认 user_id。
+"""
+
+import json
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import AsyncGenerator, Dict, Optional
+
+logger = logging.getLogger("pipeline.gateway")
+
+
+@dataclass
+class GatewaySession:
+    """一个会话 = (channel, user_id) 的 agent 上下文（元数据，executor 每次新建）。"""
+    channel: str
+    user_id: str
+    project_id: str = ""
+    pipeline_id: str = ""
+    project_name: str = ""
+    created_at: float = 0.0
+    updated_at: float = 0.0
+
+
+class Gateway:
+    """agent 网关。"""
+
+    def __init__(self):
+        self._sessions: Dict[str, GatewaySession] = {}   # key = f"{channel}:{user_id}"
+        self._channels: Dict[str, object] = {}
+
+    # ── 通道注册 ──
+
+    def register_channel(self, name: str, channel) -> None:
+        """注册一个通道适配器（如 web/wechat）。"""
+        self._channels[name] = channel
+        logger.info(f"Gateway channel registered: {name}")
+
+    def list_channels(self) -> list:
+        return list(self._channels.keys())
+
+    # ── 会话生命周期 ──
+
+    @staticmethod
+    def _session_key(channel: str, user_id: str) -> str:
+        return f"{channel}:{user_id}"
+
+    def get_session(self, channel: str, user_id: str) -> Optional[GatewaySession]:
+        return self._sessions.get(self._session_key(channel, user_id))
+
+    def clear_session(self, channel: str, user_id: str) -> None:
+        """清除会话（/reset 时用）。"""
+        self._sessions.pop(self._session_key(channel, user_id), None)
+
+    # ── 上下文解析 ──
+
+    async def resolve_project(self, user_id: str) -> dict:
+        """解析用户当前项目 → {pid, pipeline_id, name}。"""
+        ctx = {"pid": "", "pipeline_id": "", "name": ""}
+        if not user_id:
+            return ctx
+        try:
+            from sqlor.dbpools import DBPools
+            db = DBPools()
+            async with db.sqlorContext("pipeline") as sor:
+                recs = await sor.sqlExe(
+                    "SELECT current_project_id FROM pipeline_agent_settings WHERE user_id=${u}$",
+                    {"u": user_id})
+                if recs:
+                    ctx["pid"] = getattr(recs[0], "current_project_id", "") or ""
+                if ctx["pid"]:
+                    proj = await sor.sqlExe(
+                        "SELECT name, pipeline_id FROM sd_projects WHERE id=${p}$",
+                        {"p": ctx["pid"]})
+                    if proj:
+                        ctx["name"] = getattr(proj[0], "name", "")
+                        ctx["pipeline_id"] = getattr(proj[0], "pipeline_id", "") or ""
+        except Exception as e:
+            logger.warning(f"gateway resolve_project failed: {e}")
+        return ctx
+
+    # ── 统一消息入口 ──
+
+    async def run_message(self, channel: str, user_id: str, content: str,
+                          history=None, role: str = "") -> AsyncGenerator[str, None]:
+        """统一消息入口：解析上下文 → 加载产线能力 → AgentExecutor 执行 → yield 事件流。
+
+        Args:
+            channel: 通道名（web / wechat / ...）
+            user_id: 系统用户 ID（通道适配器负责把 openid 映射到 user_id）
+            content: 用户消息内容
+            history: 可选历史消息
+            role: 可选产线内角色（驾驶舱 agent 为空）
+        """
+        from pipeline_core.agent_config import load_agent_config
+        from .agent_loop_v2 import AgentExecutor
+
+        # 1. 解析项目上下文
+        ctx = await self.resolve_project(user_id)
+
+        # 2. 会话生命周期（跟踪 project 切换）
+        key = self._session_key(channel, user_id)
+        sess = self._sessions.get(key)
+        now = time.time()
+        if sess is None or sess.project_id != ctx["pid"]:
+            sess = GatewaySession(
+                channel=channel, user_id=user_id,
+                project_id=ctx["pid"], pipeline_id=ctx["pipeline_id"],
+                project_name=ctx["name"], created_at=now, updated_at=now,
+            )
+            self._sessions[key] = sess
+        else:
+            sess.project_id = ctx["pid"]
+            sess.pipeline_id = ctx["pipeline_id"]
+            sess.project_name = ctx["name"]
+            sess.updated_at = now
+
+        # 3. 加载产线能力 + 创建 executor
+        config = await load_agent_config(
+            pipeline_id=ctx["pipeline_id"], project_id=ctx["pid"])
+        executor = AgentExecutor(
+            config=config, project_id=ctx["pid"], user_id=user_id, role=role)
+
+        # 4. 转发标准化事件流
+        async for chunk in executor.run(content, history=history):
+            yield chunk
+
+
+# ── 全局单例 ──
+
+_gateway: Optional[Gateway] = None
+
+
+def get_gateway() -> Gateway:
+    global _gateway
+    if _gateway is None:
+        _gateway = Gateway()
+    return _gateway

pipeline_service/init.py

@@ -521,6 +521,12 @@ def load_pipeline_service():
     env.AgentExecutor = AgentExecutor
     env.run_agent = run_agent
 
+    # gateway：统一消息入口（Web AgentIO / 微信通道共用）
+    from .gateway import get_gateway, Gateway
+    env.Gateway = Gateway
+    env.get_gateway = get_gateway
+    env.gateway = get_gateway()
+
     # PM agent (v3.3.0: 项目经理审核 + 任务链)
     env.pm_review_run = pm_review_run
     env.pm_review_loop = pm_review_loop