From addf119ac5e23248b61261546ef14d892f044487 Mon Sep 17 00:00:00 2001 From: shuiyu67 Date: Wed, 5 Aug 2026 19:43:30 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20TUI=20=E6=B7=B1=E5=BA=A6=EF=BC=88?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E6=A0=91=20+=20=E4=BC=9A=E8=AF=9D=E5=88=97?= =?UTF-8?q?=E8=A1=A8=20+=20Plan=20=E7=8A=B6=E6=80=81=E6=A0=8F=EF=BC=8Chead?= =?UTF-8?q?less=20=E8=87=AA=E9=AA=8C=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_tui_features.py | 60 +++++++++ tools/tui_real_snapshot.py | 72 +++++++++++ xcli_core/tui.py | 248 +++++++++++++++++++++++++++++++++++-- 3 files changed, 367 insertions(+), 13 deletions(-) create mode 100644 tests/test_tui_features.py diff --git a/tests/test_tui_features.py b/tests/test_tui_features.py new file mode 100644 index 0000000..cba1c68 --- /dev/null +++ b/tests/test_tui_features.py @@ -0,0 +1,60 @@ +"""TUI 深度功能测试：文件树 / 会话相对时间 / Plan 状态（纯函数可测部分）""" +import os +import tempfile +from datetime import datetime, timedelta + +import pytest + +from xcli_core.tui import XiaoliTUI, TEXTUAL_AVAILABLE + +pytestmark = pytest.mark.skipif(not TEXTUAL_AVAILABLE, reason="textual 未安装") + + +def test_build_file_tree_lists_dirs_and_files(): + with tempfile.TemporaryDirectory() as d: + os.mkdir(os.path.join(d, "sub")) + open(os.path.join(d, "a.py"), "w").close() + open(os.path.join(d, "b.txt"), "w").close() + items = XiaoliTUI._build_file_tree(d) + names = [n for n, _ in items] + assert "sub" in names and "a.py" in names and "b.txt" in names + dirs = [n for n, is_dir in items if is_dir] + assert "sub" in dirs + + +def test_build_file_tree_skips_noise(): + with tempfile.TemporaryDirectory() as d: + os.mkdir(os.path.join(d, ".git")) + os.mkdir(os.path.join(d, "__pycache__")) + open(os.path.join(d, "real.py"), "w").close() + items = XiaoliTUI._build_file_tree(d) + names = [n for n, _ in items] + assert ".git" not in names and "__pycache__" not in names + assert "real.py" in names + + +def test_build_file_tree_max_items(): + with tempfile.TemporaryDirectory() as d: + for i in range(100): + open(os.path.join(d, f"f{i}.txt"), "w").close() + items = XiaoliTUI._build_file_tree(d, max_items=10) + assert any(n == "…" for n, _ in items) + real = [n for n, _ in items if n != "…"] + assert len(real) <= 10 + + +def test_format_rel_time(): + now = datetime.now() + assert "刚刚" in XiaoliTUI._format_rel_time(now.isoformat()) + assert "分钟前" in XiaoliTUI._format_rel_time((now - timedelta(minutes=5)).isoformat()) + assert "小时前" in XiaoliTUI._format_rel_time((now - timedelta(hours=3)).isoformat()) + assert "天前" in XiaoliTUI._format_rel_time((now - timedelta(days=2)).isoformat()) + assert XiaoliTUI._format_rel_time("") == "" + + +def test_tui_has_plan_and_sidebar_hooks(): + """TUI 深度接入点存在（方法/反应变量/侧栏节点 id）。""" + assert hasattr(XiaoliTUI, "_tui_plan") + assert hasattr(XiaoliTUI, "_update_file_tree") + assert hasattr(XiaoliTUI, "_update_session_list") + assert "plan_mode" in XiaoliTUI._reactives if hasattr(XiaoliTUI, "_reactives") else True diff --git a/tools/tui_real_snapshot.py b/tools/tui_real_snapshot.py index e1c2506..6f78133 100644 --- a/tools/tui_real_snapshot.py +++ b/tools/tui_real_snapshot.py @@ -35,6 +35,25 @@ class _MockMgr: ] +class _MockSession: + def __init__(self, title, updated_at, sid): + self.title = title + self.updated_at = updated_at + self.id = sid + + +class _MockSessionManager: + """模拟 SessionManager.list_sessions，返回几个示例会话。""" + def list_sessions(self): + from datetime import datetime, timedelta + now = datetime.now() + return [ + _MockSession("修复内存泄漏", (now - timedelta(minutes=5)).isoformat(), "s1"), + _MockSession("重构 CLI 入口", (now - timedelta(hours=2)).isoformat(), "s2"), + _MockSession("接入 MCP 客户端", (now - timedelta(days=1)).isoformat(), "s3"), + ] + + class _MockCLI: def __init__(self): self.engines = {"openai": _MockEngine()} @@ -43,6 +62,10 @@ class _MockCLI: self.shared_conversation_history = [] self.liugin_commands = {} self.tui_output_callback = None + # TUI 深度：会话列表 / Plan 状态依赖 + self.session_manager = _MockSessionManager() + self.plan_mode = False + self.current_plan = "" def get_current_engine_name(self): return "openai" @@ -69,6 +92,18 @@ def _txt(w): return str(getattr(w, "content", "") or getattr(w, "renderable", "") or "") +async def _svg_to_png(svg_path, png_path, width=1280, height=720): + """用 Playwright Chromium 把 textual 导出的 SVG 渲染成 PNG（AI 肉眼核对用）。""" + from playwright.async_api import async_playwright + async with async_playwright() as p: + browser = await p.chromium.launch() + page = await browser.new_page(viewport={"width": width, "height": height}) + await page.goto("file://" + os.path.abspath(svg_path)) + await page.wait_for_timeout(400) + await page.screenshot(path=png_path, full_page=True) + await browser.close() + + async def run(): app = XiaoliTUI(_MockCLI()) async with app.run_test(size=(120, 64)) as pilot: @@ -107,6 +142,35 @@ async def run(): for k, ok in checks.items(): print(f"[{'PASS' if ok else 'FAIL'}] {k}") + # 3.5) 侧栏文件树 + 会话列表（TUI 深度） + file_tree = app.query_one("#file-tree") + session_list = app.query_one("#session-list") + checks["文件树渲染"] = len(file_tree.children) > 1 + checks["会话列表渲染"] = len(session_list.children) > 1 + print(f"[{'PASS' if checks['文件树渲染'] else 'FAIL'}] 文件树渲染 (children={len(file_tree.children)})") + print(f"[{'PASS' if checks['会话列表渲染'] else 'FAIL'}] 会话列表渲染 (children={len(session_list.children)})") + + # 3.6) 进入 PLAN 模式，断言状态栏 + 侧栏同步 + ta.text = "/plan 重构核心模块" + app.action_send_message() + for _ in range(80): + await pilot.pause(0.05) + if not app.is_generating: + break + await pilot.pause(0.2) + status_txt_plan = _txt(app.query_one("#status-bar")) + sidebar_widget = app.query_one("#status-info") + sidebar_txt_plan = "\n".join(_txt(w) for w in sidebar_widget.children) + checks["PLAN 状态栏"] = "PLAN" in status_txt_plan + checks["PLAN 侧栏"] = "PLAN" in sidebar_txt_plan + print(f"[{'PASS' if checks['PLAN 状态栏'] else 'FAIL'}] PLAN 状态栏: {status_txt_plan!r}") + print(f"[{'PASS' if checks['PLAN 侧栏'] else 'FAIL'}] PLAN 侧栏: {sidebar_txt_plan!r}") + + # 3.7) 导出侧栏完整文本（文件树 / 会话列表 / 状态）供核对 + sidebar_full = "\n".join(_txt(w) for w in app.query_one("#sidebar").walk_children()) + with open(os.path.join(OUT_DIR, "real_demo_sidebar.txt"), "w", encoding="utf-8") as f: + f.write(sidebar_full) + # 4) 导出 SVG 截图（AI/人都能看） os.makedirs(OUT_DIR, exist_ok=True) svg_path = os.path.join(OUT_DIR, "real_demo.svg") @@ -115,6 +179,14 @@ async def run(): f.write(svg) print(f"\nSVG 截图已保存: {svg_path} ({len(svg)} bytes)") + # 4b) SVG → PNG（供 AI 肉眼核对布局） + png_path = os.path.join(OUT_DIR, "real_demo.png") + try: + await _svg_to_png(svg_path, png_path) + print(f"PNG 截图已保存: {png_path}") + except Exception as e: + print(f"PNG 转换跳过: {e}") + # 5) 同时导出文本快照（供 AI 快速读回核对） text_path = os.path.join(OUT_DIR, "real_demo.txt") with open(text_path, "w", encoding="utf-8") as f: diff --git a/xcli_core/tui.py b/xcli_core/tui.py index 36581f7..00bffc5 100644 --- a/xcli_core/tui.py +++ b/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 + import os from rich.syntax import Syntax from rich.panel import Panel from rich.box import ROUNDED @@ -197,6 +198,28 @@ if TEXTUAL_AVAILABLE: transition: color 300ms in_out_cubic; }} + /* ── 文件树 / 会话列表 ── */ + .file-tree-item {{ + padding: 0 0 0 1; + color: {_Theme.TEXT_MUTED}; + }} + .file-tree-dir {{ + color: {_Theme.ACCENT}; + text-style: bold; + }} + .session-item {{ + padding: 0 0 0 1; + color: {_Theme.TEXT_MUTED}; + }} + .session-item-active {{ + color: {_Theme.SUCCESS}; + text-style: bold; + }} + .plan-badge {{ + color: {_Theme.WARNING}; + text-style: bold; + }} + /* ── 消息样式 ── */ .msg-user {{ color: {_Theme.USER}; @@ -317,6 +340,24 @@ if TEXTUAL_AVAILABLE: return True return False + def sessions(self) -> list: + mgr = getattr(self.cli, 'session_manager', None) + if mgr and hasattr(mgr, 'list_sessions'): + try: + return mgr.list_sessions() + except Exception: + return [] + return [] + + def plan_mode(self) -> bool: + return bool(getattr(self.cli, 'plan_mode', False)) + + def current_plan(self) -> str: + return getattr(self.cli, 'current_plan', '') or "" + + def cwd(self) -> str: + return os.getcwd() + # ═══════════════════════════════════════════════════ # 动画辅助组件 @@ -445,11 +486,13 @@ if TEXTUAL_AVAILABLE: Binding("ctrl+enter", "send_message", "发送", show=False), Binding("shift+enter", "newline", "换行", show=False), Binding("f1", "toggle_sidebar", "侧栏", show=True), + Binding("ctrl+p", "toggle_plan", "PLAN", show=True), Binding("escape", "cancel", "取消", show=False), ] sidebar_visible = var(True) is_generating = var(False) + plan_mode = var(False) def __init__(self, cli): super().__init__() @@ -475,10 +518,16 @@ if TEXTUAL_AVAILABLE: " 回车发送 | Shift+回车换行 | Ctrl+L 清屏 | F1 侧栏", id="input-hint" ) - with Vertical(id="sidebar"): + with VerticalScroll(id="sidebar"): yield Static(" 引擎", classes="sidebar-header") yield Vertical(id="engine-list", classes="sidebar-section") yield Rule(line_style="heavy") + yield Static(" 文件", classes="sidebar-header") + yield Vertical(id="file-tree", classes="sidebar-section") + yield Rule(line_style="heavy") + yield Static(" 会话", classes="sidebar-header") + yield Vertical(id="session-list", classes="sidebar-section") + yield Rule(line_style="heavy") yield Static(" 工具", classes="sidebar-header") yield Vertical(id="tool-list", classes="sidebar-section") yield Rule(line_style="heavy") @@ -489,9 +538,15 @@ if TEXTUAL_AVAILABLE: def on_mount(self): self._render_welcome_animated() self._update_sidebar() + self._sync_plan() self.query_one("#user-input").focus() # 延迟显示输入提示（淡入效果） - self.set_timer(1.5, lambda: self.query_one("#input-hint").add_class("visible")) + def _show_hint(): + try: + self.query_one("#input-hint").add_class("visible") + except Exception: + pass + self.set_timer(1.5, _show_hint) # ── 动画欢迎界面 ── @@ -573,6 +628,9 @@ if TEXTUAL_AVAILABLE: else: engine_container.mount(Static(f" {name}", classes="engine-item")) + self._update_file_tree() + self._update_session_list() + tool_container = self.query_one("#tool-list") tool_container.remove_children() for tool in self.bridge.tools(): @@ -585,14 +643,89 @@ if TEXTUAL_AVAILABLE: status_container.mount(Static(f" 工具: {len(self.bridge.tools())} 个", classes="tool-item")) status_container.mount(Static(f" 引擎: {len(self.bridge.engines())} 个", classes="tool-item")) + # Plan 模式指示 + if self.bridge.plan_mode(): + status_container.mount(Static(" 模式: 📋 PLAN", classes="plan-badge")) + else: + status_container.mount(Static(" 模式: 普通", classes="tool-item")) + # 工具数 if hasattr(self.cli, 'liugin_manager'): status_container.mount(Static(f" 工具: {len(self.cli.liugin_manager.tools)} 个", classes="tool-item")) + # ── 文件树 ── + + @staticmethod + def _build_file_tree(root, max_depth=2, max_items=40): + """生成 (name, is_dir) 列表，用于侧栏文件树展示。""" + items = [] + try: + entries = sorted(os.listdir(root)) + except OSError: + return items + dirs = [e for e in entries if os.path.isdir(os.path.join(root, e))] + files = [e for e in entries if os.path.isfile(os.path.join(root, e))] + skip = {'.git', '__pycache__', '.tui_snapshots', '.pytest_cache', + 'node_modules', '.workbuddy', 'chat_history', 'memory', + '.idea', '.vscode', '.venv', 'venv'} + shown = 0 + for name in dirs + files: + if name in skip or name.startswith('.'): + continue + if shown >= max_items: + items.append(("…", False)) + break + is_dir = os.path.isdir(os.path.join(root, name)) + items.append((name, is_dir)) + shown += 1 + return items + + def _update_file_tree(self): + container = self.query_one("#file-tree") + container.remove_children() + root = self.bridge.cwd() + container.mount(Static(f" 📂 {os.path.basename(root)}", classes="file-tree-dir")) + for name, is_dir in self._build_file_tree(root): + icon = "📁" if is_dir else "📄" + container.mount(Static(f" {icon} {name}", classes="file-tree-item")) + + # ── 会话列表 ── + + @staticmethod + def _format_rel_time(ts: str) -> str: + if not ts: + return "" + try: + from datetime import datetime + dt = datetime.fromisoformat(ts.replace('Z', '+00:00')) + diff = (datetime.now() - dt).total_seconds() + if diff < 60: + return "刚刚" + if diff < 3600: + return f"{int(diff // 60)}分钟前" + if diff < 86400: + return f"{int(diff // 3600)}小时前" + return f"{int(diff // 86400)}天前" + except Exception: + return "" + + def _update_session_list(self): + container = self.query_one("#session-list") + container.remove_children() + sessions = self.bridge.sessions()[:8] + if not sessions: + container.mount(Static(" (无历史会话)", classes="session-item")) + return + for i, s in enumerate(sessions, 1): + title = (s.title or s.id)[:22] + t = self._format_rel_time(s.updated_at) + container.mount(Static(f" {i}. {title} {t}", classes="session-item")) + def _update_status(self, text: str): bar = self.query_one("#status-bar") engine = self.bridge.current_engine() - bar.update(f" {text} | {engine} | {len(self.bridge.history())} 条对话{self._token_str()}") + prefix = "📋 PLAN | " if self.plan_mode else "" + bar.update(f" {prefix}{text} | {engine} | {len(self.bridge.history())} 条对话{self._token_str()}") def _token_str(self): """token 用量进度（v8.0 token 感知压缩联动，引擎支持时显示）""" @@ -775,6 +908,11 @@ if TEXTUAL_AVAILABLE: 'engines': lambda: self._show_engines(), 'tui': lambda: self._system("已在 TUI 模式中"), 'manual': lambda: self._handle_manual(args), + 'plan': lambda: self._tui_plan(args), + 'build': lambda: self._tui_build(), + 'sessions': lambda: self._tui_sessions(), + 'session': lambda: self._tui_session(args), + 'resume': lambda: self._tui_resume(args), 'snapshot': lambda: self._save_screenshot(), 'screenshot': lambda: self._save_screenshot(), } @@ -782,16 +920,17 @@ if TEXTUAL_AVAILABLE: handler = cmds.get(name) if handler: handler() - else: - if name in self.cli.liugin_commands: - try: - result = self.cli.liugin_commands[name](args) - if result: - self._system(result[:500]) - except Exception as e: - self._error(f"插件命令失败: {e}") - else: - self._error(f"未知命令: /{name}，输入 /help 查看帮助") + return + if name in self.cli.liugin_commands: + try: + result = self.cli.liugin_commands[name](args) + if result: + self._system(result[:500]) + except Exception as e: + self._error(f"插件命令失败: {e}") + return + # 未知命令转发给 cli 完整路由（/agent /providers /chat 等） + self._forward_to_cli(cmd) def _show_help(self): help_text = """ 命令: @@ -873,6 +1012,84 @@ if TEXTUAL_AVAILABLE: self._system('\n'.join(status_lines)) + # ── Plan 模式 / 会话 委派 ── + + def _tui_plan(self, args): + args = (args or "").strip() + if args in ('off', 'exit', '退出', 'cancel'): + self._set_plan_mode(False) + self._system("已退出 PLAN 模式（计划未执行）") + return + self._set_plan_mode(True) + if args: + self._system(f"📋 已进入 PLAN 模式，正在只读调研: {args}") + self._user_msg(f"/plan {args}") + self._show_thinking() + self.is_generating = True + self._update_status("思考中 (PLAN)...") + self.run_worker(self._generate(args), exclusive=True) + else: + self._system("📋 已进入 PLAN 模式。描述任务，AI 将只做只读调研并给出实施计划；/build 批准执行，/plan off 取消。") + + def _tui_build(self): + if not self.bridge.plan_mode() and not self.bridge.current_plan().strip(): + self._system("当前不在 PLAN 模式，且无已生成的计划") + return + self._set_plan_mode(False) + self._system("✅ 已批准 PLAN，开始执行。") + + def _tui_sessions(self): + sessions = self.bridge.sessions() + if not sessions: + self._system(" 无历史会话") + return + lines = [" 历史会话:"] + for i, s in enumerate(sessions[:10], 1): + title = (s.title or s.id)[:30] + t = self._format_rel_time(s.updated_at) + lines.append(f" {i}. {title} ({t})") + self._system('\n'.join(lines)) + + def _tui_session(self, args): + args = (args or "").strip() + if not args: + self._tui_sessions() + return + if args == 'new': + self.action_new_chat() + return + self._forward_to_cli(f"/resume {args}") + + def _tui_resume(self, args): + self._forward_to_cli(f"/resume {args}") + + def _forward_to_cli(self, cmd: str): + """把未知/委派命令交给 cli 完整路由处理。""" + try: + self.cli.process_conversation(cmd) + except Exception as e: + self._error(f"命令执行失败: {e}") + self.call_after_refresh(self._update_sidebar) + self.call_after_refresh(self._sync_plan) + + def _set_plan_mode(self, on: bool): + self.plan_mode = on + try: + self.cli.plan_mode = on + if not on: + self.cli.current_plan = "" + except Exception: + pass + self._update_sidebar() + self._update_status("📋 PLAN" if on else "就绪") + + def _sync_plan(self): + """从 cli 同步 plan 状态到 TUI 反应变量。""" + remote = self.bridge.plan_mode() + if remote != self.plan_mode: + self.plan_mode = remote + self._update_sidebar() + # ── AI 生成 ── async def _generate(self, user_input): @@ -902,6 +1119,7 @@ if TEXTUAL_AVAILABLE: self.call_after_refresh(self._remove_thinking) self.call_after_refresh(self._remove_typing_indicator) self.is_generating = False + self.call_after_refresh(self._sync_plan) self.call_after_refresh(lambda: self._update_status("就绪")) self.call_after_refresh(self._update_sidebar) @@ -939,6 +1157,10 @@ if TEXTUAL_AVAILABLE: sidebar.add_class("hidden") self.sidebar_visible = False + def action_toggle_plan(self): + self._set_plan_mode(not self.plan_mode) + self._system("📋 PLAN 模式: " + ("开" if self.plan_mode else "关")) + def action_cancel(self): if self.is_generating: self._remove_thinking()