diff --git a/tests/test_session.py b/tests/test_session.py new file mode 100644 index 0000000..f572518 --- /dev/null +++ b/tests/test_session.py @@ -0,0 +1,187 @@ +"""会话持久化与 /resume 的回归测试。 + +全程使用临时目录,不触碰真实 chat_history/ 与 config.json。 +""" +import os +import json +import tempfile +import shutil +import uuid + +import pytest + +from xcli_core.session import Session, SessionManager, _make_session_id, _title_from_messages + + +@pytest.fixture +def tmp_history(): + d = tempfile.mkdtemp(prefix="xiaoli_sess_") + yield d + shutil.rmtree(d, ignore_errors=True) + + +def _write_old_format(history_dir, name, messages): + """模拟旧 /chat save 写入(无 id 字段、用 conversation 键)。""" + path = os.path.join(history_dir, f"{name}.json") + with open(path, "w", encoding="utf-8") as f: + json.dump({"timestamp": "2026-08-01 10:00:00", "conversation": messages}, f, ensure_ascii=False) + return path + + +# ── SessionManager 单元 ── + +def test_session_id_format(): + sid = _make_session_id() + assert len(sid.split("-")) == 3 # YYYYMMDD-HHMMSS-xxxxxx + + +def test_new_session_basic_fields(tmp_history): + mgr = SessionManager(tmp_history) + s = mgr.new_session(engine="openai", model="xophunyuan7bmt") + assert s.id and s.engine == "openai" and s.model == "xophunyuan7bmt" + assert s.messages == [] and s.title == "(新会话)" + + +def test_save_then_load_roundtrip(tmp_history): + mgr = SessionManager(tmp_history) + s = mgr.new_session(engine="openai", model="gpt-4o") + msgs = [{"role": "user", "content": "你好"}, {"role": "assistant", "content": "你也好"}] + assert mgr.save(s, msgs, engine="openai", model="gpt-4o") is True + loaded = mgr.load(s.id) + assert loaded is not None + assert loaded.msg_count == 2 + assert loaded.messages[0]["content"] == "你好" + assert loaded.engine == "openai" and loaded.model == "gpt-4o" + + +def test_title_generated_from_first_user_msg(tmp_history): + mgr = SessionManager(tmp_history) + s = mgr.new_session() + msgs = [{"role": "user", "content": "帮我写一个快排函数"}] + mgr.save(s, msgs) + loaded = mgr.load(s.id) + assert "快排" in loaded.title + + +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"}]) + sessions = mgr.list_sessions() + assert len(sessions) == 2 + # 倒序:更新的在前(updated_at 时间戳更大) + assert sessions[0].id == s2.id + + +def test_resolve_by_index_and_prefix(tmp_history): + mgr = SessionManager(tmp_history) + s1 = mgr.new_session(); mgr.save(s1, [{"role": "user", "content": "x"}]) + s2 = mgr.new_session(); mgr.save(s2, [{"role": "user", "content": "y"}]) + sessions = mgr.list_sessions() + by_idx = mgr.resolve("1", sessions) + by_id = mgr.resolve(s2.id, sessions) + by_prefix = mgr.resolve(s2.id[:20], sessions) + assert by_idx is not None and by_idx.id == sessions[0].id + assert by_id is not None and by_id.id == s2.id + assert by_prefix is not None and by_prefix.id == s2.id + + +def test_compat_old_chat_format(tmp_history): + """兼容旧 /chat save 写入的文件(无 id、conversation 键)。""" + _write_old_format(tmp_history, "我的旧记录", + [{"role": "user", "content": "旧消息"}]) + mgr = SessionManager(tmp_history) + sessions = mgr.list_sessions() + assert len(sessions) == 1 + s = sessions[0] + assert s.id == "我的旧记录" # 回退用文件名 + assert s.msg_count == 1 + assert s.messages[0]["content"] == "旧消息" + + +# ── 与 CLI 集成(用 stub 复用 HistoryMixin 的会话方法) ── + +from xcli_core.cli_history import HistoryMixin as HistoryMixinStubBase + + +class _Stub(HistoryMixinStubBase): + def __init__(self, history_dir): + self.chat_history_dir = history_dir + self.shared_conversation_history = [] + self.current_session = None + self.current_engine = None + self.current_model = None + self.session_manager = SessionManager(history_dir) + + def _set_shared_conversation_history(self): + # 真实 AICLI 会把列表同步给所有引擎,这里做成空操作 + pass + + +def test_autosave_creates_session_lazily(tmp_history): + stub = _Stub(tmp_history) + stub.shared_conversation_history = [{"role": "user", "content": "首条"}] + stub._autosave_session() + assert stub.current_session is not None + sessions = stub.session_manager.list_sessions() + assert len(sessions) == 1 + assert sessions[0].msg_count == 1 + + +def test_autosave_updates_existing_session(tmp_history): + stub = _Stub(tmp_history) + stub.shared_conversation_history = [{"role": "user", "content": "a"}] + stub._autosave_session() # 建 + stub.shared_conversation_history.append({"role": "assistant", "content": "b"}) + stub._autosave_session() # 更新 + sessions = stub.session_manager.list_sessions() + assert len(sessions) == 1 + assert sessions[0].msg_count == 2 + + +def test_handle_resume_lists_and_restores(tmp_history, capsys): + stub = _Stub(tmp_history) + # 先写一个会话 + stub.shared_conversation_history = [{"role": "user", "content": "历史问题"}] + stub._autosave_session() + first_id = stub.current_session.id + + # 清空当前上下文,模拟新会话 + stub.shared_conversation_history = [] + stub.current_session = None + + # /resume 列出 + stub.handle_resume_command("") + out = capsys.readouterr().out + assert "已保存的会话" in out + + # /resume 恢复 + stub.handle_resume_command(first_id) + assert stub.shared_conversation_history == [{"role": "user", "content": "历史问题"}] + assert stub.current_session.id == first_id + + +def test_handle_resume_new_clears_current(tmp_history): + stub = _Stub(tmp_history) + stub.shared_conversation_history = [{"role": "user", "content": "x"}] + stub._autosave_session() + assert stub.current_session is not None + stub.handle_resume_command("new") + assert stub.current_session is None + + +def test_handle_resume_unknown_token(tmp_history, capsys): + stub = _Stub(tmp_history) + # 先写一个会话,使目录非空,才能走到「未找到」分支而非「没有可恢复」 + stub.shared_conversation_history = [{"role": "user", "content": "历史"}] + stub._autosave_session() + stub.handle_resume_command("nonexistent-id") + out = capsys.readouterr().out + assert "未找到会话" in out diff --git a/xcli_core/cli_base.py b/xcli_core/cli_base.py index ba77c44..dab83cb 100644 --- a/xcli_core/cli_base.py +++ b/xcli_core/cli_base.py @@ -97,6 +97,8 @@ class BaseAICLI: self.chat_history_dir = os.path.join(project_dir, "chat_history") if not os.path.exists(self.chat_history_dir): os.makedirs(self.chat_history_dir) + # 初始化会话管理器(opencode 式 /resume 持久化) + self.init_session_manager() # 获取 skills 目录路径 skills_dir = os.path.join(project_dir, "skills") diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py index 7b140fa..ff96ce0 100644 --- a/xcli_core/cli_core.py +++ b/xcli_core/cli_core.py @@ -1040,7 +1040,17 @@ multi 操作支持一次修改多处: # 一行速览,完整命令清单收进 /help print(f"{Fore.CYAN}/help 全部命令 · /quit 退出 · /tui 图形界面 · " - f"/model 换模型 · @文件路径 读文件给 AI{Style.RESET_ALL}") + f"/model 换模型 · /resume 恢复会话 · @文件路径 读文件给 AI{Style.RESET_ALL}") + + # 历史会话提示(opencode 式 /resume) + try: + sessions = self.session_manager.list_sessions() + if sessions: + latest = sessions[0] + print(f"{Fore.YELLOW}有 {len(sessions)} 个历史会话,/resume 可恢复" + f"(最近: {latest.id} · {latest.title}){Style.RESET_ALL}") + except Exception: + pass extras = [] loaded_engines = list(self.engines.keys()) @@ -1146,6 +1156,19 @@ multi 操作支持一次修改多处: self.handle_model_command(user_input[7:].strip()) continue + # /resume 会话恢复(opencode 式自动持久化 + 一键恢复) + if user_input in ('/resume', '/sessions') or \ + user_input.startswith('/resume ') or user_input.startswith('/sessions ') or \ + user_input.startswith('/session '): + if user_input.startswith('/resume'): + self.handle_resume_command(user_input[7:].strip()) + elif user_input.startswith('/sessions'): + self.handle_resume_command(user_input[9:].strip()) + else: # /session <子命令> + sub = user_input[8:].strip() + self.handle_resume_command(sub if sub != 'new' else 'new') + continue + # /manual 快捷命令 — 切换到 manual 引擎或执行 manual 子命令 if user_input == '/manual' or user_input.startswith('/manual '): args = user_input[8:].strip() if user_input.startswith('/manual ') else '' @@ -1393,6 +1416,8 @@ multi 操作支持一次修改多处: print(f"{Fore.CYAN}{'='*50}{Style.RESET_ALL}") self.process_conversation(user_input) + # 自动持久化当前会话(opencode 式 /resume 落盘) + self._autosave_session() except KeyboardInterrupt: user_id_display = f"[用户ID: {self.user_id}]" print(f"\n{Fore.GREEN}再见! {user_id_display}{Style.RESET_ALL}") diff --git a/xcli_core/cli_display.py b/xcli_core/cli_display.py index 7536764..8b85a10 100644 --- a/xcli_core/cli_display.py +++ b/xcli_core/cli_display.py @@ -399,6 +399,9 @@ class DisplayMixin: ('@图片路径', '自动分析图片并发送描述给 AI'), ('/image engines', '查看图像识别引擎'), ('/chat save|list|open', '保存/查看/加载聊天记录'), + ('/resume', '列出历史会话 (自动持久化)'), + ('/resume <序号或ID>', '恢复指定会话继续对话'), + ('/resume new', '开始一个全新的会话'), ('/remote', '查看远程连接帮助'), ('/plugin', '管理插件市场 (安装/卸载/搜索)'), ('/protect', '查看进程保护状态'), diff --git a/xcli_core/cli_history.py b/xcli_core/cli_history.py index 82c2538..5a07fe1 100644 --- a/xcli_core/cli_history.py +++ b/xcli_core/cli_history.py @@ -2,10 +2,11 @@ import os import time import json from colorama import Fore, Style +from .session import SessionManager class HistoryMixin: - """聊天记录管理相关方法""" + """聊天记录管理相关方法(含 opencode 式会话持久化 /resume)""" def save_chat_history(self, name): """保存当前聊天记录到文件""" @@ -58,3 +59,62 @@ class HistoryMixin: except Exception as e: print(f"{Fore.RED}加载聊天记录失败: {e}{Style.RESET_ALL}") return False + + # ── opencode 式会话(自动持久化 + /resume) ── + + def init_session_manager(self): + """初始化会话管理器与当前会话状态。在 chat_history_dir 就绪后调用。""" + self.session_manager = SessionManager(self.chat_history_dir) + self.current_session = None # 首轮对话时惰性创建 + + def _autosave_session(self): + """每轮对话后调用:把当前会话落盘(首轮自动建会话)。""" + if not getattr(self, 'shared_conversation_history', None): + return + if not hasattr(self, 'session_manager'): + return + engine = getattr(self.current_engine, 'name', None) if self.current_engine else None + model = getattr(self, 'current_model', None) + try: + if self.current_session is None: + self.current_session = self.session_manager.new_session(engine=engine, model=model) + self.session_manager.save( + self.current_session, self.shared_conversation_history, + engine=engine, model=model, + ) + except Exception: + pass # 自动保存失败不应中断对话 + + def handle_resume_command(self, args): + """/resume 命令:列出或恢复历史会话。""" + args = (args or "").strip() + if not hasattr(self, 'session_manager'): + self.init_session_manager() + if args == 'new': + self.current_session = None + print(f"{Fore.GREEN}已开始新会话(下一轮对话自动创建){Style.RESET_ALL}") + return + sessions = self.session_manager.list_sessions() + if not sessions: + print(f"{Fore.YELLOW}没有可恢复的会话{Style.RESET_ALL}") + return + if not args: + print(f"{Fore.GREEN}已保存的会话 (按时间倒序):{Style.RESET_ALL}") + for i, s in enumerate(sessions, 1): + line = f"{Fore.GREEN}{i}. {s.id}{Style.RESET_ALL} {s.title}" + meta = f" {s.updated_at} {s.engine or '-'}/{s.model or '-'} {s.msg_count} 条" + print(line + f"{Fore.CYAN}{meta}{Style.RESET_ALL}") + print(f"{Fore.CYAN}用法: /resume <序号或ID> 恢复 · /resume new 开新会话{Style.RESET_ALL}") + return + s = self.session_manager.load(args) + if s is None: + print(f"{Fore.RED}未找到会话: {args}{Style.RESET_ALL}") + return + self.shared_conversation_history = list(s.messages) + self.current_session = s + self._set_shared_conversation_history() + print(f"{Fore.GREEN}已恢复会话: {s.id}{Style.RESET_ALL}") + print(f"{Fore.CYAN}标题: {s.title} · {s.msg_count} 条记录 · 引擎 {s.engine or '-'}/{s.model or '-'}{Style.RESET_ALL}") + cur = getattr(self.current_engine, 'name', None) if self.current_engine else None + if s.engine and cur and cur != s.engine: + print(f"{Fore.YELLOW}提示: 当前引擎为 {cur},可用 /engine switch {s.engine} 切回该会话的引擎{Style.RESET_ALL}") diff --git a/xcli_core/session.py b/xcli_core/session.py new file mode 100644 index 0000000..da0457d --- /dev/null +++ b/xcli_core/session.py @@ -0,0 +1,201 @@ +"""会话(Session)持久化与恢复。 + +把 openai/ollama/manual 等引擎的对话历史(self.shared_conversation_history) +按「会话」维度持久化到 chat_history/.json,支持: + - 自动落盘(每轮对话后调用 SessionManager.save) + - 列出历史会话(/resume、/sessions) + - 一键恢复(把 messages 重新载入 shared_conversation_history) + +存储格式(新): + { + "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 + + +def _make_session_id() -> str: + """生成形如 20260805-171045-a1b2c3 的会话 ID。""" + ts = time.strftime("%Y%m%d-%H%M%S") + suffix = uuid.uuid4().hex[:6] + return f"{ts}-{suffix}" + + +def _title_from_messages(messages, limit: int = 40) -> str: + """从首条 user 消息生成会话标题。""" + for m in messages or []: + if isinstance(m, dict) and m.get("role") == "user": + content = m.get("content", "") + if isinstance(content, list): # 多模态 content 可能是列表 + parts = [c.get("text", "") for c in content if isinstance(c, dict)] + content = " ".join(parts) + content = (content or "").strip().replace("\n", " ") + return content[:limit] if content else "(空消息)" + return "(未命名会话)" + + +class Session: + """单个会话的轻量容器。""" + + def __init__(self, sid, title, created_at, updated_at, engine, model, messages): + self.id = sid + self.title = title + self.created_at = created_at + self.updated_at = updated_at + self.engine = engine + self.model = model + self.messages = messages or [] + + @property + def msg_count(self) -> int: + return len(self.messages) + + def to_dict(self) -> dict: + return { + "id": self.id, + "title": self.title, + "created_at": self.created_at, + "updated_at": self.updated_at, + "engine": self.engine, + "model": self.model, + "messages": self.messages, + } + + @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: + messages = data["conversation"] + return cls( + sid=sid, + title=data.get("title") or _title_from_messages(messages), + created_at=data.get("created_at") or data.get("timestamp") or "", + updated_at=data.get("updated_at") or data.get("timestamp") or "", + engine=data.get("engine") or "", + model=data.get("model") or "", + messages=messages or [], + ) + + +class SessionManager: + """会话的读写与索引管理。""" + + def __init__(self, history_dir: str): + self.history_dir = history_dir + if history_dir and not os.path.exists(history_dir): + try: + os.makedirs(history_dir) + except OSError: + pass + + # ── 新建 ── + def new_session(self, engine: str = "", model: str = ""): + now = time.strftime("%Y-%m-%d %H:%M:%S") + return Session( + sid=_make_session_id(), + title="(新会话)", + created_at=now, + updated_at=now, + engine=engine or "", + model=model or "", + messages=[], + ) + + # ── 保存 ── + def save(self, session: Session, messages, engine: str = "", model: str = ""): + """把当前对话内容写回到 session 对应的文件。""" + if session is None: + return False + session.messages = list(messages or []) + if not session.engine and engine: + 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") + if not session.created_at: + session.created_at = session.updated_at + file_path = os.path.join(self.history_dir, f"{session.id}.json") + try: + with open(file_path, "w", encoding="utf-8") as f: + json.dump(session.to_dict(), f, ensure_ascii=False, indent=2, default=str) + return True + except OSError: + return False + + # ── 列出 ── + def list_sessions(self) -> list: + """返回按更新时间倒序的 Session 列表。""" + 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): + continue + sessions.sort(key=lambda s: s.updated_at, reverse=True) + return sessions + + # ── 按 id 或序号解析 ── + def resolve(self, token: str, sessions: list = None) -> Session: + """token 可以是会话 id(含文件名去后缀)或列表序号(从1)。""" + 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: + if s.id == token: + return s + return None + + 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") + 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