feat(session): gateway/executor 支持 session_id 多会话隔离

web 多 tab 独立会话：run_message 加 session_id，会话 key 从 channel:user_id
扩展为 channel:user_id:session_id（空则向后兼容单会话）；AgentExecutor 历史
加载/保存按 session_id 隔离，空则回退 project 级隔离。

pipeline_service/agent_loop_v2.py

@@ -65,6 +65,7 @@ class AgentExecutor:
         base_url: str = "",            # 请求 base_url（scheme://host/），供 slash 命令生成 widget 绝对 URL
         generic: bool = False,         # True = 纯通用会话（不解析项目、不挂产线能力）
         session=None,                  # GatewaySession（会话级状态：approve_all/pending_confirm）
+        session_id: str = "",          # 会话内唯一标识（web 多 tab 独立会话；历史隔离键）
     ):
         self.config = config
         self.project_id = project_id
@@ -74,6 +75,7 @@ class AgentExecutor:
         self.role = role               # 产线内角色（技能/工具/prompt 按角色加载）
         self.generic = generic         # 是否纯通用会话
         self._session = session        # GatewaySession（approve_all/pending_confirm 会话级状态）
+        self.session_id = session_id   # 会话内唯一标识（历史消息按此隔离）
         self.space = GENERAL_SPACE     # 项目空间键：generic→'general'，产线→真实 pipeline_id
         self.workspace_dir = workspace_dir or WORKSPACE_BASE
         self.model_name = model_name or config.model_name
@@ -1196,7 +1198,11 @@ class AgentExecutor:
     # ═══════════════════════════════════════════════════════
 
     async def _load_history(self, external_history: List[dict] = None) -> List[dict]:
-        """加载历史消息。隔离策略：project 级别。"""
+        """加载历史消息。隔离策略：session_id > project 级别。
+
+        多 tab 独立会话：session_id 非空时按 session_id 隔离（每个 tab 独立历史）；
+        否则退回 config.session_isolation（project/user/none）。
+        """
         if external_history:
             return external_history
 
@@ -1208,22 +1214,32 @@ class AgentExecutor:
 
             db = DBPools()
             async with db.sqlorContext("pipeline") as sor:
-                # 项目隔离
+                # 会话级隔离（web 多 tab 独立会话）
-                pid = self.project_id
+                if self.session_id:
-                if self.config.session_isolation == "project" and pid:
                     recs = await sor.sqlExe(
                         "SELECT role, content FROM pipeline_conversations "
-                        "WHERE iteration_id=${pid}$ AND created_by=${uid}$ "
+                        "WHERE session_id=${sid}$ AND created_by=${uid}$ "
                         "ORDER BY created_at ASC LIMIT ${lim}$",
-                        {"pid": pid, "uid": self.user_id or "", "lim": self.config.history_limit},
+                        {"sid": self.session_id, "uid": self.user_id or "",
+                         "lim": self.config.history_limit},
                     )
                 else:
-                    recs = await sor.sqlExe(
+                    # 项目隔离
-                        "SELECT role, content FROM pipeline_conversations "
+                    pid = self.project_id
-                        "WHERE created_by=${uid}$ "
+                    if self.config.session_isolation == "project" and pid:
-                        "ORDER BY created_at ASC LIMIT ${lim}$",
+                        recs = await sor.sqlExe(
-                        {"uid": self.user_id or "", "lim": self.config.history_limit},
+                            "SELECT role, content FROM pipeline_conversations "
-                    )
+                            "WHERE iteration_id=${pid}$ AND created_by=${uid}$ "
+                            "ORDER BY created_at ASC LIMIT ${lim}$",
+                            {"pid": pid, "uid": self.user_id or "", "lim": self.config.history_limit},
+                        )
+                    else:
+                        recs = await sor.sqlExe(
+                            "SELECT role, content FROM pipeline_conversations "
+                            "WHERE created_by=${uid}$ "
+                            "ORDER BY created_at ASC LIMIT ${lim}$",
+                            {"uid": self.user_id or "", "lim": self.config.history_limit},
+                        )
 
                 if not recs:
                     return []
@@ -1263,6 +1279,7 @@ class AgentExecutor:
                     "content": user_input,
                     "created_by": self.user_id or "",
                     "iteration_id": self.project_id or "",
+                    "session_id": self.session_id or "",
                 })
                 # 助手回复（不含 tool_call JSON！）
                 reply_clean = reply
@@ -1274,6 +1291,7 @@ class AgentExecutor:
                     "content": reply_clean[:4000],
                     "created_by": self.user_id or "",
                     "iteration_id": self.project_id or "",
+                    "session_id": self.session_id or "",
                 })
         except Exception as e:
             logger.warning(f"Session save failed: {e}")

pipeline_service/gateway.py

@@ -28,9 +28,10 @@ logger = logging.getLogger("pipeline.gateway")
 
 @dataclass
 class GatewaySession:
-    """一个会话 = (channel, user_id) 的 agent 上下文（元数据，executor 每次新建）。"""
+    """一个会话 = (channel, user_id, session_id) 的 agent 上下文（元数据，executor 每次新建）。"""
     channel: str
     user_id: str
+    session_id: str = ""          # 会话内唯一标识（web 多 tab 独立会话）；空 = 单会话（向后兼容）
     project_id: str = ""
     pipeline_id: str = ""
     project_name: str = ""
@@ -60,15 +61,17 @@ class Gateway:
     # ── 会话生命周期 ──
 
     @staticmethod
-    def _session_key(channel: str, user_id: str) -> str:
+    def _session_key(channel: str, user_id: str, session_id: str = "") -> str:
+        if session_id:
+            return f"{channel}:{user_id}:{session_id}"
         return f"{channel}:{user_id}"
 
-    def get_session(self, channel: str, user_id: str) -> Optional[GatewaySession]:
+    def get_session(self, channel: str, user_id: str, session_id: str = "") -> Optional[GatewaySession]:
-        return self._sessions.get(self._session_key(channel, user_id))
+        return self._sessions.get(self._session_key(channel, user_id, session_id))
 
-    def clear_session(self, channel: str, user_id: str) -> None:
+    def clear_session(self, channel: str, user_id: str, session_id: str = "") -> None:
         """清除会话（/reset 时用）。"""
-        self._sessions.pop(self._session_key(channel, user_id), None)
+        self._sessions.pop(self._session_key(channel, user_id, session_id), None)
 
     # ── 上下文解析 ──
 
@@ -112,7 +115,7 @@ class Gateway:
 
     async def run_message(self, channel: str, user_id: str, content: str,
                           history=None, role: str = "", generic: bool = False,
-                          base_url: str = "") -> AsyncGenerator[str, None]:
+                          base_url: str = "", session_id: str = "") -> AsyncGenerator[str, None]:
         """统一消息入口：解析上下文 → 加载产线能力 → AgentExecutor 执行 → yield 事件流。
 
         Args:
@@ -122,6 +125,7 @@ class Gateway:
             history: 可选历史消息
             role: 可选产线内角色（驾驶舱 agent 为空）
             generic: True = 纯通用会话（不解析项目、不加载产线能力，只用 GENERAL_TOOLS + 通用心智）
+            session_id: 会话内唯一标识（web 多 tab 独立会话用；空 = 单会话）
         """
         from pipeline_core.agent_config import load_agent_config
         from .agent_loop_v2 import AgentExecutor
@@ -132,13 +136,13 @@ class Gateway:
         else:
             ctx = await self.resolve_project(user_id)
 
-        # 2. 会话生命周期（跟踪 project 切换）
+        # 2. 会话生命周期（跟踪 project 切换；session_id 区分多 tab 会话）
-        key = self._session_key(channel, user_id)
+        key = self._session_key(channel, user_id, session_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,
+                channel=channel, user_id=user_id, session_id=session_id,
                 project_id=ctx["pid"], pipeline_id=ctx["pipeline_id"],
                 project_name=ctx["name"], created_at=now, updated_at=now,
             )
@@ -157,7 +161,7 @@ class Gateway:
             config.model_name = ctx["default_llm_name"]
         executor = AgentExecutor(
             config=config, project_id=ctx["pid"], user_id=user_id, role=role,
-            base_url=base_url, generic=generic, session=sess)
+            base_url=base_url, generic=generic, session=sess, session_id=session_id)
 
         # 4. 转发标准化事件流
         async for chunk in executor.run(content, history=history):