release: v8.0.4 — 会话 SQLite 存储 + 去加载动画 + 展示截断 + 工具调用解析加固

- 会话数据改用 SQLite（chat_history/conversations.db，WAL）：sessions / snapshots 两表，
  首次启动自动迁移旧 JSON，/chat 与 /resume 改走 SQLite，memory 搜索兼容数据库
- 去掉流式显示（_typeprint 改整段打印）与常驻加载动画（移除 show_animation 线程及逐帧输出）
- 展示给用户的消息过长自动截断（仅展示层，底层 messages 完整保存）
- 工具调用解析加固：_loads_json 四级兜底修复；新增 _extract_last_action_json 成对错括号扫描；
  续轮也下发 Function Calling 工具，减少长文本格式错误浪费 token
- 版本号统一更新至 8.0.4（README/about.txt/constants/TUI banner/LSP·MCP clientInfo）
- 12 项 session 单测 + 8 项 JSON 提取边界用例全部通过

Co-Authored-By: WorkBuddy <workbuddy@tencent.com>

CHANGELOG.md

@@ -4,6 +4,42 @@
 
 ---
 
+## v8.0.4 (2026-08-08) — 稳定性 & 存储升级
+
+### 🚀 重大改进
+
+- **会话数据改用 SQLite 存储**（替代原先每个会话一个 JSON 文件）
+  - 新增 `xcli_core/session.py` SQLite 后端：库文件 `chat_history/conversations.db`，启用 WAL 模式、`check_same_thread=False`
+  - 两张表：`sessions(id,title,created_at,updated_at,engine,model,messages)` 与 `snapshots(name,saved_at,messages)`
+  - **首次启动自动迁移**：扫描旧的 `chat_history/*.json`，`id`+`messages` 结构写入 `sessions`、纯 `conversation` 结构写入 `snapshots`，迁移后移入 `_migrated_json/` 并写 `.migrated` 标记；迁移包在 try/except 中，绝不阻塞启动
+  - `/chat save|list|open` 与 `/resume` 命令改走 SQLite；`memory.py` 关键词搜索同时检索数据库会话，向后兼容未迁移的 JSON
+  - `SessionManager` 的类名与 `new_session/save/list_sessions/resolve/load` 接口保持不变，旧调用方（`cli_history.py`、`tests/test_session.py`）零改动
+
+### 🐛 Bug 修复 / 体验
+
+- **去掉「一直打印加载动画」的根源**
+  - 移除流式显示：`_typeprint` 改为一次整段打印（去掉逐字符 `time.sleep` 打字机循环）
+  - 移除常驻加载动画：`_generate_response_with_animation` 去掉 `show_animation` 线程、`love_sentences`、`animation_running` 事件与逐帧输出；TTY 仅打印一行静态 `💭 思考中… (按 ESC 取消)`，完成后清除该行
+  - 经核查 OpenAI / Ollama 引擎本就用 `stream: False`，「流式」实为打字机 + 逐帧动画，本次一并去除
+
+### ✨ 新功能
+
+- **展示给用户的消息过长自动截断（仅展示层，存储完整）**
+  - `cli_display.py` 新增 `_truncate_for_display`：超过 `max_chars=4000` 字符 / `max_lines=160` 行时，保留前若干行与字符，末尾追加 `… (展示已折叠，完整内容共 N 字符 / M 行，已完整保存到会话)`
+  - 底层会话数据（`messages`）不做任何截断，完整保存
+
+### 🛠 工程
+
+- **工具调用解析更稳，减少因格式错误浪费 token**
+  - `_loads_json` 升级为 4 级兜底修复：①标准 → ②未配对反斜杠转义 → ③字符串内未转义换行/制表符（`\n`/`\t` 修正）→ ④多余尾逗号删除（`{...",}`），全局统一复用
+  - 新增 `_extract_last_action_json`：扫描所有「成对错括号」对象（字符串感知，追踪 `in_str`/`esc`），返回键集合与 `CONTROL_KEYS={"action","continue","need_continue","think_more","message"}` 相交的最后一个候选；开头即 `[` 纯数组（多次工具调用）直接返回数组；替代原先脆弱的「方法5/6/7」正则抓取
+  - `_build_fc_tools` 去除「续轮剥离 Function Calling 工具」逻辑，续轮也始终下发 FC 工具，模型可用结构化调用而非散文 JSON
+  - 配套独立校验脚本 8 用例全过（含未转义换行、纯数组、尾逗号、正文误触发不应判为工具等边界）
+
+- 版本号统一更新至 v8.0.4（含 README 中英 badge、`about.txt`、`xcli_core/constants.py` 中央 `VERSION`、TUI banner、LSP/MCP `clientInfo.version`）
+
+---
+
 ## v8.0.3 (2026-08-07) — 体验增强
 
 ### ✨ 新功能

README.en.md

@@ -5,7 +5,7 @@
 </p>
 
 <p align="center">
-  <img src="https://img.shields.io/badge/version-8.0.3-blue" alt="version">
+  <img src="https://img.shields.io/badge/version-8.0.4-blue" alt="version">
   <img src="https://img.shields.io/badge/python-3.10+-green" alt="python">
   <img src="https://img.shields.io/badge/license-MIT-orange" alt="license">
   <img src="https://img.shields.io/badge/tests-113%20passed-brightgreen" alt="tests">

README.md

@@ -9,7 +9,7 @@
 </p>
 
 <p align="center">
-  <img src="https://img.shields.io/badge/version-8.0.3-blue" alt="version">
+  <img src="https://img.shields.io/badge/version-8.0.4-blue" alt="version">
   <img src="https://img.shields.io/badge/python-3.10+-green" alt="python">
   <img src="https://img.shields.io/badge/license-MIT-orange" alt="license">
   <img src="https://img.shields.io/badge/tests-150+%20passed-brightgreen" alt="tests">

about.txt

@@ -1,7 +1,7 @@
 **产品标识**  
 - **产品名称**：小狸智能体 增强-终端（Xiaoli Pro-CLI）  
 - **内部版本号**：20260801
-- **发布版本**：发行版8.0.3
+- **发布版本**：发行版8.0.4
 
 **技术规格**  
 - **主要语言**：简体中文（UTF-8）  

lsp_client.py

@@ -268,7 +268,7 @@ class LspClient:
                 "workspace": {"workspaceFolders": True,
                                "configuration": True},
             },
-            "clientInfo": {"name": "xiaoli-cli", "version": "8.0.3"},
+            "clientInfo": {"name": "xiaoli-cli", "version": "8.0.4"},
         })
         if isinstance(result, dict):
             self.capabilities = result.get("capabilities", {}) or {}

mcp_client.py

@@ -153,7 +153,7 @@ class McpClient:
         result = self._send("initialize", {
             "protocolVersion": PROTOCOL_VERSION,
             "capabilities": {},
-            "clientInfo": {"name": "xiaoli-cli", "version": "8.0.3"},
+            "clientInfo": {"name": "xiaoli-cli", "version": "8.0.4"},
         })
         # 通知 server 初始化完成（无响应，无需等待）
         self._send("notifications/initialized", {}, is_notification=True)

tests/test_session.py

@@ -65,15 +65,11 @@ def test_title_generated_from_first_user_msg(tmp_history):
 
 def test_list_sessions_sorted_desc(tmp_history):
     mgr = SessionManager(tmp_history)
-    # 写入两个会话，并强制 s1 的更新时间更早，制造稳定可排序差
+    # 写入两个会话，并用 updated_at 覆盖制造稳定可排序差（SQLite 后端）
-    s1 = mgr.new_session(); mgr.save(s1, [{"role": "user", "content": "旧会话A"}])
+    s1 = mgr.new_session()
-    path1 = os.path.join(tmp_history, f"{s1.id}.json")
+    mgr.save(s1, [{"role": "user", "content": "旧会话A"}], updated_at="2020-01-01 00:00:00")
-    with open(path1, "r", encoding="utf-8") as f:
+    s2 = mgr.new_session()
-        d = json.load(f)
+    mgr.save(s2, [{"role": "user", "content": "新会话B"}], updated_at="2026-08-08 00:00:00")
-    d["updated_at"] = "2020-01-01 00:00:00"
-    with open(path1, "w", encoding="utf-8") as f:
-        json.dump(d, f, ensure_ascii=False)
-    s2 = mgr.new_session(); mgr.save(s2, [{"role": "user", "content": "新会话B"}])
     sessions = mgr.list_sessions()
     assert len(sessions) == 2
     # 倒序：更新的在前（updated_at 时间戳更大）

xcli_core/cli_core.py

@@ -375,16 +375,44 @@ multi 操作支持一次修改多处：
         return "\n".join(prompts)
 
     def _loads_json(self, s):
-        """json.loads，但容忍未成对出现的反斜杠（Windows 路径 C:\\Users 常见）。
+        """json.loads，带多级容错修复（模型输出的 JSON 常常不规整）。
 
-        纯 JSON 模式下模型常把路径写成单反斜杠，直接 json.loads 会报非法转义。
+        尝试顺序：
-        这里先按标准解析；失败时把「未成对的反斜杠」转义后再试一次。
+          1) 标准解析
+          2) 未成对的反斜杠转义（Windows 路径 C:\\Users 常见）
+          3) 字符串内的未转义换行/制表符（多行 args 常见）
+          4) 去掉对象/数组里多余的尾随逗号（模型高频错误）
         """
         try:
             return json.loads(s)
         except json.JSONDecodeError:
+            pass
+        # 2) 反斜杠修复
+        try:
             fixed = re.sub(r'(?<!\\)\\(?!\\\\)', r'\\\\', s)
             return json.loads(fixed)
+        except json.JSONDecodeError:
+            pass
+        # 3) 字符串内未转义换行/制表符修复
+        try:
+            repaired = re.sub(
+                r'"([^"]*(?:\n[^"]*)*)"',
+                lambda m: '"' + (m.group(1)
+                                 .replace('\\', '\\\\')
+                                 .replace('\n', '\\n')
+                                 .replace('\r', '\\r')
+                                 .replace('\t', '\\t')
+                                 .replace('"', '\\"')) + '"',
+                s, flags=re.DOTALL)
+            return json.loads(repaired)
+        except json.JSONDecodeError:
+            pass
+        # 4) 尾随逗号修复
+        try:
+            trimmed = re.sub(r',\s*([}\]])', r'\1', s)
+            return json.loads(trimmed)
+        except json.JSONDecodeError:
+            raise
 
     def _parse_mixed_response(self, response):
         """
@@ -465,69 +493,87 @@ multi 操作支持一次修改多处：
             else:
                 return text_content, json_objects
 
-        # 方法5: 查找内联的JSON对象
+        # 方法5: 稳健抽取最后一个「控制指令」JSON（兼容长文本后格式不规整的情况）。
-        json_start_markers = [
+        # 结构化 FC 已是主路径；这里是纯 JSON 兜底：不再依赖脆弱的 marker 匹配，
-            '{"action": "use_tool"', '{"action": "continue"',
+        # 而是扫描所有平衡括号的 JSON 对象，取最后一个含控制键的，最大限度避免
-            '{"action":"use_tool"', '{"action":"continue"',
+        # 「长篇文本 + 格式错乱的工具调用」被解析失败、白白浪费 token。
-            '{ "action": "use_tool"', '{ "action": "continue"',
+        text_before, json_data = self._extract_last_action_json(response)
-            '{ "action":"use_tool"', '{ "action":"continue"'
+        if json_data is not None:
-        ]
+            return text_before, json_data
-        for marker in json_start_markers:
+
-            json_start = response.find(marker)
+        return response, None
-            if json_start != -1:
+
-                brace_count = 0
+    def _extract_last_action_json(self, response):
-                in_json = False
+        """在可能夹杂长篇正文的响应里，稳健抽取最后一个控制指令 JSON。
-                for i in range(json_start, len(response)):
+
-                    char = response[i]
+        控制指令指含以下任一字段的 JSON 对象：
-                    if char == '{':
+          action / continue / need_continue / think_more / message
-                        if not in_json:
+        返回 (text_before, json_data)；找不到返回 (response, None)。
-                            in_json = True
+        """
-                        brace_count += 1
+        # 整段就是 JSON 数组（多个工具调用）时直接返回，保留并发执行能力
-                    elif char == '}':
+        stripped = response.strip()
-                        brace_count -= 1
+        if stripped.startswith('['):
-                        if in_json and brace_count == 0:
-                            json_str = response[json_start:i+1]
             try:
-                                json_data = self._loads_json(json_str)
+                arr = self._loads_json(stripped)
-                                if isinstance(json_data, dict) and json_data.get('action') in ['use_tool', 'continue']:
+                if isinstance(arr, list) and arr:
-                                    text_content = response[:json_start].strip()
+                    return "", arr
-                                    return text_content, json_data
             except json.JSONDecodeError:
                 pass
-                            break
 
-        # 方法6: 通用JSON对象查找
+        CONTROL_KEYS = {"action", "continue", "need_continue", "think_more", "message"}
-        potential_jsons = re.findall(r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', response, re.DOTALL)
+        candidates = []  # (start_index, parsed_dict)
-        for json_str in potential_jsons:
+        n = len(response)
-            try:
+        i = 0
-                json_data = self._loads_json(json_str)
+        while i < n:
-                if isinstance(json_data, dict) and json_data.get('action') in ['use_tool', 'continue']:
+            if response[i] == '{':
-                    json_start = response.find(json_str)
+                # 从 i 起做括号/字符串感知的平衡解析，取出一个完整 JSON 对象
-                    text_content = response[:json_start].strip()
+                depth = 0
-                    return text_content, json_data
+                in_str = False
-            except json.JSONDecodeError:
+                esc = False
+                end = -1
+                for j in range(i, n):
+                    c = response[j]
+                    if in_str:
+                        if esc:
+                            esc = False
+                        elif c == '\\':
+                            esc = True
+                        elif c == '"':
+                            in_str = False
                         continue
-
+                    if c == '"':
-        # 方法7: 批量JSON
+                        in_str = True
+                    elif c == '{':
+                        depth += 1
+                    elif c == '}':
+                        depth -= 1
+                        if depth == 0:
+                            end = j
+                            break
+                if end != -1:
+                    frag = response[i:end + 1]
                     try:
-            cleaned = response.strip()
+                        obj = self._loads_json(frag)
-            for action in ['use_tool', 'continue']:
+                    except json.JSONDecodeError:
-                last_brace_pos = cleaned.rfind(f'{{"action": "{action}"')
+                        obj = None
-                if last_brace_pos == -1:
+                    if isinstance(obj, dict) and (set(obj.keys()) & CONTROL_KEYS):
-                    last_brace_pos = cleaned.rfind(f'{{"action":"{action}"')
+                        candidates.append((i, obj))
-                if last_brace_pos != -1:
+                    i = end + 1
-                    json_part = cleaned[last_brace_pos:]
+                    continue
-                    if json_part.startswith('['):
+            i += 1
-                        parsed = self._loads_json(json_part)
+
-                        if isinstance(parsed, list):
+        if candidates:
-                            text_content = cleaned[:last_brace_pos].strip()
+            start, obj = candidates[-1]
-                            return text_content, parsed
+            return response[:start].strip(), obj
-                    else:
+
-                        parsed = self._loads_json(json_part)
+        # 退路：整段若为 JSON 数组（如 [{"action":"use_tool"}, ...]）
-                        if isinstance(parsed, dict):
+        try:
-                            text_content = cleaned[:last_brace_pos].strip()
+            stripped = response.strip()
-                            return text_content, parsed
+            if stripped.startswith('['):
-        except:
+                arr = self._loads_json(stripped)
+                if isinstance(arr, list) and arr:
+                    return "", arr
+        except json.JSONDecodeError:
             pass
 
         return response, None
@@ -922,10 +968,9 @@ multi 操作支持一次修改多处：
                 return None
         except Exception:
             return None
-        # 工具结果续轮：模型只需决定「继续/收尾」，不需要 FC schema，省下每轮重发的 token。
+        # 工具结果续轮也下发 FC 工具：让模型在续轮里继续用结构化 function calling
-        # 若模型仍想调工具，走纯 JSON action（无需 schema 即可），行为等价、成本更低。
+        # 调用工具，避免把工具调用写成散文 JSON（长文本后格式错乱会浪费 token）。
-        if current_input and str(current_input).lstrip().startswith("工具执行结果"):
+        # 仅对带 _fc_supported() 的引擎下发 tools（Manual 等不支持的引擎不传 tools）。
-            return None
         try:
             from .fc_tools import mcp_to_openai_tools, prune_tools
             all_tools = mcp_to_openai_tools(self.liugin_manager)
@@ -956,38 +1001,27 @@ multi 操作支持一次修改多处：
             response = self._call_engine_with_tools(current_input, system_prompt, fc_tools)
             return response
 
-        # CLI 模式：带动画和跨平台 ESC 取消
+        # CLI 模式：去掉流式显示与逐帧加载动画，改为「一次性等待 + 完整输出」。
-        animation_running = threading.Event()
+        # 仍用线程跑生成以支持 ESC 取消，但不再喷任何中间动画/情话帧。
-        animation_running.set()
         esc_pressed = threading.Event()
         response_ready = threading.Event()
         response_result = [None]  # 用列表存储，方便线程内修改
 
-        # 是否真实交互终端：非 TTY（管道 / 日志镜像 / 聊天前端捕获 stdout）时，
+        # 是否真实交互终端：非 TTY（管道 / 日志镜像 / 聊天前端捕获 stdout）时
-        # \r 无法原地覆盖，逐帧动画会污染输出流，必须禁用。
+        # 不打印任何等待提示，避免污染捕获流。
         is_tty = sys.stdout.isatty()
-
+        if is_tty:
-        love_sentences = []
+            # 仅一行静态提示，非动画
-        love_file_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), LOVE_FILE_PATH)
+            print(f"{Fore.MAGENTA}💭 思考中… (按 ESC 取消){Style.RESET_ALL}", end="", flush=True)
-        if os.path.exists(love_file_path):
-            try:
-                with open(love_file_path, 'r', encoding='utf-8') as f:
-                    love_sentences = [line.strip() for line in f.readlines()
-                                      if line.strip()
-                                      and not line.strip().lower().startswith(('http://', 'https://'))]
-            except Exception:
-                pass
-        if not love_sentences:
-            love_sentences = ["AI正在思考中...", "请稍等片刻...", "正在处理您的请求..."]
 
         # ── 跨平台 ESC 检测 ──
         def check_for_esc():
             """跨平台 ESC 键检测"""
             import platform
+            while not response_ready.is_set() and not esc_pressed.is_set():
                 if platform.system() == 'Windows':
                     try:
                         import msvcrt
-                    while animation_running.is_set():
                         if msvcrt.kbhit():
                             key = msvcrt.getch()
                             if ord(key) == 27:  # ESC
@@ -995,21 +1029,19 @@ multi 操作支持一次修改多处：
                                 break
                         time.sleep(0.05)
                     except ImportError:
-                    pass
+                        break
                 else:
                     # Linux / macOS — 用 select 做非阻塞读取
                     import select
-                import sys
                     try:
-                    # 保存原始终端设置
+                        import sys as _sys
                         import tty, termios
-                    fd = sys.stdin.fileno()
+                        fd = _sys.stdin.fileno()
                         old_settings = termios.tcgetattr(fd)
                         try:
                             tty.setcbreak(fd)  # cbreak 模式：无需回车即可读取
-                        while animation_running.is_set():
+                            if select.select([_sys.stdin], [], [], 0.05)[0]:
-                            if select.select([sys.stdin], [], [], 0.05)[0]:
+                                ch = _sys.stdin.read(1)
-                                ch = sys.stdin.read(1)
                                 if ord(ch) == 27:  # ESC
                                     esc_pressed.set()
                                     break
@@ -1017,29 +1049,7 @@ multi 操作支持一次修改多处：
                             termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
                     except Exception:
                         # 降级：无法检测 ESC，只能等 AI 完成
-                    pass
+                        break
-
-        # ── 动画线程 ──
-        def show_animation():
-            # 非 TTY：不跑逐帧 \r 动画（避免残帧进入捕获流），但保留一句随机情话保人味
-            if not is_tty:
-                try:
-                    print(f"💭 {random.choice(love_sentences)}", flush=True)
-                except Exception:
-                    pass
-                return
-            last_change_time = time.time()
-            current_sentence = random.choice(love_sentences)
-            while animation_running.is_set():
-                current_time = time.time()
-                if current_time - last_change_time >= 5:
-                    current_sentence = random.choice(love_sentences)
-                    last_change_time = current_time
-                animation_chars = "|/-\\"
-                char_idx = int((current_time * 10) % len(animation_chars))
-                display_sentence = current_sentence[:47] + "..." if len(current_sentence) > 50 else current_sentence
-                print(f"\r{Fore.MAGENTA}{display_sentence} {animation_chars[char_idx]}{Style.RESET_ALL}", end="", flush=True)
-                time.sleep(0.1)
 
         # ── AI 响应线程 ──
         def generate_in_thread():
@@ -1056,9 +1066,9 @@ multi 操作支持一次修改多处：
             finally:
                 response_ready.set()
 
-        # 启动三个线程
+        # 启动线程（生成 + ESC 检测）
         threads = []
-        for target in (show_animation, check_for_esc, generate_in_thread):
+        for target in (check_for_esc, generate_in_thread):
             t = threading.Thread(target=target, daemon=True)
             t.start()
             threads.append(t)
@@ -1070,19 +1080,16 @@ multi 操作支持一次修改多处：
 
             # ESC 被按下 → 取消
             if esc_pressed.is_set():
-                # 清除动画行
                 if is_tty:
                     print("\r" + " " * 80 + "\r", end="", flush=True)
                 print(f"\n{Fore.YELLOW}⚡ AI 请求已取消 (ESC){Style.RESET_ALL}")
                 return "[AI请求已取消]"
 
-            # 响应正常返回
+            # 响应正常返回：清除等待行，不做逐字流式打印。
+            # 完整结果由 process_conversation 统一经 _display_response 透出（含展示层截断）。
             response = response_result[0]
-            if response:
-                # 清除动画行后再输出
             if is_tty:
                 print("\r" + " " * 80 + "\r", end="", flush=True)
-                self._typeprint(response, Fore.CYAN)
 
             if response and self.shared_conversation_history is not None:
                 self.shared_conversation_history.append({
@@ -1093,7 +1100,6 @@ multi 操作支持一次修改多处：
             return response
 
         finally:
-            animation_running.clear()
             for t in threads:
                 t.join(timeout=1)
 

xcli_core/cli_display.py

@@ -39,9 +39,24 @@ class DisplayMixin:
         else:
             print(message)
 
+    def _truncate_for_display(self, content, max_chars=4000, max_lines=160):
+        """仅用于「展示层」的截断：过长时折叠，不影响底层存储的对话数据。"""
+        if not content:
+            return content
+        lines = content.split('\n')
+        if len(lines) > max_lines or len(content) > max_chars:
+            kept = "\n".join(lines[:max_lines])
+            kept = kept[:max_chars]
+            hidden = len(content) - len(kept)
+            note = f"\n\n… (展示已折叠，完整内容共 {len(content)} 字符 / {len(lines)} 行，已完整保存到会话)"
+            return kept + note
+        return content
+
     def _display_response(self, content, is_continue=False):
-        """统一显示 AI 响应，支持 TUI 和 CLI 模式"""
+        """统一显示 AI 响应，支持 TUI 和 CLI 模式（仅展示层截断，不改存储）"""
         user_id_display = f"[用户ID: {self.user_id}]"
+        # 仅展示截断：存储用的 shared_conversation_history 仍是完整内容
+        content = self._truncate_for_display(content)
         response_lines = content.split('\n')
         if self.tui_output_callback:
             for i, line in enumerate(response_lines):
@@ -98,14 +113,14 @@ class DisplayMixin:
                 self._output(f"{color}          {line}{Style.RESET_ALL}")
 
     def _typeprint(self, text, color=None, delay=0.01):
-        """逐字输出文本，实现打字机效果"""
+        """输出文本（已取消逐字打字机效果，改为一次性完整输出）。
-        for char in text:
+
+        保留方法签名以兼容潜在调用方；不再逐字符 sleep，避免「流式显示」。
+        """
         if color:
-                print(f"{color}{char}{Style.RESET_ALL}", end="", flush=True)
+            print(f"{color}{text}{Style.RESET_ALL}")
         else:
-                print(char, end="", flush=True)
+            print(text)
-            time.sleep(delay)
-        print()
 
     # ── 图片显示 ──
 

xcli_core/cli_history.py

@@ -9,17 +9,17 @@ class HistoryMixin:
     """聊天记录管理相关方法（含 opencode 式会话持久化 /resume）"""
 
     def save_chat_history(self, name):
-        """保存当前聊天记录到文件"""
+        """保存当前聊天记录（SQLite 快照）"""
         try:
-            chat_data = {
+            if not hasattr(self, 'session_manager'):
-                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
+                self.init_session_manager()
-                "conversation": self.shared_conversation_history.copy()
+            ok = self.session_manager.save_snapshot(
-            }
+                name, self.shared_conversation_history.copy())
-            file_path = os.path.join(self.chat_history_dir, f"{name}.json")
+            if ok:
-            with open(file_path, 'w', encoding='utf-8') as f:
-                json.dump(chat_data, f, ensure_ascii=False, indent=2, default=str)
                 print(f"{Fore.GREEN}聊天记录已保存为: {name}{Style.RESET_ALL}")
                 return True
+            print(f"{Fore.RED}保存聊天记录失败{Style.RESET_ALL}")
+            return False
         except Exception as e:
             print(f"{Fore.RED}保存聊天记录失败: {e}{Style.RESET_ALL}")
             return False
@@ -27,30 +27,28 @@ class HistoryMixin:
     def list_chat_history(self):
         """列出所有已保存的聊天记录"""
         try:
-            chat_files = [f for f in os.listdir(self.chat_history_dir) if f.endswith('.json')]
+            if not hasattr(self, 'session_manager'):
-            if not chat_files:
+                self.init_session_manager()
+            snaps = self.session_manager.list_snapshots()
+            if not snaps:
                 print(f"{Fore.YELLOW}没有已保存的聊天记录{Style.RESET_ALL}")
                 return
             print(f"{Fore.GREEN}已保存的聊天记录:{Style.RESET_ALL}")
-            for i, file in enumerate(chat_files, 1):
+            for i, (chat_name, saved_at) in enumerate(snaps, 1):
-                file_path = os.path.join(self.chat_history_dir, file)
+                print(f"{Fore.GREEN}{i}. {chat_name} (保存时间: {saved_at}){Style.RESET_ALL}")
-                mod_time = os.path.getmtime(file_path)
-                mod_time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(mod_time))
-                chat_name = file[:-5]
-                print(f"{Fore.GREEN}{i}. {chat_name} (保存时间: {mod_time_str}){Style.RESET_ALL}")
         except Exception as e:
             print(f"{Fore.RED}列出聊天记录失败: {e}{Style.RESET_ALL}")
 
     def load_chat_history(self, name):
-        """加载指定的聊天记录"""
+        """加载指定的聊天记录（SQLite 快照）"""
         try:
-            file_path = os.path.join(self.chat_history_dir, f"{name}.json")
+            if not hasattr(self, 'session_manager'):
-            if not os.path.exists(file_path):
+                self.init_session_manager()
+            msgs = self.session_manager.load_snapshot(name)
+            if msgs is None:
                 print(f"{Fore.RED}聊天记录 '{name}' 不存在{Style.RESET_ALL}")
                 return False
-            with open(file_path, 'r', encoding='utf-8') as f:
+            self.shared_conversation_history = list(msgs)
-                chat_data = json.load(f)
-            self.shared_conversation_history = chat_data.get("conversation", [])
             self._set_shared_conversation_history()
             print(f"{Fore.GREEN}已加载聊天记录: {name}{Style.RESET_ALL}")
             if self.shared_conversation_history:

xcli_core/constants.py

@@ -7,7 +7,7 @@ LOVE_FILE_PATH = "love.txt"
 DEFAULT_MAX_HISTORY = 999999  # 无限制对话历史
 
 # 版本号（v8.0 基于 v5.4.1 重新立项）
-VERSION = "8.0.3"
+VERSION = "8.0.4"
 VERSION_NAME = "小狸 Pro-CLI"
 
 import os

xcli_core/memory.py

@@ -248,7 +248,7 @@ class MemoryManager:
                 md_file, keyword_lower, name, limit - len(results)
             ))
 
-        # 搜索聊天记录
+        # 搜索聊天记录（遗留 JSON 文件，兼容迁移前的历史）
         for json_file in sorted(glob.glob(os.path.join(self.chat_dir, "*.json")), reverse=True):
             if len(results) >= limit:
                 break
@@ -257,6 +257,42 @@ class MemoryManager:
                 json_file, keyword_lower, name, limit - len(results)
             ))
 
+        # 搜索聊天记录（SQLite conversations.db，迁移后的主存储）
+        db_path = os.path.join(self.chat_dir, "conversations.db")
+        if os.path.exists(db_path):
+            try:
+                import sqlite3 as _sqlite3
+                conn = _sqlite3.connect(db_path)
+                conn.row_factory = _sqlite3.Row
+                rows = conn.execute(
+                    "SELECT id, messages FROM sessions "
+                    "UNION ALL SELECT name AS id, messages FROM snapshots"
+                ).fetchall()
+                conn.close()
+                for row in rows:
+                    if len(results) >= limit:
+                        break
+                    try:
+                        msgs = json.loads(row["messages"] or "[]")
+                    except (json.JSONDecodeError, ValueError, TypeError):
+                        msgs = []
+                    text = "\n".join(
+                        str(m.get("content", "")) for m in msgs
+                        if isinstance(m, dict)
+                    )
+                    source = f"chat/{row['id']}"
+                    for i, line in enumerate(text.split("\n"), 1):
+                        if keyword_lower in line.lower():
+                            results.append({
+                                "source": source,
+                                "line": i,
+                                "snippet": line.strip()[:120],
+                            })
+                            if len(results) >= limit:
+                                break
+            except Exception:
+                pass
+
         return results[:limit]
 
     def _search_file(self, path: str, keyword: str, source: str, limit: int) -> List[Dict]:

xcli_core/rich_tui.py

@@ -28,7 +28,7 @@ class RichTUI:
         self.cli = cli
         # record=True 是关键：渲染过程被记录，可随时导出文本/SVG/HTML 自验
         self.console = Console(record=True, width=100)
-        self.version = "8.0.3"
+        self.version = "8.0.4"
         self.engine = ""
         self.model = ""
         self.mode = "普通"

xcli_core/session.py

@@ -1,30 +1,22 @@
-"""会话（Session）持久化与恢复。
+"""会话（Session）持久化与恢复 —— SQLite 存储后端。
 
 把 openai/ollama/manual 等引擎的对话历史（self.shared_conversation_history）
-按「会话」维度持久化到 chat_history/<id>.json，支持：
+按「会话」维度持久化到单一 SQLite 数据库 chat_history/conversations.db，
-  - 自动落盘（每轮对话后调用 SessionManager.save）
+取代原先每个会话一个 JSON 文件的方式。两张逻辑表：
-  - 列出历史会话（/resume、/sessions）
+  - sessions  : 自动持久化的会话，供 /resume 使用
-  - 一键恢复（把 messages 重新载入 shared_conversation_history）
+  - snapshots : 用户命名的 /chat save 快照
 
-存储格式（新）：
+兼容旧版 chat_history/<id>.json：首次打开历史目录时，自动把遗留的 *.json
-  {
+（含旧 /chat save 的 conversation 格式）迁入库，并移到 _migrated_json/ 备份，
-    "id": "20260805-171045-a1b2c3",
+避免重复计数。
-    "title": "用户首条消息前若干个字",
-    "created_at": "2026-08-05 17:10:45",
-    "updated_at": "2026-08-05 17:15:22",
-    "engine": "openai",
-    "model": "xophunyuan7bmt",
-    "messages": [ ... ]
-  }
-
-兼容旧 /chat save 写入的文件（无 id 字段、用 "conversation" 键）：
-  读取时回退——id 取文件名，messages 取 conversation 键。
 """
 import os
 import re
 import time
 import json
 import uuid
+import sqlite3
+import shutil
 
 
 def _make_session_id() -> str:
@@ -50,7 +42,8 @@ def _title_from_messages(messages, limit: int = 40) -> str:
 class Session:
     """单个会话的轻量容器。"""
 
-    def __init__(self, sid, title, created_at, updated_at, engine, model, messages):
+    def __init__(self, sid, title, created_at, updated_at, engine, model, messages,
+                 is_snapshot: bool = False):
         self.id = sid
         self.title = title
         self.created_at = created_at
@@ -58,6 +51,7 @@ class Session:
         self.engine = engine
         self.model = model
         self.messages = messages or []
+        self.is_snapshot = is_snapshot  # /chat save 快照在 /resume 里归并
 
     @property
     def msg_count(self) -> int:
@@ -76,7 +70,6 @@ class Session:
 
     @classmethod
     def from_dict(cls, data: dict, fallback_id: str = ""):
-        # 兼容旧 /chat save 格式：无 id、用 "conversation" 键
         sid = data.get("id") or fallback_id
         messages = data.get("messages")
         if messages is None and "conversation" in data:
@@ -93,7 +86,7 @@ class Session:
 
 
 class SessionManager:
-    """会话的读写与索引管理。"""
+    """基于 SQLite 的会话读写与索引管理。"""
 
     def __init__(self, history_dir: str):
         self.history_dir = history_dir
@@ -102,6 +95,93 @@ class SessionManager:
                 os.makedirs(history_dir)
             except OSError:
                 pass
+        self.db_path = os.path.join(self.history_dir, "conversations.db") if history_dir else ":memory:"
+        self._conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        self._conn.row_factory = sqlite3.Row
+        try:
+            self._conn.execute("PRAGMA journal_mode=WAL")
+        except sqlite3.Error:
+            pass
+        self._init_schema()
+        self._migrate_legacy()
+
+    # ── 表结构 ──
+    def _init_schema(self):
+        c = self._conn.cursor()
+        c.execute(
+            """CREATE TABLE IF NOT EXISTS sessions(
+                id TEXT PRIMARY KEY,
+                title TEXT,
+                created_at TEXT,
+                updated_at TEXT,
+                engine TEXT,
+                model TEXT,
+                messages TEXT
+            )"""
+        )
+        c.execute(
+            """CREATE TABLE IF NOT EXISTS snapshots(
+                name TEXT PRIMARY KEY,
+                saved_at TEXT,
+                messages TEXT
+            )"""
+        )
+        self._conn.commit()
+
+    # ── 遗留 JSON 迁移（仅一次）──
+    def _migrate_legacy(self):
+        if not self.history_dir:
+            return
+        marker = os.path.join(self.history_dir, ".migrated")
+        if os.path.exists(marker):
+            return
+        try:
+            legacy = []
+            for fn in os.listdir(self.history_dir):
+                if not fn.endswith(".json"):
+                    continue
+                if fn == "conversations.db":
+                    continue
+                path = os.path.join(self.history_dir, fn)
+                try:
+                    with open(path, "r", encoding="utf-8") as fh:
+                        data = json.load(fh)
+                except (OSError, json.JSONDecodeError, ValueError):
+                    continue
+                if not isinstance(data, dict):
+                    continue
+                legacy.append((fn, data))
+
+            if legacy:
+                backup = os.path.join(self.history_dir, "_migrated_json")
+                os.makedirs(backup, exist_ok=True)
+                for fn, data in legacy:
+                    src = os.path.join(self.history_dir, fn)
+                    sid = data.get("id")
+                    if sid and "messages" in data:
+                        s = Session.from_dict(data, fallback_id=sid)
+                        self._upsert_session_row(s)
+                    elif "conversation" in data:
+                        name = fn[:-5]
+                        self.save_snapshot(
+                            name, data["conversation"], saved_at=data.get("timestamp")
+                        )
+                    else:
+                        continue
+                    # 迁完后移走，避免被 list_sessions 的遗留扫描重复计数
+                    try:
+                        shutil.move(src, os.path.join(backup, fn))
+                    except OSError:
+                        pass
+                self._conn.commit()
+            try:
+                with open(marker, "w", encoding="utf-8") as f:
+                    f.write("migrated")
+            except OSError:
+                pass
+        except Exception:
+            # 迁移失败绝不阻断启动
+            pass
 
     # ── 新建 ──
     def new_session(self, engine: str = "", model: str = ""):
@@ -116,9 +196,21 @@ class SessionManager:
             messages=[],
         )
 
+    # ── 落库辅助 ──
+    def _upsert_session_row(self, s: Session):
+        self._conn.execute(
+            """INSERT OR REPLACE INTO sessions
+               (id, title, created_at, updated_at, engine, model, messages)
+               VALUES (?,?,?,?,?,?,?)""",
+            (s.id, s.title, s.created_at, s.updated_at, s.engine, s.model,
+             json.dumps(s.messages, ensure_ascii=False, default=str)),
+        )
+        self._conn.commit()
+
     # ── 保存 ──
-    def save(self, session: Session, messages, engine: str = "", model: str = ""):
+    def save(self, session: Session, messages, engine: str = "", model: str = "",
-        """把当前对话内容写回到 session 对应的文件。"""
+             updated_at: str = None) -> bool:
+        """把当前对话内容写回到会话表。updated_at 可覆盖（测试/迁移用）。"""
         if session is None:
             return False
         session.messages = list(messages or [])
@@ -126,39 +218,48 @@ class SessionManager:
             session.engine = engine
         if not session.model and model:
             session.model = model
-        # 首条用户消息生成标题（仅当仍是占位标题时）
         if session.title in ("(新会话)", "(未命名会话)", ""):
             session.title = _title_from_messages(session.messages)
-        session.updated_at = time.strftime("%Y-%m-%d %H:%M:%S")
+        now = updated_at or time.strftime("%Y-%m-%d %H:%M:%S")
+        session.updated_at = now
         if not session.created_at:
-            session.created_at = session.updated_at
+            session.created_at = now
-        file_path = os.path.join(self.history_dir, f"{session.id}.json")
         try:
-            with open(file_path, "w", encoding="utf-8") as f:
+            self._upsert_session_row(session)
-                json.dump(session.to_dict(), f, ensure_ascii=False, indent=2, default=str)
             return True
-        except OSError:
+        except sqlite3.Error:
             return False
 
     # ── 列出 ──
     def list_sessions(self) -> list:
-        """返回按更新时间倒序的 Session 列表。"""
+        """返回按更新时间倒序的 Session 列表（sessions ∪ snapshots ∪ 遗留 JSON）。"""
         sessions = []
         try:
-            files = [f for f in os.listdir(self.history_dir) if f.endswith(".json")]
+            rows = self._conn.execute(
-        except OSError:
+                "SELECT id, title, created_at, updated_at, engine, model, messages "
-            return sessions
+                "FROM sessions ORDER BY updated_at DESC"
-        for f in files:
+            ).fetchall()
-            path = os.path.join(self.history_dir, f)
+            for r in rows:
-            try:
+                msgs = self._safe_messages(r["messages"])
-                with open(path, "r", encoding="utf-8") as fh:
+                sessions.append(Session(
-                    data = json.load(fh)
+                    sid=r["id"], title=r["title"], created_at=r["created_at"],
-                sid = data.get("id") or f[:-5]
+                    updated_at=r["updated_at"], engine=r["engine"], model=r["model"],
-                s = Session.from_dict(data, fallback_id=sid)
+                    messages=msgs,
-                # 跳过明显空会话（无消息且非手动命名）
+                ))
-                sessions.append(s)
+        except sqlite3.Error:
-            except (OSError, json.JSONDecodeError, ValueError):
+            pass
+
+        # 合并 /chat save 快照（向后兼容：旧版里它们也出现在 /resume 列表）
+        for snap in self._iter_snapshots():
+            sessions.append(snap)
+
+        # 合并目录里残留的遗留 JSON（迁移遗漏或非本工具写入）
+        for legacy in self._iter_legacy_json():
+            # 去重：已入库的同名不重复
+            if any(s.id == legacy.id for s in sessions):
                 continue
+            sessions.append(legacy)
+
         sessions.sort(key=lambda s: s.updated_at, reverse=True)
         return sessions
 
@@ -168,16 +269,13 @@ class SessionManager:
         if sessions is None:
             sessions = self.list_sessions()
         token = (token or "").strip()
-        # 序号
         if token.isdigit():
             idx = int(token) - 1
             if 0 <= idx < len(sessions):
                 return sessions[idx]
-        # id（精确或前缀）
         for s in sessions:
             if s.id == token or s.id.startswith(token):
                 return s
-        # 文件名（去掉 .json）
         if token.endswith(".json"):
             token = token[:-5]
         for s in sessions:
@@ -187,15 +285,91 @@ class SessionManager:
 
     def load(self, token: str) -> Session:
         """读取指定会话的完整内容（含 messages）。"""
-        sessions = self.list_sessions()
+        return self.resolve(token)
-        s = self.resolve(token, sessions)
+
-        if s is None:
+    # ── 快照（/chat save）──
+    def save_snapshot(self, name: str, messages, saved_at: str = None) -> bool:
+        try:
+            self._conn.execute(
+                "INSERT OR REPLACE INTO snapshots (name, saved_at, messages) VALUES (?,?,?)",
+                (name, saved_at or time.strftime("%Y-%m-%d %H:%M:%S"),
+                 json.dumps(list(messages or []), ensure_ascii=False, default=str)),
+            )
+            self._conn.commit()
+            return True
+        except sqlite3.Error:
+            return False
+
+    def list_snapshots(self) -> list:
+        """返回 [(name, saved_at), ...] 倒序。"""
+        try:
+            rows = self._conn.execute(
+                "SELECT name, saved_at FROM snapshots ORDER BY saved_at DESC"
+            ).fetchall()
+            return [(r["name"], r["saved_at"]) for r in rows]
+        except sqlite3.Error:
+            return []
+
+    def load_snapshot(self, name: str):
+        """返回快照的消息列表，或 None。"""
+        try:
+            row = self._conn.execute(
+                "SELECT messages FROM snapshots WHERE name=?", (name,)
+            ).fetchone()
+            if row is None:
                 return None
-        # 重新从文件读取以确保 messages 完整
+            return self._safe_messages(row["messages"])
-        path = os.path.join(self.history_dir, f"{s.id}.json")
+        except sqlite3.Error:
+            return None
+
+    # ── 内部辅助 ──
+    @staticmethod
+    def _safe_messages(text):
+        try:
+            data = json.loads(text or "[]")
+            return data if isinstance(data, list) else []
+        except (json.JSONDecodeError, ValueError, TypeError):
+            return []
+
+    def _iter_snapshots(self):
+        try:
+            rows = self._conn.execute(
+                "SELECT name, saved_at, messages FROM snapshots"
+            ).fetchall()
+        except sqlite3.Error:
+            return
+        for r in rows:
+            msgs = self._safe_messages(r["messages"])
+            yield Session(
+                sid=r["name"], title=_title_from_messages(msgs),
+                created_at=r["saved_at"], updated_at=r["saved_at"],
+                engine="", model="", messages=msgs, is_snapshot=True,
+            )
+
+    def _iter_legacy_json(self):
+        if not self.history_dir:
+            return
+        skip = {"conversations.db", ".migrated"}
+        for fn in os.listdir(self.history_dir):
+            if not fn.endswith(".json"):
+                continue
+            if fn in skip:
+                continue
+            path = os.path.join(self.history_dir, fn)
             try:
                 with open(path, "r", encoding="utf-8") as fh:
                     data = json.load(fh)
-            return Session.from_dict(data, fallback_id=s.id)
             except (OSError, json.JSONDecodeError, ValueError):
-            return s
+                continue
+            if not isinstance(data, dict):
+                continue
+            sid = data.get("id")
+            if sid and "messages" in data:
+                yield Session.from_dict(data, fallback_id=sid)
+            elif "conversation" in data:
+                msgs = data["conversation"]
+                yield Session(
+                    sid=fn[:-5], title=_title_from_messages(msgs),
+                    created_at=data.get("timestamp", ""), updated_at=data.get("timestamp", ""),
+                    engine="", model="", messages=msgs, is_snapshot=True,
+                )

xcli_core/tui.py

@@ -810,7 +810,7 @@ if TEXTUAL_AVAILABLE:
                 "",
                 "  ╔══════════════════════════════════════════════╗",
                 "  ║                                              ║",
-                "  ║    小狸 Pro-CLI v8.0.3                        ║",
+                "  ║    小狸 Pro-CLI v8.0.4                        ║",
                 "  ║    智能编程助手 · 动画增强版                  ║",
                 "  ║                                              ║",
                 "  ╚══════════════════════════════════════════════╝",
@@ -854,7 +854,7 @@ if TEXTUAL_AVAILABLE:
                 ("", "msg-dim"),
                 ("  ╔══════════════════════════════════════════════╗", "msg-welcome"),
                 ("  ║                                              ║", "msg-welcome"),
-                ("  ║    小狸 Pro-CLI v8.0.3                        ║", "msg-welcome"),
+                ("  ║    小狸 Pro-CLI v8.0.4                        ║", "msg-welcome"),
                 ("  ║    智能编程助手 · 动画增强版                  ║", "msg-welcome"),
                 ("  ║                                              ║", "msg-welcome"),
                 ("  ╚══════════════════════════════════════════════╝", "msg-welcome"),