From 2d9b4eb5a6f49da1708474b7f8e583d8a887ce23 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Tue, 25 Aug 2026 22:43:58 +0800 Subject: [PATCH] =?UTF-8?q?fix(conversation):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E7=89=88=E6=9C=AC=E5=9B=9E=E6=BA=AF=E8=A2=AB=E6=97=A7=E5=86=85?= =?UTF-8?q?=E5=AD=98merge=E8=A6=86=E7=9B=96=EF=BC=8C=E7=BB=9F=E4=B8=80pres?= =?UTF-8?q?end=E7=94=A8=E6=88=B7=E8=BE=93=E5=85=A5=E5=8F=A3=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 回溯后同步对话级 terminal 内存为磁盘快照，避免 merge-on-save 把被裁消息当"内存独有"救回（此前只能回溯后立刻重启进程） - 工作区级服务实例不再挂载对话历史（focus-only 仅恢复焦点+模式），settings/通知判定/工作流游标等保存路径改读磁盘回传，消除旧内存写回污染源 - 左侧输入导航与深度压缩的用户输入口径补 presend（提前输入排队转正），guidance 维持不计入；统一字段缺失老消息视为 user --- core/web_terminal.py | 26 ++++++++-- server/chat/settings.py | 55 +++++++++++++-------- server/conversation.py | 55 +++++++++++++++++---- server/deep_compression.py | 6 ++- server/workflow_runtime_api.py | 8 ++- static/src/components/chat/ChatArea.vue | 7 ++- utils/context_manager/conversation_mixin.py | 36 +++++++++++--- 7 files changed, 147 insertions(+), 46 deletions(-) diff --git a/core/web_terminal.py b/core/web_terminal.py index b4d1fc91..8338bbb4 100644 --- a/core/web_terminal.py +++ b/core/web_terminal.py @@ -39,8 +39,10 @@ class WebTerminal(MainTerminal): 对话级 terminal 只服务绑定对话，必须恢复该对话保存的模型 （restore_model=True），否则重启后新建的对话级 terminal 会用默认 模型跑任务/回写 metadata，把用户切换的模型覆盖掉。 - 工作区级 terminal（未绑定）：保持原有“最近对话”行为，且刻意不恢复 - 模型（restore_model=False），避免 /new 页面显示旧对话的模型。 + 工作区级 terminal（未绑定）：保持原有“最近对话”行为（恢复焦点与模式）， + 且刻意不恢复模型（restore_model=False），避免 /new 页面显示旧对话的模型。 + 经由 load_conversation 的 attach_history 分流，工作区级不挂载消息历史—— + 历史权威在磁盘 + 对话级实例（防 merge-on-save 旧内存写回污染源）。 """ if self.context_manager.current_conversation_id: return @@ -107,6 +109,15 @@ class WebTerminal(MainTerminal): container_session=container_session, usage_tracker=usage_tracker ) + + # 工作区级服务实例标记：不持有/不回写对话消息历史（历史权威在磁盘 + 对话级实例）。 + # 工作区级挂载历史只会成为 merge-on-save 的污染源（版本回溯被旧内存“救回”覆盖的事故根因）。 + # 注意必须在 super().__init__ 之后设置（context_manager 由父类创建）； + # 而 load 分流不依赖此标记（用 _bound_conversation_id，super 之前已设），初始化时序安全。 + try: + self.context_manager._service_instance_no_history = conversation_id is None + except Exception: + pass # Web特有属性 self.message_callback = message_callback @@ -417,7 +428,12 @@ class WebTerminal(MainTerminal): Dict: 加载结果 """ try: - success = self.context_manager.load_conversation_by_id(conversation_id) + # 工作区级服务实例（_bound_conversation_id 为空）：仅恢复会话焦点 + # （current_conversation_id）与运行模式，不挂载消息历史——历史权威在磁盘 + # 与对话级实例，服务实例持历史只会成为 merge-on-save 的写回污染源。 + # 对话级实例（绑定对话）：挂载历史，供任务执行链路使用。 + attach_history = bool(getattr(self, "_bound_conversation_id", None)) + success = self.context_manager.load_conversation_by_id(conversation_id, attach_history=attach_history) if success: # 根据对话元数据同步运行模式与推理强度 try: @@ -516,7 +532,8 @@ class WebTerminal(MainTerminal): "success": True, "conversation_id": conversation_id, "title": conversation_data.get("title", "未知对话"), - "messages_count": len(self.context_manager.conversation_history), + # 消息数以磁盘数据为准：服务实例不挂载历史，len(conversation_history) 恒为 0 + "messages_count": len(conversation_data.get("messages") or []), "run_mode": self.run_mode, "thinking_mode": self.thinking_mode, "model_key": getattr(self, "model_key", None), @@ -630,6 +647,7 @@ class WebTerminal(MainTerminal): "context": { "usage_percent": context_status['usage_percent'], "total_size": context_status['sizes']['total'], + # 本实例内存上下文中的消息数；工作区级服务实例不挂载历史，恒为 0（前端不消费该字段） "conversation_count": len(self.context_manager.conversation_history) }, "focused_files": focused_files_dict, diff --git a/server/chat/settings.py b/server/chat/settings.py index efde868b..0c68082a 100644 --- a/server/chat/settings.py +++ b/server/chat/settings.py @@ -47,6 +47,23 @@ from server.monitor import get_cached_monitor_snapshot UPLOAD_FOLDER_NAME = ".astrion/user_upload" + +def _save_conversation_meta_via_disk(ctx, conversation_id: str, **meta_kwargs) -> None: + """设置类端点保存对话元数据的统一入口。 + + 工作区级服务实例不挂载消息历史（conversation_history 恒空），直接拿内存保存会把 + 空/陈旧历史写回；这里读磁盘消息回传（merge 语义下消息保持不变），仅更新元数据。 + """ + if not conversation_id: + return + manager = ctx._get_conversation_manager_for_id(conversation_id) + existing = manager.load_conversation(conversation_id) or {} + manager.save_conversation( + conversation_id=conversation_id, + messages=existing.get("messages") or [], + **meta_kwargs, + ) + @chat_bp.route('/api/thinking-mode', methods=['POST']) @api_login_required @with_terminal @@ -67,20 +84,19 @@ def update_thinking_mode(terminal: WebTerminal, workspace: UserWorkspace, userna terminal.set_run_mode(target_mode) session['thinking_mode'] = terminal.thinking_mode session['run_mode'] = terminal.run_mode - # 更新当前对话的元数据 + # 更新当前对话的元数据（服务实例不挂载历史：读磁盘回传，仅更新元数据） ctx = terminal.context_manager if ctx.current_conversation_id: try: - ctx.conversation_manager.save_conversation( - conversation_id=ctx.current_conversation_id, - messages=ctx.conversation_history, + _save_conversation_meta_via_disk( + ctx, + ctx.current_conversation_id, project_path=str(ctx.project_path), - todo_list=ctx.todo_list, thinking_mode=terminal.thinking_mode, run_mode=terminal.run_mode, model_key=getattr(terminal, "model_key", None), has_images=getattr(ctx, "has_images", False), - has_videos=getattr(ctx, "has_videos", False) + has_videos=getattr(ctx, "has_videos", False), ) except Exception as exc: logger.error(f"[API] 保存思考模式到对话失败: {exc}") @@ -123,21 +139,20 @@ def update_reasoning_effort(terminal: WebTerminal, workspace: UserWorkspace, use if is_current_target: terminal.set_reasoning_effort(effort) save_conversation_id = target_conversation_id or ctx.current_conversation_id - # 更新对话的元数据 + # 更新对话的元数据（服务实例不挂载历史：统一读磁盘回传，仅更新元数据） if save_conversation_id: try: if is_current_target: - ctx.conversation_manager.save_conversation( - conversation_id=save_conversation_id, - messages=ctx.conversation_history, + _save_conversation_meta_via_disk( + ctx, + save_conversation_id, project_path=str(ctx.project_path), - todo_list=ctx.todo_list, thinking_mode=terminal.thinking_mode, run_mode=terminal.run_mode, reasoning_effort=effort, model_key=getattr(terminal, "model_key", None), has_images=getattr(ctx, "has_images", False), - has_videos=getattr(ctx, "has_videos", False) + has_videos=getattr(ctx, "has_videos", False), ) else: # 所属对话已不是当前对话：只更新推理强度 meta。 @@ -212,17 +227,16 @@ def update_model(terminal: WebTerminal, workspace: UserWorkspace, username: str) if requested_cid and current_cid: if current_cid == requested_cid: try: - ctx.conversation_manager.save_conversation( - conversation_id=current_cid, - messages=ctx.conversation_history, + _save_conversation_meta_via_disk( + ctx, + current_cid, project_path=str(ctx.project_path), - todo_list=ctx.todo_list, thinking_mode=terminal.thinking_mode, run_mode=terminal.run_mode, reasoning_effort=terminal.reasoning_effort, model_key=terminal.model_key, has_images=getattr(ctx, "has_images", False), - has_videos=getattr(ctx, "has_videos", False) + has_videos=getattr(ctx, "has_videos", False), ) except Exception as exc: logger.error(f"[API] 保存模型到对话失败: {exc}") @@ -338,11 +352,10 @@ def update_personalization_settings(terminal: WebTerminal, workspace: UserWorksp ctx = getattr(terminal, 'context_manager', None) if ctx and getattr(ctx, 'current_conversation_id', None): try: - ctx.conversation_manager.save_conversation( - conversation_id=ctx.current_conversation_id, - messages=ctx.conversation_history, + _save_conversation_meta_via_disk( + ctx, + ctx.current_conversation_id, project_path=str(ctx.project_path), - todo_list=ctx.todo_list, thinking_mode=terminal.thinking_mode, run_mode=terminal.run_mode ) diff --git a/server/conversation.py b/server/conversation.py index 332370f8..ebc0f16e 100644 --- a/server/conversation.py +++ b/server/conversation.py @@ -214,6 +214,41 @@ def _normalize_conv_id(conversation_id: str) -> str: return conv if conv.startswith("conv_") else f"conv_{conv}" +def _sync_restored_conversation_memory(conversation_id: str) -> None: + """版本回溯后同步内存实例：把绑定该对话的对话级 terminal 内存替换为磁盘最新。 + + 回溯用 allow_shrink 覆写裁短磁盘；若持有旧（更长）历史的对话级实例之后保存， + merge-on-save 会把被裁消息当作「内存独有」追加救回，回溯随即被撤销 + （此前用户只能回溯后立刻重启进程的根因）。工作区级服务实例不挂载历史， + 天然无需处理（见 WebTerminal.load_conversation 的 attach_history 分流）。 + """ + from copy import deepcopy + from server import state as server_state + + normalized = _normalize_conv_id(conversation_id) + user_terminals = getattr(server_state, "user_terminals", None) or {} + for term_key, term in list(user_terminals.items()): + try: + bound = getattr(term, "_bound_conversation_id", None) + if not bound or _normalize_conv_id(str(bound)) != normalized: + continue + ctx = getattr(term, "context_manager", None) + if ctx is None: + continue + target_manager = ctx._get_conversation_manager_for_id(normalized) + data = target_manager.load_conversation(normalized) or {} + ctx.conversation_history = list(data.get("messages") or []) + ctx.conversation_metadata = deepcopy(data.get("metadata") or {}) + todo_data = data.get("todo_list") + ctx.todo_list = deepcopy(todo_data) if todo_data else None + debug_log( + f"[Versioning][Restore] synced in-memory history term={term_key} " + f"messages={len(ctx.conversation_history)}" + ) + except Exception as exc: + debug_log(f"[Versioning][Restore] sync memory failed for term={term_key}: {exc}") + + def _normalize_versioning_tracking_mode(value: Optional[str]) -> str: return ConversationVersioningManager.normalize_tracking_mode(value) @@ -1632,16 +1667,13 @@ def restore_conversation_versioning_checkpoint(conversation_id, terminal: WebTer terminal, workspace, normalized_id, target_conversation_id, seq, tracking_mode, host_mode, backup_mode ) - # overwrite 场景需要清空内存历史避免反向覆写；copy 场景目标是新对话，无需处理。 + # overwrite 场景：回溯已裁短磁盘，必须同步所有持有该对话的内存实例（对话级 + # terminal 常驻 24h），否则旧内存后续保存经 merge 会把被裁消息「救回」，回溯被撤销。 + # copy 场景目标是新对话，无缓存实例，无需处理。 if restore_mode == "overwrite": - try: - current_loaded_id = getattr(terminal.context_manager, "current_conversation_id", None) - if current_loaded_id == normalized_id: - terminal.context_manager.conversation_history = [] - debug_log(f"[Versioning][Restore] cleared in-memory history before reload conv={normalized_id}") - except Exception as exc: - debug_log(f"[Versioning][Restore] clear in-memory history failed: {exc}") + _sync_restored_conversation_memory(target_conversation_id) + # 恢复模式与焦点到当前（工作区级）terminal；服务实例不挂载历史，仅读磁盘元数据 terminal.load_conversation(target_conversation_id) debug_log( f"[Versioning][Restore] reload done conv={target_conversation_id} " @@ -1898,7 +1930,12 @@ def list_sub_agents(terminal: WebTerminal, workspace: UserWorkspace, username: s announced = terminal._announced_sub_agent_tasks notified_from_history = set() try: - history = getattr(terminal.context_manager, "conversation_history", []) or [] + # 服务实例（工作区级）不挂载历史：通知去重标记以磁盘对话为准 + history = [] + if conversation_id: + notify_manager = terminal.context_manager._get_conversation_manager_for_id(conversation_id) + notify_conv_data = notify_manager.load_conversation(conversation_id) or {} + history = notify_conv_data.get("messages") or [] for msg in history: meta = msg.get("metadata") or {} task_id = meta.get("task_id") diff --git a/server/deep_compression.py b/server/deep_compression.py index 8b50d9a7..2f316f3c 100644 --- a/server/deep_compression.py +++ b/server/deep_compression.py @@ -340,12 +340,16 @@ def _extract_text_only(content: Any) -> str: def _collect_user_texts(messages: List[Dict[str, Any]]) -> List[str]: + # 口径：用户亲手输入 = message_source 为 'user'（直接发送）或 'presend'（提前输入排队后转正）， + # 字段缺失的老消息视为 'user'；guidance（手动引导）/notify/compression 等运行期注入消息不计入。 + # 与前端左侧输入导航 isRealUserNavMessage（ChatArea.vue）口径保持一致。 result: List[str] = [] for msg in messages: if msg.get("role") != "user": continue metadata = msg.get("metadata") if isinstance(msg.get("metadata"), dict) else {} - if str(metadata.get("message_source") or "").strip().lower() != "user": + source = str(metadata.get("message_source") or "user").strip().lower() + if source not in ("user", "presend"): continue text = _extract_text_only(msg.get("content")) if text.strip(): diff --git a/server/workflow_runtime_api.py b/server/workflow_runtime_api.py index 8eac6adc..c6d2d187 100644 --- a/server/workflow_runtime_api.py +++ b/server/workflow_runtime_api.py @@ -120,12 +120,16 @@ def api_activate_workflow(terminal, workspace, username): try: # 激活提示消息将追加到历史末尾，阶段消息游标取当前长度 + 1。 - # 新建对话：激活消息是第 1 条，游标固定为 1（服务 terminal 的 history 长度不可信）。 + # 新建对话：激活消息是第 1 条，游标固定为 1。 + # 服务实例（工作区级）不挂载历史：非新建时游标以磁盘消息数为准。 if created_new: msg_index = 1 else: try: - msg_index = len(getattr(terminal.context_manager, "conversation_history", []) or []) + 1 + cm_router = getattr(terminal, "context_manager", None) + wf_manager = cm_router._get_conversation_manager_for_id(conversation_id) if cm_router else None + conv_data = wf_manager.load_conversation(conversation_id) if wf_manager else None + msg_index = len((conv_data or {}).get("messages") or []) + 1 except Exception: msg_index = 0 result = activate_workflow( diff --git a/static/src/components/chat/ChatArea.vue b/static/src/components/chat/ChatArea.vue index 15800440..5724140b 100644 --- a/static/src/components/chat/ChatArea.vue +++ b/static/src/components/chat/ChatArea.vue @@ -2397,7 +2397,9 @@ function getGeneratingLetters(message: any) { } // ===== 左侧用户输入快捷跳转导航 ===== -// 数据口径与深度压缩一致：role === 'user' 且 metadata.message_source === 'user'（引导/通知/压缩等不计入） +// 口径：用户亲手输入 = metadata.message_source 为 'user'（直接发送）或 'presend'（提前输入排队后转正）， +// 字段缺失的老消息视为 'user'；guidance（手动引导）/notify/compression 等运行期注入消息不计入。 +// 与深度压缩 _collect_user_texts（server/deep_compression.py）口径保持一致。 const shellRef = ref(null); const virtualizerRef = ref(null); const quickNavActiveIndex = ref(-1); @@ -2407,7 +2409,8 @@ const isRealUserNavMessage = (msg: any): boolean => { if (!msg || msg.role !== 'user') { return false; } - return String(msg?.metadata?.message_source || 'user').trim().toLowerCase() === 'user'; + const source = String(msg?.metadata?.message_source || 'user').trim().toLowerCase(); + return source === 'user' || source === 'presend'; }; function extractUserFirstLine(msg: any): string { diff --git a/utils/context_manager/conversation_mixin.py b/utils/context_manager/conversation_mixin.py index a1edd32a..c0251bb3 100644 --- a/utils/context_manager/conversation_mixin.py +++ b/utils/context_manager/conversation_mixin.py @@ -179,12 +179,16 @@ class ConversationMixin: print(f"📝 开始新对话: {conversation_id}") return conversation_id - def load_conversation_by_id(self, conversation_id: str) -> bool: + def load_conversation_by_id(self, conversation_id: str, attach_history: bool = True) -> bool: """ 加载指定对话 Args: conversation_id: 对话ID + attach_history: 是否把消息历史/元数据挂载到内存。工作区级服务实例应传 False + （仅恢复焦点 current_conversation_id 与运行模式）：对话历史的权威载体是 + 磁盘 + 对话级 terminal，服务实例挂载历史只会成为 merge-on-save 的污染源 + （版本回溯被旧内存“救回”覆盖即此机制事故）。 Returns: bool: 加载是否成功 @@ -221,12 +225,18 @@ class ConversationMixin: # 更新当前状态 self.current_conversation_id = conversation_id - self.conversation_history = conversation_data.get("messages", []) - todo_data = conversation_data.get("todo_list") - self.todo_list = deepcopy(todo_data) if todo_data else None - self.conversation_metadata = deepcopy(conversation_data.get("metadata", {}) or {}) + if attach_history: + self.conversation_history = conversation_data.get("messages", []) + todo_data = conversation_data.get("todo_list") + self.todo_list = deepcopy(todo_data) if todo_data else None + self.conversation_metadata = deepcopy(conversation_data.get("metadata", {}) or {}) + else: + # 服务实例：仅焦点，不挂载消息/元数据（保持空内存，merge 语义下任何保存都不会回写消息） + self.conversation_history = [] + self.todo_list = None + self.conversation_metadata = {} # 恢复项目文件树快照（如已存在） - meta = self.conversation_metadata + meta = deepcopy(conversation_data.get("metadata", {}) or {}) if meta.get("project_file_tree"): self.project_snapshot = { "file_tree": meta.get("project_file_tree"), @@ -264,7 +274,7 @@ class ConversationMixin: model_key = metadata.get("model_key") self.has_images = metadata.get("has_images", False) self.has_videos = metadata.get("has_videos", False) - if not self.has_images or not self.has_videos: + if attach_history and (not self.has_images or not self.has_videos): for msg in self.conversation_history: if not isinstance(msg, dict): continue @@ -321,6 +331,13 @@ class ConversationMixin: return True + def _is_service_instance_no_history(self) -> bool: + """是否为「不持有对话历史」的工作区级服务实例（由 WebTerminal 初始化时打标）。 + + 服务实例不挂载/不回写消息历史：历史权威在磁盘与对话级 terminal。 + """ + return bool(getattr(self, "_service_instance_no_history", False)) + def save_current_conversation(self) -> bool: """ 保存当前对话 @@ -328,6 +345,9 @@ class ConversationMixin: Returns: bool: 保存是否成功 """ + if self._is_service_instance_no_history(): + # 服务实例不持有历史，保存无语义；直接跳过，防止任何路径把空/陈旧内存写回 + return False if not self.current_conversation_id: print("⚠️ 没有当前对话ID，无法保存") return False @@ -362,6 +382,8 @@ class ConversationMixin: def auto_save_conversation(self, force: bool = False): """自动保存对话（静默模式，减少日志输出）""" + if self._is_service_instance_no_history(): + return if not self.auto_save_enabled or not self.current_conversation_id: return if not force and not self.conversation_history: