diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 11a4d8c..470a9d7 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/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}") diff --git a/pipeline_service/gateway.py b/pipeline_service/gateway.py index 5b35779..872598f 100644 --- a/pipeline_service/gateway.py +++ b/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) diff --git a/pipeline_service/workspace.py b/pipeline_service/workspace.py index a7caed6..d3c885b 100644 --- a/pipeline_service/workspace.py +++ b/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( - "SELECT current_project_id FROM pipeline_agent_settings WHERE user_id=${u}$", - {"u": uid}) - pid = getattr(recs[0], 'current_project_id', '') if recs else '' + pid = await get_session_project_id(sor, uid, session_id) 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