feat(session): 当前项目按会话隔离(session_id)，多tab各自项目互不覆盖

- workspace.get_session_project_id: 按 session_id 读 pipeline_session_settings，回退全局
- get_workspace_dir/get_workspace_path 支持 session_id 参数
- gateway.resolve_project 按会话解析 pid（default_llm 仍按用户全局）
- _persist_project: 有 session_id 时写 pipeline_session_settings + 全局兜底

pipeline_service/agent_loop_v2.py

@@ -767,15 +767,19 @@ class AgentExecutor:
     # ── 工具实现 ──
 
     async def _persist_project(self, sor, pid):
-        """持久化当前项目到 pipeline_agent_settings（user_id 唯一键）。
+        """持久化当前项目。
 
+        - 有 session_id：写 pipeline_session_settings（web 多 tab 各自项目上下文，互不覆盖），
+          同时写全局 pipeline_agent_settings.current_project_id 作为「最近项目」兜底（供无 session 消费者）。
+        - 无 session_id：只写全局（向后兼容，如微信通道 / run_agent 直连）。
         只改 self.project_id 不够——AgentExecutor 每轮新建，run 结束即销毁，
-        下一轮 cockpit_chat_v2 又从 pipeline_agent_settings 读回旧项目。
+        下一轮又从持久层读回旧项目。
         """
         if not self.user_id or not pid:
             return
         try:
             from appPublic.uniqueID import getID
+            # 全局「最近项目」兜底
             await sor.sqlExe(
                 "UPDATE pipeline_agent_settings SET current_project_id=${pid}$ "
                 "WHERE user_id=${uid}$",
@@ -788,6 +792,26 @@ class AgentExecutor:
                     "INSERT INTO pipeline_agent_settings (id, user_id, current_project_id) "
                     "VALUES (${id}$, ${uid}$, ${pid}$)",
                     {"id": getID(), "uid": self.user_id, "pid": pid})
+            # 会话级项目上下文（多 tab 隔离）
+            if self.session_id:
+                try:
+                    await sor.sqlExe(
+                        "UPDATE pipeline_session_settings SET current_project_id=${pid}$ "
+                        "WHERE user_id=${uid}$ AND session_id=${sid}$",
+                        {"pid": pid, "uid": self.user_id, "sid": self.session_id})
+                    sexists = await sor.sqlExe(
+                        "SELECT 1 FROM pipeline_session_settings "
+                        "WHERE user_id=${uid}$ AND session_id=${sid}$",
+                        {"uid": self.user_id, "sid": self.session_id})
+                    if not sexists:
+                        await sor.sqlExe(
+                            "INSERT INTO pipeline_session_settings "
+                            "(id, user_id, session_id, current_project_id) "
+                            "VALUES (${id}$, ${uid}$, ${sid}$, ${pid}$)",
+                            {"id": getID(), "uid": self.user_id,
+                             "sid": self.session_id, "pid": pid})
+                except Exception as e:
+                    logger.warning(f"persist session project failed: {e}")
         except Exception as e:
             logger.warning(f"persist project failed: {e}")
 

pipeline_service/gateway.py

@@ -75,23 +75,27 @@ class Gateway:
 
     # ── 上下文解析 ──
 
-    async def resolve_project(self, user_id: str) -> dict:
+    async def resolve_project(self, user_id: str, session_id: str = "") -> dict:
         """解析用户当前项目 → {pid, pipeline_id, name, default_llm_id, default_llm_name}。
 
-        同时读出个人选择的模型（default_llm_id → llm.name），供 run_message 覆盖 config.model_name。
+        pid 按会话隔离（session_id 非空时读 pipeline_session_settings，无记录回退全局）；
+        同时读出个人选择的模型（default_llm_id → llm.name，仍按用户全局），供 run_message 覆盖 config.model_name。
         """
         ctx = {"pid": "", "pipeline_id": "", "name": "", "default_llm_id": "", "default_llm_name": ""}
         if not user_id:
             return ctx
         try:
             from sqlor.dbpools import DBPools
+            from .workspace import get_session_project_id
             db = DBPools()
             async with db.sqlorContext("pipeline") as sor:
+                # 当前项目：按会话隔离（多 tab 各自项目，互不覆盖）
+                ctx["pid"] = await get_session_project_id(sor, user_id, session_id) or ""
+                # 个人模型选择：仍按用户全局读
                 recs = await sor.sqlExe(
-                    "SELECT current_project_id, default_llm_id FROM pipeline_agent_settings WHERE user_id=${u}$",
+                    "SELECT default_llm_id FROM pipeline_agent_settings WHERE user_id=${u}$",
                     {"u": user_id})
                 if recs:
-                    ctx["pid"] = getattr(recs[0], "current_project_id", "") or ""
                     ctx["default_llm_id"] = getattr(recs[0], "default_llm_id", "") or ""
                     # 单独查 llm.name（避免 JOIN 触发两表字段 collation 不一致）
                     if ctx["default_llm_id"]:
@@ -134,7 +138,7 @@ class Gateway:
         if generic:
             ctx = {"pid": "", "pipeline_id": "", "name": ""}
         else:
-            ctx = await self.resolve_project(user_id)
+            ctx = await self.resolve_project(user_id, session_id)
 
         # 2. 会话生命周期（跟踪 project 切换；session_id 区分多 tab 会话）
         key = self._session_key(channel, user_id, session_id)

pipeline_service/workspace.py

@@ -90,18 +90,43 @@ async def get_max_task_retry(sor):
         return 3
 
 
-async def get_workspace_dir(sor, uid):
+async def get_session_project_id(sor, uid, session_id=''):
+    """按会话解析当前项目 id。
+
+    session_id 非空时优先读 pipeline_session_settings（web 多 tab 各自项目上下文），
+    无记录或表不存在时回退全局 pipeline_agent_settings.current_project_id。
+    """
+    if session_id:
+        try:
+            recs = await sor.sqlExe(
+                "SELECT current_project_id FROM pipeline_session_settings "
+                "WHERE user_id=${u}$ AND session_id=${s}$",
+                {"u": uid, "s": session_id})
+            if recs:
+                pid = getattr(recs[0], 'current_project_id', '') or ''
+                if pid:
+                    return pid
+        except Exception:
+            pass
+    try:
+        recs = await sor.sqlExe(
+            "SELECT current_project_id FROM pipeline_agent_settings WHERE user_id=${u}$",
+            {"u": uid})
+        return getattr(recs[0], 'current_project_id', '') if recs else ''
+    except Exception:
+        return ''
+
+
+async def get_workspace_dir(sor, uid, session_id=''):
     """读当前项目的 workspace 目录（统一 workspace_*.dspy 的重复逻辑）。
 
     返回 (ws_dir, workspace_base)。ws_dir 优先用 sd_projects.workspace_dir（绝对路径），
     否则 base/org/name；无当前项目时 ws_dir 返回空字符串。
+    session_id 非空时按会话隔离项目上下文（多 tab 各自项目，互不覆盖）。
     """
     workspace_base = await get_workspace_base(sor)
 
-    recs = await sor.sqlExe(
+    pid = await get_session_project_id(sor, uid, session_id)
-        "SELECT current_project_id FROM pipeline_agent_settings WHERE user_id=${u}$",
-        {"u": uid})
-    pid = getattr(recs[0], 'current_project_id', '') if recs else ''
     if not pid:
         return '', workspace_base
 
@@ -120,9 +145,9 @@ async def get_workspace_dir(sor, uid):
     return build_workspace_path(workspace_base, org_id, space, pname, pid), workspace_base
 
 
-async def get_workspace_path(sor, user_id):
+async def get_workspace_path(sor, user_id, session_id=''):
     """获取用户当前项目的工作空间路径（无项目时返回 None）。"""
-    ws, _ = await get_workspace_dir(sor, user_id)
+    ws, _ = await get_workspace_dir(sor, user_id, session_id)
     return ws or None