From c342c4e94a42edca9be73f5682c02e632fd4b218 Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Sat, 15 Aug 2026 22:25:39 +0800 Subject: [PATCH] =?UTF-8?q?myclaude=20-r:=20=E5=9B=BA=E5=AE=9A=E5=9D=97=20?= =?UTF-8?q?ANSI=20=E9=87=8D=E7=BB=98=E4=BF=AE=E5=A4=8D=E5=AF=B9=E9=BD=90?= =?UTF-8?q?=E4=B8=8E=E6=BB=9A=E5=8A=A8=E5=86=B2=E7=AA=81=20+=20cli=20?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=20mock=20=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - picker 弃用 Live 整屏接管，改固定 12 行块增量重绘（\x1b[{n}A 回到块顶逐行覆盖），scrollback 不受影响 - 每行单行渲染，title 列定宽 43 + CJK 感知截断，布局按 80 列设计 - tests: 新增 TestPickerBlock；test_cli which mock 改 side_effect 只命中后端二进制，避免 _npm_global_bin 多跑 subprocess.run；裸 --resume 测试更新为新 picker 语义 --- myagents/launcher.py | 185 +++++++++++++++++++------------- tests/test_cli.py | 64 +++++++---- tests/test_launcher_sessions.py | 57 ++++++++++ 3 files changed, 211 insertions(+), 95 deletions(-) diff --git a/myagents/launcher.py b/myagents/launcher.py index 00c879b..c0611d4 100644 --- a/myagents/launcher.py +++ b/myagents/launcher.py @@ -10,6 +10,7 @@ import shutil import sqlite3 import subprocess import sys +import unicodedata from dataclasses import dataclass from datetime import datetime from importlib.metadata import PackageNotFoundError, version @@ -17,8 +18,7 @@ from pathlib import Path import click from rich import box -from rich.console import Console, Group -from rich.live import Live +from rich.console import Console from rich.table import Table from rich.text import Text @@ -760,7 +760,7 @@ def _claude_session_rows(chat_cwd: Path) -> list[ClaudeSessionRow]: def _render_claude_table( rows: list[ClaudeSessionRow], chat_cwd: Path, numbered: bool ) -> None: - """Shared rich table for ``myclaude -l`` and the resume picker.""" + """Rich table for ``myclaude -l`` / the non-tty resume listing.""" console.print(f"[bold]Sessions in[/bold] {chat_cwd}") table = Table( box=box.SIMPLE_HEAD, @@ -923,69 +923,111 @@ def _picker_advance( return state, "none", None -def _render_picker_window( +def _disp_width(text: str) -> int: + """Visible width of ``text``; CJK/fullwidth chars count as 2 columns.""" + return sum( + 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1 for ch in text + ) + + +def _pad_disp(text: str, width: int) -> str: + """Left-pad/justify ``text`` to ``width`` visible columns (CJK-aware).""" + pad = width - _disp_width(text) + return text + " " * pad if pad > 0 else text + + +def _truncate_disp(text: str, width: int) -> str: + """Truncate ``text`` to ``width`` visible columns, ellipsize overflow.""" + if _disp_width(text) <= width: + return text + out: list[str] = [] + used = 0 + for ch in text: + w = 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1 + if used + w > width - 1: + break + out.append(ch) + used += w + return "".join(out) + "…" + + +def _picker_line( + row: ClaudeSessionRow, idx: int, *, selected: bool, title_width: int +) -> str: + """One terminal-safe line for a session row (single line, CJK-truncated).""" + title = _truncate_disp(row.title or "(empty)", title_width) + updated = datetime.fromtimestamp(row.updated_at).strftime("%Y-%m-%d %H:%M") + line = ( + f"{idx:>4} {_pad_disp(title, title_width)} " + f"{_pad_disp(row.origin, 6)} {_pad_disp(row.status or '', 7)} {updated}" + ) + if selected: + return f"\x1b[7m{line}\x1b[0m" + if row.jsonl_path is None: + return f"\x1b[2m{line}\x1b[0m" # client-only: no local jsonl to resume + return line + + +def _picker_block( state: _PickerState, chat_cwd: Path, message: str | None -) -> Group: - """Live renderable for the windowed picker (10 rows + status line).""" +) -> list[str]: + """Lines for the picker block (header + window rows + status line). + + Layout is designed for an 80-column terminal: title column is fixed so + rows stay aligned regardless of the actual terminal width. + """ rows = state.rows n = len(rows) start = state.offset end = min(start + _PICKER_WINDOW, n) - - table = Table( - box=box.SIMPLE_HEAD, - show_header=True, - header_style="dim", - expand=True, - pad_edge=False, - collapse_padding=True, + # 80 cols total: "#(5) + title(43) + Src(7) + Status(8) + Updated(17)". + title_width = 43 + header = ( + f"Sessions in {chat_cwd} ({start + 1}-{end}/{n})" + if n + else f"Sessions in {chat_cwd}" ) - table.add_column("#", justify="right", no_wrap=True, style="dim", min_width=3) - table.add_column("Title", overflow="ellipsis", no_wrap=True, ratio=1, min_width=24) - table.add_column("Src", no_wrap=True, justify="center", min_width=6) - table.add_column("Status", overflow="ellipsis", no_wrap=True, style="dim", min_width=8) - table.add_column("Updated", justify="right", no_wrap=True, style="dim", min_width=16) + lines = [header] for i in range(start, end): - row = rows[i] - if row.jsonl_path is None: - title_style = "dim italic" - elif not row.title: - title_style = "dim" - else: - title_style = "" - cells = [ - Text(str(i + 1)), - Text(row.title or "(empty)", style=title_style), - Text(row.origin, style="magenta" if row.origin == "xiaohe" else "dim"), - Text(row.status or ""), - Text(datetime.fromtimestamp(row.updated_at).strftime("%Y-%m-%d %H:%M")), - ] - table.add_row(*cells, style="reverse" if i - start == state.cursor else "") - header = Text.assemble( - ("Sessions in ", "bold"), - (str(chat_cwd), ""), - (f" ({start + 1}-{end}/{n})", "dim"), - ) + lines.append( + _picker_line(rows[i], i + 1, selected=(i - start == state.cursor), title_width=title_width) + ) if message: - status = Text(message, style="yellow") + status = f"\x1b[33m{message}\x1b[0m" elif state.buffer: - status = Text.assemble( - ("select #", "cyan"), - (state.buffer, "bold cyan"), - (" · Enter confirm · ⌫ clear", "dim"), - ) + status = f"select #{state.buffer} · Enter confirm · ⌫ clear" else: - status = Text( - "↑↓ scroll · Enter select · digits+Enter jump · PgUp/PgDn page · q quit", - style="dim", - ) - return Group(header, table, status) + status = "↑↓ move · Enter select · digits+Enter jump · q quit" + lines.append(status) + return lines + + +def _draw_picker_block(lines: list[str], prev_height: int) -> None: + """Rewrite the picker block in place: cursor up ``prev_height`` then overwrite. + + Only the fixed block region is touched — the rest of the screen and the + terminal scrollback are left alone, so there is no scroll conflict. + """ + if prev_height: + sys.stdout.write(f"\x1b[{prev_height}A") + for line in lines: + sys.stdout.write("\r\x1b[2K" + line + "\n") + sys.stdout.flush() + + +def _clear_picker_block(prev_height: int) -> None: + """Blank the picker block lines before exiting so no residue remains.""" + if prev_height: + sys.stdout.write(f"\x1b[{prev_height}A") + for _ in range(prev_height): + sys.stdout.write("\r\x1b[2K\n") + sys.stdout.flush() def _pick_session( rows: list[ClaudeSessionRow], chat_cwd: Path ) -> ClaudeSessionRow | None: - """Run the interactive windowed picker on a raw tty; None on quit.""" + """Run the interactive picker on a raw tty; None on quit.""" import termios import tty @@ -993,27 +1035,25 @@ def _pick_session( fd = sys.stdin.fileno() old = termios.tcgetattr(fd) message: str | None = None + prev_height = 0 try: tty.setraw(fd) - live = Live( - _render_picker_window(state, chat_cwd, message), - console=console, - refresh_per_second=4, - transient=True, - ) - with live: - while True: - key = _read_key(fd) - if key == "timeout": - continue # idle: no redraw; Live's low-rate refresh covers resize - state, action, payload = _picker_advance(state, key) - message = payload if isinstance(payload, str) else None - live.update(_render_picker_window(state, chat_cwd, message)) - if action == "select": - assert isinstance(payload, ClaudeSessionRow) - return payload - if action == "quit": - return None + while True: + lines = _picker_block(state, chat_cwd, message) + _draw_picker_block(lines, prev_height) + prev_height = len(lines) + key = _read_key(fd) + if key == "timeout": + continue # idle: block is stable, no redraw needed + state, action, payload = _picker_advance(state, key) + message = payload if isinstance(payload, str) else None + if action == "select": + _clear_picker_block(prev_height) + assert isinstance(payload, ClaudeSessionRow) + return payload + if action == "quit": + _clear_picker_block(prev_height) + return None finally: termios.tcsetattr(fd, termios.TCSADRAIN, old) @@ -1021,8 +1061,9 @@ def _pick_session( def _resume_picker_claude(chat_cwd: Path, use_tmux: bool = False) -> None: """Interactive unified resume picker for the claude backend. - Lists jsonl + xiaohe client sessions for chat_cwd in a 10-row scrollable - window; selecting one resumes the jsonl via ``claude --resume ``. + Lists jsonl + xiaohe client sessions for chat_cwd in a fixed 10-row block + (↑↓ navigate, digits+Enter jump, Enter select, q quit); selecting one + resumes the jsonl via ``claude --resume ``. """ rows = _claude_session_rows(chat_cwd) if not rows: diff --git a/tests/test_cli.py b/tests/test_cli.py index e817765..73cee89 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -11,6 +11,20 @@ import myagents.launcher from myagents.cli import cli +def _backend_which(*names: str): + """Patch ``shutil.which`` to find backend binaries and nothing else. + + The launcher also probes ``which("npm")`` while resolving npm-global + installs; returning None there keeps that probe from spawning a real + ``subprocess.run`` in tests (which would break ``assert_called_once``). + """ + + def _which(name: str) -> str | None: + return f"/usr/bin/{name}" if name in names else None + + return _which + + class TestMyagentsHelp: """Tests for top-level myagents command.""" @@ -74,7 +88,7 @@ class TestClaudeSubcommand: def test_runs_claude(self) -> None: runner = CliRunner() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/claude"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("claude")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -99,7 +113,7 @@ class TestClaudeSubcommand: test_dir.mkdir() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/claude"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("claude")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -121,7 +135,7 @@ class TestClaudePassthrough: def _invoke(self, args: list[str], tmp_path: Path) -> list[str]: runner = CliRunner() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/claude"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("claude")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -132,13 +146,17 @@ class TestClaudePassthrough: mock_run.assert_called_once() return mock_run.call_args[0][0] - def test_resume_flag_passes_through(self, tmp_path: Path) -> None: - cmd = self._invoke(["--resume"], tmp_path) - assert cmd == [ - "/usr/bin/claude", - "--dangerously-skip-permissions", - "--resume", - ] + def test_bare_resume_opens_picker_not_launch(self, tmp_path: Path) -> None: + """Bare --resume opens the unified picker; non-tty lists and exits.""" + runner = CliRunner() + with ( + patch("myagents.launcher.shutil.which", side_effect=_backend_which("claude")), + patch("myagents.launcher.subprocess.run") as mock_run, + ): + mock_run.return_value = MagicMock(returncode=0) + result = runner.invoke(cli, ["claude", "--cwd", str(tmp_path), "--resume"]) + assert result.exit_code == 0, result.output + mock_run.assert_not_called() def test_resume_with_session_id(self, tmp_path: Path) -> None: cmd = self._invoke(["-r", "abc123"], tmp_path) @@ -175,7 +193,7 @@ class TestKimiSubcommand: def test_runs_kimi(self) -> None: runner = CliRunner() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/kimi"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("kimi")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -197,7 +215,7 @@ class TestKimiSubcommand: test_dir.mkdir() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/kimi"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("kimi")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -212,7 +230,7 @@ class TestKimiPassthrough: def _invoke(self, args: list[str], tmp_path: Path) -> list[str]: runner = CliRunner() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/kimi"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("kimi")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -265,7 +283,7 @@ class TestCodexSubcommand: def test_runs_codex(self) -> None: runner = CliRunner() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/codex"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("codex")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -287,7 +305,7 @@ class TestCodexSubcommand: test_dir.mkdir() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/codex"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("codex")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -302,7 +320,7 @@ class TestCodexPassthrough: def _invoke(self, args: list[str], tmp_path: Path) -> list[str]: runner = CliRunner() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/codex"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("codex")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -328,7 +346,7 @@ class TestCodexResumeOptions: def _invoke(self, args: list[str], tmp_path: Path) -> list[str]: runner = CliRunner() with ( - patch("myagents.launcher.shutil.which", return_value="/usr/bin/codex"), + patch("myagents.launcher.shutil.which", side_effect=_backend_which("codex")), patch("myagents.launcher.subprocess.run") as mock_run, ): mock_run.return_value = MagicMock(returncode=0) @@ -437,7 +455,7 @@ class TestHermesSubcommand: runner = CliRunner() with ( patch( - "myagents.launcher.shutil.which", return_value="/usr/bin/hermes" + "myagents.launcher.shutil.which", side_effect=_backend_which("hermes") ), patch("myagents.launcher.subprocess.run") as mock_run, ): @@ -461,7 +479,7 @@ class TestHermesSubcommand: with ( patch( - "myagents.launcher.shutil.which", return_value="/usr/bin/hermes" + "myagents.launcher.shutil.which", side_effect=_backend_which("hermes") ), patch("myagents.launcher.subprocess.run") as mock_run, ): @@ -478,7 +496,7 @@ class TestHermesPassthrough: runner = CliRunner() with ( patch( - "myagents.launcher.shutil.which", return_value="/usr/bin/hermes" + "myagents.launcher.shutil.which", side_effect=_backend_which("hermes") ), patch("myagents.launcher.subprocess.run") as mock_run, ): @@ -616,7 +634,7 @@ class TestCursorSubcommand: runner = CliRunner() with ( patch( - "myagents.launcher.shutil.which", return_value="/usr/bin/agent" + "myagents.launcher.shutil.which", side_effect=_backend_which("agent") ), patch("myagents.launcher.subprocess.run") as mock_run, ): @@ -664,7 +682,7 @@ class TestCursorSubcommand: with ( patch( - "myagents.launcher.shutil.which", return_value="/usr/bin/agent" + "myagents.launcher.shutil.which", side_effect=_backend_which("agent") ), patch("myagents.launcher.subprocess.run") as mock_run, ): @@ -681,7 +699,7 @@ class TestCursorPassthrough: runner = CliRunner() with ( patch( - "myagents.launcher.shutil.which", return_value="/usr/bin/agent" + "myagents.launcher.shutil.which", side_effect=_backend_which("agent") ), patch("myagents.launcher.subprocess.run") as mock_run, ): diff --git a/tests/test_launcher_sessions.py b/tests/test_launcher_sessions.py index c123391..3ede9dd 100644 --- a/tests/test_launcher_sessions.py +++ b/tests/test_launcher_sessions.py @@ -189,6 +189,63 @@ def _make_rows(claude_env, n: int): return L._claude_session_rows(claude_env.cwd) +def _ansi_strip(text: str) -> str: + return re.sub(r"\x1b\[[0-9;]*m", "", text) + + +class TestPickerBlock: + """Fixed 80-col block rendering (header + rows + status), CJK-aware.""" + + def _rows(self, claude_env, n: int): + return _make_rows(claude_env, n) + + def test_rows_are_single_line_within_80(self, claude_env) -> None: + rows = self._rows(claude_env, 3) + state = L._PickerState(rows=rows) + lines = L._picker_block(state, claude_env.cwd, None) + # header + 3 rows + status + assert len(lines) == 5 + for line in lines[1:4]: + assert len(_ansi_strip(line)) <= 80 + assert "Sessions in" in lines[0] + + def test_selected_row_reversed(self, claude_env) -> None: + rows = self._rows(claude_env, 3) + state = L._PickerState(rows=rows, cursor=1) + lines = L._picker_block(state, claude_env.cwd, None) + assert "\x1b[7m" in lines[2] + assert "\x1b[7m" not in lines[1] + + def test_title_truncated_with_ellipsis(self, claude_env) -> None: + # _make_rows writes title-less jsonl; inject a long CJK title directly. + row = L.ClaudeSessionRow( + cli_session_id="long", + title="很长的中文标题" * 10, + origin="cli", + status=None, + updated_at=1000.0, + jsonl_path=Path("x.jsonl"), + xiaohe_session_id=None, + ) + state = L._PickerState(rows=[row]) + lines = L._picker_block(state, claude_env.cwd, None) + plain = _ansi_strip(lines[1]) + assert "…" in plain + assert len(plain) <= 80 + + def test_message_line_replaces_hint(self, claude_env) -> None: + rows = self._rows(claude_env, 3) + state = L._PickerState(rows=rows) + lines = L._picker_block(state, claude_env.cwd, "client-only: no jsonl") + assert "client-only" in lines[-1] + assert "↑↓" not in lines[-1] + + def test_cjk_display_width(self) -> None: + assert L._disp_width("ab") == 2 + assert L._disp_width("中文") == 4 + assert L._disp_width("a中") == 3 + + class TestPickerAdvance: """Windowed picker state machine (no tty needed)."""