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,63 @@
 
 ---
 
+## 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) — 体验增强
+
+### ✨ 新功能
+
+- **TUI 新增 Vim 风格键位**（默认开，可用 `/vim` 开关）
+  - 输入框**模态编辑**：`Esc` 在 NORMAL / INSERT 间切换；NORMAL 下
+    - 移动：`h j k l`、`w / b`（跳词）、`0 / $`（行首/行尾）、`^`（首个非空白）、`gg / G`（文首/文尾）
+    - 进入插入：`i / a / o`（插入/追加/下方新行）、`I / A / O`（行首/行尾/上方新行）
+    - 编辑：`dd`（删行）、`x`（删字）、`dw`（删词）、`d$`（删到行尾）、`u`（撤销）
+    - 普通模式下 `回车` = 下移一行（与 vim 一致）
+  - 核心状态机抽成纯模块 `xcli_core/vim_keys.py`（`VimInputState`，不依赖 textual），配 `tests/test_vim_keys.py` 单测，18 项动作全部验证通过
+  - 聊天区导航（输入框未聚焦时生效）：`j / k` 逐行滚动、`Ctrl+F / B` 翻页、`Ctrl+D / U` 半页、`Ctrl+Space` 切换焦点、`i` 聚焦并进入插入
+  - 输入框下方提示栏实时显示 `-- INSERT --` / `-- NORMAL --` 模式；`/help` 已补充 Vim 键位说明
+  - 可通过 `config.json` 的 `vim_mode: false` 默认关闭
+
+### 🛠 工程
+
+- 版本号统一更新至 v8.0.3（含 README 中英 badge、`about.txt`、`xcli_core/constants.py` 中央 `VERSION`、TUI banner、LSP/MCP `clientInfo.version`）
+
+---
+
 ## v8.0.2 (2026-08-07) — 小修复版本
 
 ### 🐛 Bug 修复

README.en.md

@@ -5,7 +5,7 @@
 </p>
 
 <p align="center">
-  <img src="https://img.shields.io/badge/version-8.0.2-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.2-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.2
+- **发布版本**：发行版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.2"},
+            "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.2"},
+            "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 的更新时间更早，制造稳定可排序差
-    s1 = mgr.new_session(); mgr.save(s1, [{"role": "user", "content": "旧会话A"}])
-    path1 = os.path.join(tmp_history, f"{s1.id}.json")
-    with open(path1, "r", encoding="utf-8") as f:
-        d = json.load(f)
-    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"}])
+    # 写入两个会话，并用 updated_at 覆盖制造稳定可排序差（SQLite 后端）
+    s1 = mgr.new_session()
+    mgr.save(s1, [{"role": "user", "content": "旧会话A"}], updated_at="2020-01-01 00:00:00")
+    s2 = mgr.new_session()
+    mgr.save(s2, [{"role": "user", "content": "新会话B"}], updated_at="2026-08-08 00:00:00")
     sessions = mgr.list_sessions()
     assert len(sessions) == 2
     # 倒序：更新的在前（updated_at 时间戳更大）

tests/test_vim_keys.py

@@ -0,0 +1,164 @@
+"""VimInputState 纯逻辑单测（不依赖 textual）。
+
+用一个极简 FakeBuffer 持有 (text, offset)，把 EditResult 应用回去，
+逐动作断言文本/光标/模式变化。
+"""
+import sys
+import os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from xcli_core.vim_keys import VimInputState, EditResult
+
+
+class FakeBuffer:
+    """模拟输入框：持有 text 与 cursor offset。"""
+    def __init__(self, text, offset, mode="INSERT"):
+        self.st = VimInputState(mode=mode)
+        self.text = text
+        self.offset = offset
+
+    def press(self, token):
+        res = self.st.feed(self.text, self.offset, token)
+        self.text = res.text
+        self.offset = res.offset
+        return res
+
+    def press_seq(self, *tokens):
+        return [self.press(t) for t in tokens]
+
+
+def test_escape_enters_normal_and_steps_left():
+    b = FakeBuffer("hello", 5, mode="INSERT")
+    r = b.press("escape")
+    assert b.st.mode == "NORMAL"
+    assert b.offset == 4
+    assert r.message == "-- NORMAL --"
+
+
+def test_normal_h_l_clamp():
+    b = FakeBuffer("hello", 0, mode="NORMAL")
+    b.press("h")                       # 行首不动
+    assert b.offset == 0
+    b.press_seq("l", "l", "l", "l", "l")  # 走到末尾
+    assert b.offset == 5
+    b.press("l")                       # 末尾不动
+    assert b.offset == 5
+
+
+def test_j_k_move_lines():
+    b = FakeBuffer("aa\nbb\ncc", 1, mode="NORMAL")  # 第0行 col1
+    b.press("j")                       # 第1行 col1
+    assert b.offset == 4              # "aa\n"=3 +1
+    b.press("k")                       # 回第0行 col1
+    assert b.offset == 1
+
+
+def test_0_and_dollar():
+    b = FakeBuffer("hello world", 5, mode="NORMAL")
+    b.press("0")
+    assert b.offset == 0
+    b.press("$")
+    assert b.offset == 11
+
+
+def test_i_enters_insert():
+    b = FakeBuffer("abc", 1, mode="NORMAL")
+    r = b.press("i")
+    assert b.st.mode == "INSERT"
+    assert r.action is None
+
+
+def test_a_insert_after():
+    b = FakeBuffer("abc", 1, mode="NORMAL")
+    r = b.press("a")
+    assert b.st.mode == "INSERT"
+    assert b.offset == 2
+
+
+def test_o_opens_new_line():
+    b = FakeBuffer("line1\nline2", 2, mode="NORMAL")  # 第0行
+    r = b.press("o")
+    assert b.st.mode == "INSERT"
+    assert b.text == "line1\n\nline2"
+    assert b.offset == 6               # "line1\n"=6
+
+
+def test_O_opens_above():
+    b = FakeBuffer("line1\nline2", 7, mode="NORMAL")  # 第1行 'l' of line2
+    r = b.press("O")
+    assert b.st.mode == "INSERT"
+    assert b.text == "line1\n\nline2"
+    assert b.offset == 6               # 新空行起始
+
+
+def test_dd_deletes_line():
+    b = FakeBuffer("aaa\nbbb\nccc", 4, mode="NORMAL")  # 第1行
+    b.press_seq("d", "d")
+    assert b.text == "aaa\nccc"
+    assert b.offset == 4                # 回到第1行行首(aaa\n 之后)
+
+
+def test_dw_deletes_word():
+    b = FakeBuffer("foo bar baz", 0, mode="NORMAL")
+    b.press_seq("d", "w")
+    assert b.text == "bar baz"
+    assert b.offset == 0
+
+
+def test_d_dollar_deletes_to_eol():
+    b = FakeBuffer("abc def", 2, mode="NORMAL")  # 第0行 col2 ('c')
+    b.press_seq("d", "$")
+    assert b.text == "ab"
+    assert b.offset == 2
+
+
+def test_x_deletes_char():
+    b = FakeBuffer("abc", 1, mode="NORMAL")  # 'b'
+    b.press("x")
+    assert b.text == "ac"
+    assert b.offset == 1
+
+
+def test_w_b_word_motion():
+    b = FakeBuffer("foo bar baz", 0, mode="NORMAL")
+    b.press("w")
+    assert b.offset == 4               # 下一词 "bar" 词首
+    b.press("b")
+    assert b.offset == 0
+
+
+def test_gg_and_G():
+    b = FakeBuffer("a\nb\nc", 4, mode="NORMAL")
+    b.press_seq("g", "g")
+    assert b.offset == 0
+    b.press("G")
+    assert b.offset == 4               # 末行同列
+
+
+def test_u_sets_undo_action():
+    b = FakeBuffer("hello", 2, mode="NORMAL")
+    r = b.press("u")
+    assert r.action == "undo"
+
+
+def test_pending_cleared_on_invalid_combo():
+    b = FakeBuffer("abc", 1, mode="NORMAL")
+    b.press("d")                        # 进入待定
+    assert b.st.pending == "d"
+    b.press("x")                        # 非法组合 -> 清待定, 当普通键
+    assert b.st.pending is None
+    assert b.st.mode == "NORMAL"
+
+
+def test_unmapped_key_ignored_in_normal():
+    b = FakeBuffer("abc", 1, mode="NORMAL")
+    r = b.press("z")                    # 未映射
+    assert r.handled is True
+    assert b.text == "abc"
+    assert b.offset == 1
+
+
+def test_insert_typing_delegated():
+    b = FakeBuffer("abc", 1, mode="INSERT")
+    r = b.press("h")                    # 非 Esc：交给输入框默认处理
+    assert r.handled is False

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对象
-        json_start_markers = [
-            '{"action": "use_tool"', '{"action": "continue"',
-            '{"action":"use_tool"', '{"action":"continue"',
-            '{ "action": "use_tool"', '{ "action": "continue"',
-            '{ "action":"use_tool"', '{ "action":"continue"'
-        ]
-        for marker in json_start_markers:
-            json_start = response.find(marker)
-            if json_start != -1:
-                brace_count = 0
-                in_json = False
-                for i in range(json_start, len(response)):
-                    char = response[i]
-                    if char == '{':
-                        if not in_json:
-                            in_json = True
-                        brace_count += 1
-                    elif char == '}':
-                        brace_count -= 1
-                        if in_json and brace_count == 0:
-                            json_str = response[json_start:i+1]
-                            try:
-                                json_data = self._loads_json(json_str)
-                                if isinstance(json_data, dict) and json_data.get('action') in ['use_tool', 'continue']:
-                                    text_content = response[:json_start].strip()
-                                    return text_content, json_data
-                            except json.JSONDecodeError:
-                                pass
-                            break
+        # 方法5: 稳健抽取最后一个「控制指令」JSON（兼容长文本后格式不规整的情况）。
+        # 结构化 FC 已是主路径；这里是纯 JSON 兜底：不再依赖脆弱的 marker 匹配，
+        # 而是扫描所有平衡括号的 JSON 对象，取最后一个含控制键的，最大限度避免
+        # 「长篇文本 + 格式错乱的工具调用」被解析失败、白白浪费 token。
+        text_before, json_data = self._extract_last_action_json(response)
+        if json_data is not None:
+            return text_before, json_data
 
-        # 方法6: 通用JSON对象查找
-        potential_jsons = re.findall(r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', response, re.DOTALL)
-        for json_str in potential_jsons:
+        return response, None
+
+    def _extract_last_action_json(self, response):
+        """在可能夹杂长篇正文的响应里，稳健抽取最后一个控制指令 JSON。
+
+        控制指令指含以下任一字段的 JSON 对象：
+          action / continue / need_continue / think_more / message
+        返回 (text_before, json_data)；找不到返回 (response, None)。
+        """
+        # 整段就是 JSON 数组（多个工具调用）时直接返回，保留并发执行能力
+        stripped = response.strip()
+        if stripped.startswith('['):
             try:
-                json_data = self._loads_json(json_str)
-                if isinstance(json_data, dict) and json_data.get('action') in ['use_tool', 'continue']:
-                    json_start = response.find(json_str)
-                    text_content = response[:json_start].strip()
-                    return text_content, json_data
+                arr = self._loads_json(stripped)
+                if isinstance(arr, list) and arr:
+                    return "", arr
             except json.JSONDecodeError:
-                continue
+                pass
 
-        # 方法7: 批量JSON
+        CONTROL_KEYS = {"action", "continue", "need_continue", "think_more", "message"}
+        candidates = []  # (start_index, parsed_dict)
+        n = len(response)
+        i = 0
+        while i < n:
+            if response[i] == '{':
+                # 从 i 起做括号/字符串感知的平衡解析，取出一个完整 JSON 对象
+                depth = 0
+                in_str = False
+                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 == '"':
+                        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:
+                        obj = self._loads_json(frag)
+                    except json.JSONDecodeError:
+                        obj = None
+                    if isinstance(obj, dict) and (set(obj.keys()) & CONTROL_KEYS):
+                        candidates.append((i, obj))
+                    i = end + 1
+                    continue
+            i += 1
+
+        if candidates:
+            start, obj = candidates[-1]
+            return response[:start].strip(), obj
+
+        # 退路：整段若为 JSON 数组（如 [{"action":"use_tool"}, ...]）
         try:
-            cleaned = response.strip()
-            for action in ['use_tool', 'continue']:
-                last_brace_pos = cleaned.rfind(f'{{"action": "{action}"')
-                if last_brace_pos == -1:
-                    last_brace_pos = cleaned.rfind(f'{{"action":"{action}"')
-                if last_brace_pos != -1:
-                    json_part = cleaned[last_brace_pos:]
-                    if json_part.startswith('['):
-                        parsed = self._loads_json(json_part)
-                        if isinstance(parsed, list):
-                            text_content = cleaned[:last_brace_pos].strip()
-                            return text_content, parsed
-                    else:
-                        parsed = self._loads_json(json_part)
-                        if isinstance(parsed, dict):
-                            text_content = cleaned[:last_brace_pos].strip()
-                            return text_content, parsed
-        except:
+            stripped = response.strip()
+            if stripped.startswith('['):
+                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。
-        # 若模型仍想调工具，走纯 JSON action（无需 schema 即可），行为等价、成本更低。
-        if current_input and str(current_input).lstrip().startswith("工具执行结果"):
-            return None
+        # 工具结果续轮也下发 FC 工具：让模型在续轮里继续用结构化 function calling
+        # 调用工具，避免把工具调用写成散文 JSON（长文本后格式错乱会浪费 token）。
+        # 仅对带 _fc_supported() 的引擎下发 tools（Manual 等不支持的引擎不传 tools）。
         try:
             from .fc_tools import mcp_to_openai_tools, prune_tools
             all_tools = mcp_to_openai_tools(self.liugin_manager)
@@ -956,90 +1001,55 @@ multi 操作支持一次修改多处：
             response = self._call_engine_with_tools(current_input, system_prompt, fc_tools)
             return response
 
-        # CLI 模式：带动画和跨平台 ESC 取消
-        animation_running = threading.Event()
-        animation_running.set()
+        # CLI 模式：去掉流式显示与逐帧加载动画，改为「一次性等待 + 完整输出」。
+        # 仍用线程跑生成以支持 ESC 取消，但不再喷任何中间动画/情话帧。
         esc_pressed = threading.Event()
         response_ready = threading.Event()
         response_result = [None]  # 用列表存储，方便线程内修改
 
-        # 是否真实交互终端：非 TTY（管道 / 日志镜像 / 聊天前端捕获 stdout）时，
-        # \r 无法原地覆盖，逐帧动画会污染输出流，必须禁用。
+        # 是否真实交互终端：非 TTY（管道 / 日志镜像 / 聊天前端捕获 stdout）时
+        # 不打印任何等待提示，避免污染捕获流。
         is_tty = sys.stdout.isatty()
-
-        love_sentences = []
-        love_file_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), LOVE_FILE_PATH)
-        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正在思考中...", "请稍等片刻...", "正在处理您的请求..."]
+        if is_tty:
+            # 仅一行静态提示，非动画
+            print(f"{Fore.MAGENTA}💭 思考中… (按 ESC 取消){Style.RESET_ALL}", end="", flush=True)
 
         # ── 跨平台 ESC 检测 ──
         def check_for_esc():
             """跨平台 ESC 键检测"""
             import platform
-            if platform.system() == 'Windows':
-                try:
-                    import msvcrt
-                    while animation_running.is_set():
+            while not response_ready.is_set() and not esc_pressed.is_set():
+                if platform.system() == 'Windows':
+                    try:
+                        import msvcrt
                         if msvcrt.kbhit():
                             key = msvcrt.getch()
                             if ord(key) == 27:  # ESC
                                 esc_pressed.set()
                                 break
                         time.sleep(0.05)
-                except ImportError:
-                    pass
-            else:
-                # Linux / macOS — 用 select 做非阻塞读取
-                import select
-                import sys
-                try:
-                    # 保存原始终端设置
-                    import tty, termios
-                    fd = sys.stdin.fileno()
-                    old_settings = termios.tcgetattr(fd)
+                    except ImportError:
+                        break
+                else:
+                    # Linux / macOS — 用 select 做非阻塞读取
+                    import select
                     try:
-                        tty.setcbreak(fd)  # cbreak 模式：无需回车即可读取
-                        while animation_running.is_set():
-                            if select.select([sys.stdin], [], [], 0.05)[0]:
-                                ch = sys.stdin.read(1)
+                        import sys as _sys
+                        import tty, termios
+                        fd = _sys.stdin.fileno()
+                        old_settings = termios.tcgetattr(fd)
+                        try:
+                            tty.setcbreak(fd)  # cbreak 模式：无需回车即可读取
+                            if select.select([_sys.stdin], [], [], 0.05)[0]:
+                                ch = _sys.stdin.read(1)
                                 if ord(ch) == 27:  # ESC
                                     esc_pressed.set()
                                     break
-                    finally:
-                        termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
-                except Exception:
-                    # 降级：无法检测 ESC，只能等 AI 完成
-                    pass
-
-        # ── 动画线程 ──
-        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)
+                        finally:
+                            termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
+                    except Exception:
+                        # 降级：无法检测 ESC，只能等 AI 完成
+                        break
 
         # ── 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 is_tty:
+                print("\r" + " " * 80 + "\r", end="", flush=True)
 
             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:
-            if color:
-                print(f"{color}{char}{Style.RESET_ALL}", end="", flush=True)
-            else:
-                print(char, end="", flush=True)
-            time.sleep(delay)
-        print()
+        """输出文本（已取消逐字打字机效果，改为一次性完整输出）。
+
+        保留方法签名以兼容潜在调用方；不再逐字符 sleep，避免「流式显示」。
+        """
+        if color:
+            print(f"{color}{text}{Style.RESET_ALL}")
+        else:
+            print(text)
 
     # ── 图片显示 ──
 

xcli_core/cli_history.py

@@ -9,17 +9,17 @@ class HistoryMixin:
     """聊天记录管理相关方法（含 opencode 式会话持久化 /resume）"""
 
     def save_chat_history(self, name):
-        """保存当前聊天记录到文件"""
+        """保存当前聊天记录（SQLite 快照）"""
         try:
-            chat_data = {
-                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
-                "conversation": self.shared_conversation_history.copy()
-            }
-            file_path = os.path.join(self.chat_history_dir, f"{name}.json")
-            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
+            if not hasattr(self, 'session_manager'):
+                self.init_session_manager()
+            ok = self.session_manager.save_snapshot(
+                name, self.shared_conversation_history.copy())
+            if ok:
+                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 chat_files:
+            if not hasattr(self, 'session_manager'):
+                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):
-                file_path = os.path.join(self.chat_history_dir, file)
-                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}")
+            for i, (chat_name, saved_at) in enumerate(snaps, 1):
+                print(f"{Fore.GREEN}{i}. {chat_name} (保存时间: {saved_at}){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 os.path.exists(file_path):
+            if not hasattr(self, 'session_manager'):
+                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:
-                chat_data = json.load(f)
-            self.shared_conversation_history = chat_data.get("conversation", [])
+            self.shared_conversation_history = list(msgs)
             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.2"
+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.2"
+        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，支持：
-  - 自动落盘（每轮对话后调用 SessionManager.save）
-  - 列出历史会话（/resume、/sessions）
-  - 一键恢复（把 messages 重新载入 shared_conversation_history）
+按「会话」维度持久化到单一 SQLite 数据库 chat_history/conversations.db，
+取代原先每个会话一个 JSON 文件的方式。两张逻辑表：
+  - sessions  : 自动持久化的会话，供 /resume 使用
+  - snapshots : 用户命名的 /chat save 快照
 
-存储格式（新）：
-  {
-    "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 键。
+兼容旧版 chat_history/<id>.json：首次打开历史目录时，自动把遗留的 *.json
+（含旧 /chat save 的 conversation 格式）迁入库，并移到 _migrated_json/ 备份，
+避免重复计数。
 """
 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 = ""):
-        """把当前对话内容写回到 session 对应的文件。"""
+    def save(self, session: Session, messages, engine: str = "", model: str = "",
+             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
-        file_path = os.path.join(self.history_dir, f"{session.id}.json")
+            session.created_at = now
         try:
-            with open(file_path, "w", encoding="utf-8") as f:
-                json.dump(session.to_dict(), f, ensure_ascii=False, indent=2, default=str)
+            self._upsert_session_row(session)
             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")]
-        except OSError:
-            return sessions
-        for f in files:
-            path = os.path.join(self.history_dir, f)
-            try:
-                with open(path, "r", encoding="utf-8") as fh:
-                    data = json.load(fh)
-                sid = data.get("id") or f[:-5]
-                s = Session.from_dict(data, fallback_id=sid)
-                # 跳过明显空会话（无消息且非手动命名）
-                sessions.append(s)
-            except (OSError, json.JSONDecodeError, ValueError):
+            rows = self._conn.execute(
+                "SELECT id, title, created_at, updated_at, engine, model, messages "
+                "FROM sessions ORDER BY updated_at DESC"
+            ).fetchall()
+            for r in rows:
+                msgs = self._safe_messages(r["messages"])
+                sessions.append(Session(
+                    sid=r["id"], title=r["title"], created_at=r["created_at"],
+                    updated_at=r["updated_at"], engine=r["engine"], model=r["model"],
+                    messages=msgs,
+                ))
+        except sqlite3.Error:
+            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()
-        s = self.resolve(token, sessions)
-        if s is None:
-            return None
-        # 重新从文件读取以确保 messages 完整
-        path = os.path.join(self.history_dir, f"{s.id}.json")
+        return self.resolve(token)
+
+    # ── 快照（/chat save）──
+    def save_snapshot(self, name: str, messages, saved_at: str = None) -> bool:
         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
+            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
+            return self._safe_messages(row["messages"])
+        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)
+            except (OSError, json.JSONDecodeError, ValueError):
+                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

@@ -11,6 +11,7 @@ if TEXTUAL_AVAILABLE:
     from textual.reactive import var
     from textual.binding import Binding
     from textual import on
+    from .vim_keys import VimInputState
     import os
     from rich.syntax import Syntax
     from rich.panel import Panel
@@ -560,6 +561,125 @@ if TEXTUAL_AVAILABLE:
                 return
             # 其他键正常处理
 
+
+    # ═══════════════════════════════════════════════════
+    #  Vim 风格模态输入框（NORMAL / INSERT）
+    # ═══════════════════════════════════════════════════
+
+    class VimSendTextArea(SendTextArea):
+        """
+        带 Vim 模态编辑的输入框。
+
+        模式：
+          INSERT  —— 默认，正常打字；Esc 进入 NORMAL
+          NORMAL  —— h/j/k/l 移动，i/a/o 进入插入，dd 删行，x 删字，
+                    w/b 跳词，0/$ 行首行尾，gg/G 文首文尾，u 撤销 …
+
+        坐标：VimInputState 用字符偏移 offset，这里负责与 TextArea 的
+        (row, col) 光标互相转换。所有 vim 逻辑都在 vim_keys.py（纯函数，
+        可单测）；本类只做「按键映射 + 把结果写回 TextArea」。
+        """
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.vim_enabled = True
+            self.vim = VimInputState(mode="INSERT")
+
+        # ── offset <-> (row, col) ──
+        def _cur_off(self) -> int:
+            r, c = self.cursor_location
+            text = self.text
+            lines = text.split("\n")
+            return sum(len(l) + 1 for l in lines[:r]) + c
+
+        def _set_off(self, off: int):
+            text = self.text
+            off = max(0, min(off, len(text)))
+            r = text.count("\n", 0, off)
+            line_start = text.rfind("\n", 0, off)
+            c = off - (line_start + 1)
+            self.cursor_location = (r, c)
+
+        # ── key 名字 -> vim token ──
+        @staticmethod
+        def _vim_token(key: str):
+            m = {
+                "escape": "escape", "h": "h", "l": "l", "j": "j", "k": "k",
+                "w": "w", "b": "b", "e": "e", "0": "0", "$": "$", "^": "^",
+                "g": "g", "G": "G", "d": "d", "x": "x", "u": "u",
+                "i": "i", "a": "a", "o": "o", "O": "O", "I": "I", "A": "A",
+                "left": "left", "right": "right", "up": "up", "down": "down",
+                "shift+i": "I", "shift+a": "A", "shift+o": "O",
+                "shift+g": "G", "shift+6": "^", "shift+4": "$",
+            }
+            return m.get(key)
+
+        def _notify_mode(self):
+            app = self.app
+            if hasattr(app, "_update_vim_mode_display"):
+                try:
+                    app._update_vim_mode_display()
+                except Exception:
+                    pass
+
+        def _apply(self, res):
+            try:
+                if res.text != self.text:
+                    self.text = res.text
+                self._set_off(res.offset)
+                self.vim.mode = res.mode
+                self.vim.pending = res.pending
+                if res.action == "undo":
+                    try:
+                        self.undo()
+                    except Exception:
+                        pass
+                self._notify_mode()
+            except Exception:
+                # 任何意外都不应让输入框崩；退回默认行为
+                pass
+
+        async def _on_key(self, event):
+            # 回车：INSERT（或 vim 关闭）发送；NORMAL 下移一行
+            if event.key in ("enter", "\r", "\n"):
+                if self.vim_enabled and self.vim.mode == "NORMAL":
+                    event.prevent_default()
+                    res = self.vim.feed(self.text, self._cur_off(), "j")
+                    self._apply(res)
+                else:
+                    event.prevent_default()
+                    try:
+                        self.app.action_send_message()
+                    except Exception:
+                        pass
+                return
+
+            if not self.vim_enabled:
+                return  # 退化为普通 TextArea
+
+            if self.vim.mode == "INSERT":
+                if event.key == "escape":
+                    event.prevent_default()
+                    res = self.vim.feed(self.text, self._cur_off(), "escape")
+                    self._apply(res)
+                # 其余键交给 TextArea 默认处理（打字）
+                return
+
+            # ── NORMAL 模式 ──
+            if event.key == "shift+enter":
+                token = "o"
+            else:
+                token = self._vim_token(event.key)
+            if token is None:
+                # ctrl+/shift+ 组合键放回去，让 App 级绑定处理（聚焦切换等）
+                if event.key.startswith("ctrl+") or event.key.startswith("shift+"):
+                    return
+                # 普通单键未映射：吞掉，避免 TextArea 在 NORMAL 下误编辑
+                event.prevent_default()
+                return
+            event.prevent_default()
+            res = self.vim.feed(self.text, self._cur_off(), token)
+            self._apply(res)
+
     # ═══════════════════════════════════════════════════
     #  TUI 主应用
     # ═══════════════════════════════════════════════════
@@ -583,6 +703,15 @@ if TEXTUAL_AVAILABLE:
             Binding("3", "tab_sessions", "会话", show=False),
             Binding("4", "tab_status", "状态", show=False),
             Binding("escape", "cancel", "取消", show=False),
+            # ── Vim 风格导航（输入框未聚焦时生效）──
+            Binding("j", "chat_down", "↓聊天", show=False),
+            Binding("k", "chat_up", "↑聊天", show=False),
+            Binding("i", "focus_insert", "插入", show=False),
+            Binding("ctrl+f", "chat_page_down", "翻页↓", show=True),
+            Binding("ctrl+b", "chat_page_up", "翻页↑", show=True),
+            Binding("ctrl+d", "chat_half_down", "半页↓", show=True),
+            Binding("ctrl+u", "chat_half_up", "半页↑", show=True),
+            Binding("ctrl+space", "toggle_focus", "聚焦切换", show=True),
         ]
 
         sidebar_visible = var(True)
@@ -605,8 +734,8 @@ if TEXTUAL_AVAILABLE:
                     yield VerticalScroll(id="chat-scroll")
                     with Vertical(id="input-area"):
                         with Container(id="input-wrapper"):
-                            yield SendTextArea(
-                                placeholder="输入消息... (回车发送, Shift+回车换行)",
+                            yield VimSendTextArea(
+                                placeholder="输入消息... (回车发送, Shift+回车换行, Esc 进 Vim 普通模式)",
                                 id="user-input",
                                 soft_wrap=True,
                                 tab_behavior="indent",
@@ -638,6 +767,8 @@ if TEXTUAL_AVAILABLE:
             self._switch_sidebar_tab("engine")
             # 精简欢迎(一行)
             self._render_welcome_compact()
+            # 初始化 Vim 键位开关（默认开，可被 config.vim_mode 关闭）
+            self._init_vim_mode()
             self.query_one("#user-input").focus()
             def _show_hint():
                 try:
@@ -646,6 +777,18 @@ if TEXTUAL_AVAILABLE:
                     pass
             self.set_timer(1.5, _show_hint)
 
+        def _init_vim_mode(self):
+            try:
+                ta = self.query_one("#user-input")
+            except Exception:
+                return
+            cfg = getattr(self.cli, "config", None)
+            enabled = True
+            if isinstance(cfg, dict) and "vim_mode" in cfg:
+                enabled = bool(cfg["vim_mode"])
+            ta.vim_enabled = enabled
+            self._update_vim_mode_display()
+
         # ── 欢迎界面 ──
 
         def _render_welcome_compact(self):
@@ -667,7 +810,7 @@ if TEXTUAL_AVAILABLE:
                 "",
                 "  ╔══════════════════════════════════════════════╗",
                 "  ║                                              ║",
-                "  ║    小狸 Pro-CLI v8.0.2                        ║",
+                "  ║    小狸 Pro-CLI v8.0.4                        ║",
                 "  ║    智能编程助手 · 动画增强版                  ║",
                 "  ║                                              ║",
                 "  ╚══════════════════════════════════════════════╝",
@@ -711,7 +854,7 @@ if TEXTUAL_AVAILABLE:
                 ("", "msg-dim"),
                 ("  ╔══════════════════════════════════════════════╗", "msg-welcome"),
                 ("  ║                                              ║", "msg-welcome"),
-                ("  ║    小狸 Pro-CLI v8.0.2                        ║", "msg-welcome"),
+                ("  ║    小狸 Pro-CLI v8.0.4                        ║", "msg-welcome"),
                 ("  ║    智能编程助手 · 动画增强版                  ║", "msg-welcome"),
                 ("  ║                                              ║", "msg-welcome"),
                 ("  ╚══════════════════════════════════════════════╝", "msg-welcome"),
@@ -1087,6 +1230,7 @@ if TEXTUAL_AVAILABLE:
                 'resume': lambda: self._tui_resume(args),
                 'snapshot': lambda: self._save_screenshot(),
                 'screenshot': lambda: self._save_screenshot(),
+                'vim': lambda: self._toggle_vim(),
             }
 
             handler = cmds.get(name)
@@ -1114,12 +1258,21 @@ if TEXTUAL_AVAILABLE:
   /tools         列出工具
   /status        系统状态
   /clear         清屏
+  /vim           开关 Vim 键位
 
   ⌨  快捷键:
   Ctrl+C     退出        Ctrl+L     清屏
   Ctrl+N     新对话      回车       发送消息
   Shift+回车 换行        F1         侧栏
-  Escape     取消生成    Tab        缩进"""
+  Escape     取消生成    Tab        缩进
+  Ctrl+Space 焦点切换    i          聚焦并输入
+  Ctrl+F/B   翻页下/上   Ctrl+D/U   半页下/上
+
+  VIM 输入(默认开): 输入框内 Esc 进普通模式
+    h j k l  移动    w/b  跳词    0/$  行首/行尾
+    i a o    插入/追加/换行      I A O  行首/行尾/上方
+    dd 删行   x 删字   u 撤销     gg/G 文首/文尾
+    普通模式下 回车 = 下移一行"""
             self._system(help_text)
 
         def _handle_manual(self, args):
@@ -1341,6 +1494,91 @@ if TEXTUAL_AVAILABLE:
                 self._system("已取消")
                 self.is_generating = False
 
+        # ── Vim 风格：聊天区滚动 / 聚焦切换 ──
+        def _user_input(self):
+            return self.query_one("#user-input")
+
+        def _chat_scroll(self):
+            return self.query_one("#chat-scroll")
+
+        def action_toggle_focus(self):
+            """Ctrl+Space：在输入框与聊天区之间切换焦点。"""
+            ta = self._user_input()
+            if ta.has_focus:
+                self.screen.set_focus(None)
+            else:
+                ta.focus()
+                if getattr(ta, "vim_enabled", False):
+                    ta.vim.mode = "INSERT"
+                self._update_vim_mode_display()
+
+        def action_focus_insert(self):
+            """i：输入框未聚焦时聚焦并进入 INSERT（模仿 vim 按 i 开始输入）。"""
+            ta = self._user_input()
+            if ta.has_focus:
+                return
+            ta.focus()
+            if getattr(ta, "vim_enabled", False):
+                ta.vim.mode = "INSERT"
+            self._update_vim_mode_display()
+
+        def _guard_chat(self):
+            """聊天导航仅在输入框未聚焦时生效，避免与输入冲突。"""
+            return self._user_input().has_focus
+
+        def action_chat_down(self):
+            if self._guard_chat():
+                return
+            self._chat_scroll().scroll_down(3)
+
+        def action_chat_up(self):
+            if self._guard_chat():
+                return
+            self._chat_scroll().scroll_up(3)
+
+        def action_chat_page_down(self):
+            if self._guard_chat():
+                return
+            self._chat_scroll().scroll_page_down()
+
+        def action_chat_page_up(self):
+            if self._guard_chat():
+                return
+            self._chat_scroll().scroll_page_up()
+
+        def action_chat_half_down(self):
+            if self._guard_chat():
+                return
+            self._chat_scroll().scroll_relative(y=20)
+
+        def action_chat_half_up(self):
+            if self._guard_chat():
+                return
+            self._chat_scroll().scroll_relative(y=-20)
+
+        def _update_vim_mode_display(self):
+            """在输入框下方提示栏显示当前 Vim 模式。"""
+            try:
+                hint = self.query_one("#input-hint")
+                ta = self._user_input()
+            except Exception:
+                return
+            if not getattr(ta, "vim_enabled", False):
+                hint.update("  回车发送 | Shift+回车换行 | Ctrl+L 清屏 | F1 侧栏 | Ctrl+P PLAN")
+                return
+            mode = getattr(getattr(ta, "vim", None), "mode", "INSERT")
+            tag = "-- INSERT --" if mode == "INSERT" else "-- NORMAL --"
+            hint.update(f"  {tag}  │  Esc 切模式  │  i插入 a追加 o换行  hjkl移动  dd删行 x删字  u撤销")
+
+        def _toggle_vim(self):
+            ta = self._user_input()
+            ta.vim_enabled = not getattr(ta, "vim_enabled", True)
+            if ta.vim_enabled:
+                self._system("Vim 键位: 开 (Esc 在 NORMAL/INSERT 间切换)")
+            else:
+                self._system("Vim 键位: 关 (普通输入)")
+            self._update_vim_mode_display()
+
         def _save_screenshot(self):
             """导出当前 TUI 屏幕截图（SVG）——AI 可据此自验实时渲染（/snapshot）"""
             import os

xcli_core/vim_keys.py

@@ -0,0 +1,275 @@
+"""
+Vim 风格输入编辑状态机（纯逻辑，不依赖 textual）
+================================================
+把“模态编辑”的核心（NORMAL / INSERT 双模式 + 各种 motion / operator）抽成
+纯函数 + 一个小状态类，便于在不依赖 textual 的环境里做单元测试。
+
+widget 侧（tui.py 的 VimSendTextArea）只负责：
+- 把 textual 的 key 名字映射成这里的 token（见 tui.py 的 _vim_token）
+- 调用 VimInputState.feed(text, offset, token) 拿到 EditResult
+- 把 EditResult 应用回 TextArea（设 text / 设光标 / 触发 undo 等）
+
+坐标约定：文本用单一字符串表示，光标用「字符偏移 offset」表示（从 0 开始）。
+widget 负责 offset <-> (row, col) 的转换。
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class EditResult:
+    """一次 vim 动作的执行结果。"""
+    text: str                       # 应用后的文本
+    offset: int                     # 应用后的光标偏移
+    mode: str                       # 结束时的模式: "INSERT" / "NORMAL"
+    pending: Optional[str]          # 多键序列的待定前缀: "d" / "g" / None
+    handled: bool = True            # False 表示“交给输入框默认处理”（仅 INSERT 非 Esc 键）
+    action: Optional[str] = None    # 额外指令: "undo"
+    message: str = ""               # 状态栏提示语
+
+
+# ── 文本/坐标辅助 ──────────────────────────────────────────────────────────
+
+def _line_start_off(text: str, off: int) -> int:
+    """当前行行首偏移。"""
+    s = text.rfind("\n", 0, off)
+    return s + 1
+
+
+def _line_end_off(text: str, off: int) -> int:
+    """当前行行尾（不含换行符）偏移。"""
+    e = text.find("\n", off)
+    return e if e != -1 else len(text)
+
+
+def _first_nonblank_off(text: str, off: int) -> int:
+    """当前行第一个非空白字符偏移，若整行空白则回到行首。"""
+    s = _line_start_off(text, off)
+    e = _line_end_off(text, off)
+    i = s
+    while i < e and text[i].isspace():
+        i += 1
+    return i
+
+
+def _word_forward_off(text: str, off: int) -> int:
+    """跳到下一个 word 的词首（vim 的 w）：越过当前词 + 词间空白。"""
+    n = len(text)
+    i = off
+    # 越过错当前空白
+    while i < n and text[i].isspace():
+        i += 1
+    # 越过当前 word
+    if i < n and (text[i].isalnum() or text[i] == "_"):
+        while i < n and (text[i].isalnum() or text[i] == "_"):
+            i += 1
+    else:
+        while i < n and not text[i].isspace() and not (text[i].isalnum() or text[i] == "_"):
+            i += 1
+    # 越过词间空白，落到下一词首
+    while i < n and text[i].isspace():
+        i += 1
+    return i
+
+
+def _word_back_off(text: str, off: int) -> int:
+    """跳到上一个 word 的词首（vim 的 b）。"""
+    i = off
+    # 越过光标前空白
+    while i > 0 and text[i - 1].isspace():
+        i -= 1
+    # 越过当前 word
+    if i > 0 and (text[i - 1].isalnum() or text[i - 1] == "_"):
+        while i > 0 and (text[i - 1].isalnum() or text[i - 1] == "_"):
+            i -= 1
+    else:
+        while i > 0 and not text[i - 1].isspace() and not (text[i - 1].isalnum() or text[i - 1] == "_"):
+            i -= 1
+    # 越过词间空白，落到上一词首
+    while i > 0 and text[i - 1].isspace():
+        i -= 1
+    return i
+
+
+def _offset_to_rc(text: str, off: int):
+    off = max(0, min(off, len(text)))
+    row = text.count("\n", 0, off)
+    line_start = text.rfind("\n", 0, off)
+    col = off - (line_start + 1)
+    return row, col
+
+
+def _rc_to_offset(text: str, row: int, col: int) -> int:
+    lines = text.split("\n")
+    row = max(0, min(row, len(lines) - 1))
+    line = lines[row]
+    col = max(0, min(col, len(line)))
+    off = 0
+    for l in lines[:row]:
+        off += len(l) + 1
+    return off + col
+
+
+# ── 状态机 ────────────────────────────────────────────────────────────────
+
+class VimInputState:
+    """
+    Vim 输入状态机。
+
+    mode:   "INSERT" 或 "NORMAL"
+    pending: 多键序列待定前缀（"d" / "g"），单键动作时为 None
+
+    用法：
+        st = VimInputState(mode="INSERT")   # 输入框默认进入 INSERT
+        res = st.feed(text, offset, token)
+        # 把 res 应用回输入框
+    """
+
+    def __init__(self, mode: str = "INSERT"):
+        self.mode = mode
+        self.pending: Optional[str] = None
+
+    # 进入 INSERT 时返回的结果构造器
+    @staticmethod
+    def _insert(text, offset, msg) -> EditResult:
+        return EditResult(text, offset, "INSERT", None, message=msg)
+
+    def feed(self, text: str, offset: int, token: str) -> EditResult:
+        """
+        token 是已经映射好的 vim 键名，例如:
+        "h","l","j","k","w","b","e","0","$","^","g","G","d","x","u",
+        "i","a","o","O","I","A","escape","left","right","up","down"
+        """
+        if self.mode == "INSERT":
+            if token == "escape":
+                # 回到 NORMAL，光标左移一格（vim 习惯：Esc 后光标退到本字上）
+                new_off = offset
+                if offset > 0 and text[offset - 1] != "\n":
+                    new_off = offset - 1
+                self.mode = "NORMAL"
+                self.pending = None
+                return EditResult(text, new_off, "NORMAL", None, message="-- NORMAL --")
+            # 其余键交给 TextArea 默认处理（打字）
+            return EditResult(text, offset, "INSERT", None, handled=False)
+
+        # ── NORMAL 模式 ──
+        # 多键序列待定前缀
+        if self.pending == "d":
+            self.pending = None
+            if token == "d":
+                return self._delete_line(text, offset, "dd")
+            if token == "w":
+                return self._delete_word(text, offset, "dw")
+            if token == "$":
+                return self._delete_to_eol(text, offset, "d$")
+            # 非法组合：清掉 pending 后把当前键当单键重新处理
+            # （落到下面的单键逻辑）
+
+        if self.pending == "g":
+            self.pending = None
+            if token == "g":
+                return EditResult(text, 0, "NORMAL", None, message="gg")
+            # 其余：当作普通单键处理
+
+        # ── 单键 ──
+        if token in ("i",):
+            self.mode = "INSERT"
+            return self._insert(text, offset, "-- INSERT --")
+        if token == "I":
+            self.mode = "INSERT"
+            return self._insert(text, _first_nonblank_off(text, offset), "-- INSERT (行首) --")
+        if token == "a":
+            self.mode = "INSERT"
+            return self._insert(text, min(offset + 1, len(text)), "-- INSERT (追加) --")
+        if token == "A":
+            self.mode = "INSERT"
+            return self._insert(text, _line_end_off(text, offset), "-- INSERT (行尾) --")
+        if token == "o":
+            self.mode = "INSERT"
+            e = _line_end_off(text, offset)
+            new_text = text[:e] + "\n" + text[e:]
+            return self._insert(new_text, e + 1, "-- INSERT (新行) --")
+        if token == "O":
+            self.mode = "INSERT"
+            s = _line_start_off(text, offset)
+            new_text = text[:s] + "\n" + text[s:]
+            return self._insert(new_text, s, "-- INSERT (上方新行) --")
+
+        if token in ("h", "left"):
+            new_off = max(_line_start_off(text, offset), offset - 1)
+            return EditResult(text, new_off, "NORMAL", None)
+        if token in ("l", "right"):
+            le = _line_end_off(text, offset)
+            new_off = min(le, offset + 1) if offset < len(text) else offset
+            return EditResult(text, new_off, "NORMAL", None)
+        if token in ("j", "down"):
+            r, c = _offset_to_rc(text, offset)
+            return EditResult(text, _rc_to_offset(text, r + 1, c), "NORMAL", None)
+        if token in ("k", "up"):
+            r, c = _offset_to_rc(text, offset)
+            return EditResult(text, _rc_to_offset(text, r - 1, c), "NORMAL", None)
+        if token == "w":
+            return EditResult(text, _word_forward_off(text, offset), "NORMAL", None)
+        if token == "b":
+            return EditResult(text, _word_back_off(text, offset), "NORMAL", None)
+        if token == "e":
+            i = _word_forward_off(text, offset)
+            new_off = max(offset, i - 1) if i > offset else offset
+            return EditResult(text, new_off, "NORMAL", None)
+        if token == "0":
+            return EditResult(text, _line_start_off(text, offset), "NORMAL", None)
+        if token == "^":
+            return EditResult(text, _first_nonblank_off(text, offset), "NORMAL", None)
+        if token == "$":
+            return EditResult(text, _line_end_off(text, offset), "NORMAL", None)
+        if token == "G":
+            r, c = _offset_to_rc(text, offset)
+            last_row = text.count("\n")
+            return EditResult(text, _rc_to_offset(text, last_row, c), "NORMAL", None)
+        if token == "g":
+            self.pending = "g"
+            return EditResult(text, offset, "NORMAL", "g", message="g")
+        if token == "d":
+            self.pending = "d"
+            return EditResult(text, offset, "NORMAL", "d", message="d")
+        if token == "x":
+            if offset < len(text):
+                new_text = text[:offset] + text[offset + 1:]
+                return EditResult(new_text, offset, "NORMAL", None, message="x")
+            return EditResult(text, offset, "NORMAL", None)
+        if token == "u":
+            return EditResult(text, offset, "NORMAL", None, action="undo", message="u")
+        if token == "escape":
+            return EditResult(text, offset, "NORMAL", None, message="-- NORMAL --")
+
+        # 未知键：忽略（保持 NORMAL）
+        return EditResult(text, offset, "NORMAL", None)
+
+    # ── operator 实现 ──
+    def _delete_line(self, text, offset, msg) -> EditResult:
+        s = _line_start_off(text, offset)
+        e = _line_end_off(text, offset)
+        if e < len(text):
+            e2 = e + 1          # 连同行尾换行一起删
+        else:
+            e2 = e
+            if s > 0:
+                s = s - 1       # 最后一行：删掉上一行的换行
+        new_text = text[:s] + text[e2:]
+        new_off = min(s, len(new_text))
+        return EditResult(new_text, new_off, "NORMAL", None, message=msg)
+
+    def _delete_word(self, text, offset, msg) -> EditResult:
+        s = offset
+        e = _word_forward_off(text, offset)
+        new_text = text[:s] + text[e:]
+        new_off = min(s, len(new_text))
+        return EditResult(new_text, new_off, "NORMAL", None, message=msg)
+
+    def _delete_to_eol(self, text, offset, msg) -> EditResult:
+        s = offset
+        e = _line_end_off(text, offset)
+        new_text = text[:s] + text[e:]
+        new_off = min(s, len(new_text))
+        return EditResult(new_text, new_off, "NORMAL", None, message=msg)