From 3c4ae917b2d809fd6c588dcf34083a4266dc34a0 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Mon, 13 Jul 2026 21:43:51 +0800 Subject: [PATCH] =?UTF-8?q?fix(web=5Fterminal):=20=E7=A8=8B=E5=BA=8F?= =?UTF-8?q?=E5=90=AF=E5=8A=A8=E8=87=AA=E5=8A=A8=E6=81=A2=E5=A4=8D=E6=9C=80?= =?UTF-8?q?=E8=BF=91=E5=AF=B9=E8=AF=9D=E6=97=B6=E4=B8=8D=E6=81=A2=E5=A4=8D?= =?UTF-8?q?=E6=A8=A1=E5=9E=8B=EF=BC=8C=E9=81=BF=E5=85=8D=20/new=20?= =?UTF-8?q?=E9=A1=B5=E9=9D=A2=E6=98=BE=E7=A4=BA=E6=97=A7=E5=AF=B9=E8=AF=9D?= =?UTF-8?q?=E6=A8=A1=E5=9E=8B=EF=BC=9B=E6=98=BE=E5=BC=8F=E5=8A=A0=E8=BD=BD?= =?UTF-8?q?=E5=AF=B9=E8=AF=9D=E6=97=B6=E4=BB=8D=E6=81=A2=E5=A4=8D=E4=BF=9D?= =?UTF-8?q?=E5=AD=98=E7=9A=84=E6=A8=A1=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/web_terminal.py | 19 ++- test/test_conversation_model_persistence.py | 154 ++++++++++++++++++++ 2 files changed, 165 insertions(+), 8 deletions(-) create mode 100644 test/test_conversation_model_persistence.py diff --git a/core/web_terminal.py b/core/web_terminal.py index 43f40a7..7cbd9f8 100644 --- a/core/web_terminal.py +++ b/core/web_terminal.py @@ -41,7 +41,7 @@ class WebTerminal(MainTerminal): latest = conversations[0] conv_id = latest.get("id") if conv_id: - result = self.load_conversation(conv_id) + result = self.load_conversation(conv_id, restore_model=False) if result.get("success"): print(f"[WebTerminal] 已加载最近对话: {conv_id}") return @@ -311,12 +311,14 @@ class WebTerminal(MainTerminal): ) perf_log("_ensure_conversation_versioning_enabled done", elapsed_ms=(time.perf_counter() - t0) * 1000, extra={"conv_id": normalized_id}) - def load_conversation(self, conversation_id: str) -> Dict: + def load_conversation(self, conversation_id: str, restore_model: bool = True) -> Dict: """ 加载指定对话（Web版本） Args: conversation_id: 对话ID + restore_model: 是否从对话 metadata 恢复模型。程序启动时自动恢复最近对话不恢复模型， + 避免 /new 页面显示旧对话模型；用户显式加载对话时恢复。 Returns: Dict: 加载结果 @@ -351,12 +353,13 @@ class WebTerminal(MainTerminal): self.set_network_permission(network_permission) except Exception: pass - saved_model_key = meta.get("model_key") - if saved_model_key: - try: - self.set_model(saved_model_key) - except Exception as exc: - logger.warning("加载对话模型 %s 失败: %s", saved_model_key, exc) + if restore_model: + saved_model_key = meta.get("model_key") + if saved_model_key: + try: + self.set_model(saved_model_key) + except Exception as exc: + logger.warning("加载对话模型 %s 失败: %s", saved_model_key, exc) self.pending_permission_mode = str(meta.get("pending_permission_mode") or "").strip().lower() or None self.pending_execution_mode = str(meta.get("pending_execution_mode") or "").strip().lower() or None self.pending_network_permission = str(meta.get("pending_network_permission") or "").strip().lower() or None diff --git a/test/test_conversation_model_persistence.py b/test/test_conversation_model_persistence.py new file mode 100644 index 0000000..7896e49 --- /dev/null +++ b/test/test_conversation_model_persistence.py @@ -0,0 +1,154 @@ +"""回归测试：对话模型持久化与 /new 页面默认模型行为。""" +import os +import sys +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from server.context import _apply_workspace_personalization_preferences + + +class FakeSession: + def __init__(self): + self._data = {} + + def get(self, key, default=None): + return self._data.get(key, default) + + def __setitem__(self, key, value): + self._data[key] = value + + +class TestApplyWorkspacePersonalizationPreferences(unittest.TestCase): + def _make_terminal(self, model_key="default-model"): + terminal = MagicMock() + terminal.model_key = model_key + terminal._workspace_default_model_applied = False + + def _apply_personalization_preferences(config, apply_default_model=True): + if apply_default_model: + default_model = (config or {}).get("default_model") + if default_model: + terminal.model_key = default_model + terminal.apply_personalization_preferences = MagicMock(side_effect=_apply_personalization_preferences) + return terminal + + def _make_workspace(self): + workspace = MagicMock() + workspace.data_dir = tempfile.mkdtemp() + return workspace + + @patch("server.context.load_personalization_config") + @patch("server.context.has_request_context", return_value=True) + def test_session_model_restored(self, _hrc, mock_load_config): + """session 中保存了模型时，应恢复到该模型。""" + mock_load_config.return_value = {"default_model": "default-model"} + terminal = self._make_terminal(model_key="old-model") + terminal.set_model = MagicMock(side_effect=lambda mk: setattr(terminal, "model_key", mk)) + workspace = self._make_workspace() + session = FakeSession() + session["model_key"] = "session-model" + + with patch("server.context.session", session): + _apply_workspace_personalization_preferences(terminal, workspace) + + terminal.set_model.assert_called_once_with("session-model") + self.assertEqual(session.get("model_key"), "session-model") + + @patch("server.context.load_personalization_config") + @patch("server.context.has_request_context", return_value=True) + def test_default_model_applied_for_fresh_session(self, _hrc, mock_load_config): + """没有 session 模型时，应应用默认模型（且仅一次）。""" + mock_load_config.return_value = {"default_model": "default-model"} + terminal = self._make_terminal(model_key="kimi-k2.6") + workspace = self._make_workspace() + session = FakeSession() + + with patch("server.context.session", session): + _apply_workspace_personalization_preferences(terminal, workspace) + + terminal.set_model.assert_not_called() + self.assertEqual(terminal.model_key, "default-model") + self.assertEqual(session.get("model_key"), "default-model") + self.assertTrue(terminal._workspace_default_model_applied) + + +class TestLoadConversationRestoreModel(unittest.TestCase): + @patch("core.web_terminal.logger") + def test_restore_model_true_restores_saved_model(self, _mock_logger): + """显式加载对话时应恢复对话保存的模型。""" + from core.web_terminal import WebTerminal + + terminal = MagicMock(spec=WebTerminal) + terminal.model_key = "default-model" + terminal.thinking_mode = False + terminal.run_mode = "fast" + terminal.multi_agent_mode = False + terminal.context_manager = MagicMock() + terminal.context_manager.load_conversation_by_id.return_value = True + + cm = MagicMock() + cm.load_conversation.return_value = { + "metadata": { + "thinking_mode": False, + "model_key": "saved-model", + "multi_agent_mode": False, + } + } + terminal.context_manager._get_conversation_manager_for_id.return_value = cm + + terminal.set_model = MagicMock(side_effect=lambda mk: setattr(terminal, "model_key", mk)) + terminal.set_permission_mode = MagicMock() + terminal.set_execution_mode = MagicMock() + terminal.set_network_permission = MagicMock() + terminal.api_client = MagicMock() + terminal.current_session_id = 0 + + # 调用实际方法 + result = WebTerminal.load_conversation(terminal, "conv_test_001", restore_model=True) + + terminal.set_model.assert_called_once_with("saved-model") + self.assertEqual(terminal.model_key, "saved-model") + self.assertTrue(result.get("success")) + + @patch("core.web_terminal.logger") + def test_restore_model_false_keeps_current_model(self, _mock_logger): + """程序启动自动恢复最近对话时不应恢复模型，避免 /new 页面显示旧模型。""" + from core.web_terminal import WebTerminal + + terminal = MagicMock(spec=WebTerminal) + terminal.model_key = "default-model" + terminal.thinking_mode = False + terminal.run_mode = "fast" + terminal.multi_agent_mode = False + terminal.context_manager = MagicMock() + terminal.context_manager.load_conversation_by_id.return_value = True + + cm = MagicMock() + cm.load_conversation.return_value = { + "metadata": { + "thinking_mode": False, + "model_key": "saved-model", + "multi_agent_mode": False, + } + } + terminal.context_manager._get_conversation_manager_for_id.return_value = cm + + terminal.set_model = MagicMock(side_effect=lambda mk: setattr(terminal, "model_key", mk)) + terminal.set_permission_mode = MagicMock() + terminal.set_execution_mode = MagicMock() + terminal.set_network_permission = MagicMock() + terminal.api_client = MagicMock() + terminal.current_session_id = 0 + + result = WebTerminal.load_conversation(terminal, "conv_test_001", restore_model=False) + + terminal.set_model.assert_not_called() + self.assertEqual(terminal.model_key, "default-model") + self.assertTrue(result.get("success")) + + +if __name__ == "__main__": + unittest.main()