myclaude -r: 固定块 ANSI 重绘修复对齐与滚动冲突 + cli 测试 mock 修复

- 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 语义
This commit is contained in:
Zhengshou Lai
2026-08-15 22:25:39 +08:00
parent 969fc9ac27
commit c342c4e94a
3 changed files with 211 additions and 95 deletions
+113 -72
View File
@@ -10,6 +10,7 @@ import shutil
import sqlite3 import sqlite3
import subprocess import subprocess
import sys import sys
import unicodedata
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from importlib.metadata import PackageNotFoundError, version from importlib.metadata import PackageNotFoundError, version
@@ -17,8 +18,7 @@ from pathlib import Path
import click import click
from rich import box from rich import box
from rich.console import Console, Group from rich.console import Console
from rich.live import Live
from rich.table import Table from rich.table import Table
from rich.text import Text from rich.text import Text
@@ -760,7 +760,7 @@ def _claude_session_rows(chat_cwd: Path) -> list[ClaudeSessionRow]:
def _render_claude_table( def _render_claude_table(
rows: list[ClaudeSessionRow], chat_cwd: Path, numbered: bool rows: list[ClaudeSessionRow], chat_cwd: Path, numbered: bool
) -> None: ) -> 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}") console.print(f"[bold]Sessions in[/bold] {chat_cwd}")
table = Table( table = Table(
box=box.SIMPLE_HEAD, box=box.SIMPLE_HEAD,
@@ -923,69 +923,111 @@ def _picker_advance(
return state, "none", None 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 state: _PickerState, chat_cwd: Path, message: str | None
) -> Group: ) -> list[str]:
"""Live renderable for the windowed picker (10 rows + status line).""" """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 rows = state.rows
n = len(rows) n = len(rows)
start = state.offset start = state.offset
end = min(start + _PICKER_WINDOW, n) end = min(start + _PICKER_WINDOW, n)
# 80 cols total: "#(5) + title(43) + Src(7) + Status(8) + Updated(17)".
table = Table( title_width = 43
box=box.SIMPLE_HEAD, header = (
show_header=True, f"Sessions in {chat_cwd} ({start + 1}-{end}/{n})"
header_style="dim", if n
expand=True, else f"Sessions in {chat_cwd}"
pad_edge=False,
collapse_padding=True,
) )
table.add_column("#", justify="right", no_wrap=True, style="dim", min_width=3) lines = [header]
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)
for i in range(start, end): for i in range(start, end):
row = rows[i] lines.append(
if row.jsonl_path is None: _picker_line(rows[i], i + 1, selected=(i - start == state.cursor), title_width=title_width)
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"),
)
if message: if message:
status = Text(message, style="yellow") status = f"\x1b[33m{message}\x1b[0m"
elif state.buffer: elif state.buffer:
status = Text.assemble( status = f"select #{state.buffer} · Enter confirm · ⌫ clear"
("select #", "cyan"),
(state.buffer, "bold cyan"),
(" · Enter confirm · ⌫ clear", "dim"),
)
else: else:
status = Text( status = "↑↓ move · Enter select · digits+Enter jump · q quit"
"↑↓ scroll · Enter select · digits+Enter jump · PgUp/PgDn page · q quit", lines.append(status)
style="dim", return lines
)
return Group(header, table, status)
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( def _pick_session(
rows: list[ClaudeSessionRow], chat_cwd: Path rows: list[ClaudeSessionRow], chat_cwd: Path
) -> ClaudeSessionRow | None: ) -> 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 termios
import tty import tty
@@ -993,27 +1035,25 @@ def _pick_session(
fd = sys.stdin.fileno() fd = sys.stdin.fileno()
old = termios.tcgetattr(fd) old = termios.tcgetattr(fd)
message: str | None = None message: str | None = None
prev_height = 0
try: try:
tty.setraw(fd) tty.setraw(fd)
live = Live( while True:
_render_picker_window(state, chat_cwd, message), lines = _picker_block(state, chat_cwd, message)
console=console, _draw_picker_block(lines, prev_height)
refresh_per_second=4, prev_height = len(lines)
transient=True, key = _read_key(fd)
) if key == "timeout":
with live: continue # idle: block is stable, no redraw needed
while True: state, action, payload = _picker_advance(state, key)
key = _read_key(fd) message = payload if isinstance(payload, str) else None
if key == "timeout": if action == "select":
continue # idle: no redraw; Live's low-rate refresh covers resize _clear_picker_block(prev_height)
state, action, payload = _picker_advance(state, key) assert isinstance(payload, ClaudeSessionRow)
message = payload if isinstance(payload, str) else None return payload
live.update(_render_picker_window(state, chat_cwd, message)) if action == "quit":
if action == "select": _clear_picker_block(prev_height)
assert isinstance(payload, ClaudeSessionRow) return None
return payload
if action == "quit":
return None
finally: finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old) 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: def _resume_picker_claude(chat_cwd: Path, use_tmux: bool = False) -> None:
"""Interactive unified resume picker for the claude backend. """Interactive unified resume picker for the claude backend.
Lists jsonl + xiaohe client sessions for chat_cwd in a 10-row scrollable Lists jsonl + xiaohe client sessions for chat_cwd in a fixed 10-row block
window; selecting one resumes the jsonl via ``claude --resume <id>``. (↑↓ navigate, digits+Enter jump, Enter select, q quit); selecting one
resumes the jsonl via ``claude --resume <id>``.
""" """
rows = _claude_session_rows(chat_cwd) rows = _claude_session_rows(chat_cwd)
if not rows: if not rows:
+41 -23
View File
@@ -11,6 +11,20 @@ import myagents.launcher
from myagents.cli import cli 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: class TestMyagentsHelp:
"""Tests for top-level myagents command.""" """Tests for top-level myagents command."""
@@ -74,7 +88,7 @@ class TestClaudeSubcommand:
def test_runs_claude(self) -> None: def test_runs_claude(self) -> None:
runner = CliRunner() runner = CliRunner()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -99,7 +113,7 @@ class TestClaudeSubcommand:
test_dir.mkdir() test_dir.mkdir()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -121,7 +135,7 @@ class TestClaudePassthrough:
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]: def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
runner = CliRunner() runner = CliRunner()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -132,13 +146,17 @@ class TestClaudePassthrough:
mock_run.assert_called_once() mock_run.assert_called_once()
return mock_run.call_args[0][0] return mock_run.call_args[0][0]
def test_resume_flag_passes_through(self, tmp_path: Path) -> None: def test_bare_resume_opens_picker_not_launch(self, tmp_path: Path) -> None:
cmd = self._invoke(["--resume"], tmp_path) """Bare --resume opens the unified picker; non-tty lists and exits."""
assert cmd == [ runner = CliRunner()
"/usr/bin/claude", with (
"--dangerously-skip-permissions", patch("myagents.launcher.shutil.which", side_effect=_backend_which("claude")),
"--resume", 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: def test_resume_with_session_id(self, tmp_path: Path) -> None:
cmd = self._invoke(["-r", "abc123"], tmp_path) cmd = self._invoke(["-r", "abc123"], tmp_path)
@@ -175,7 +193,7 @@ class TestKimiSubcommand:
def test_runs_kimi(self) -> None: def test_runs_kimi(self) -> None:
runner = CliRunner() runner = CliRunner()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -197,7 +215,7 @@ class TestKimiSubcommand:
test_dir.mkdir() test_dir.mkdir()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -212,7 +230,7 @@ class TestKimiPassthrough:
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]: def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
runner = CliRunner() runner = CliRunner()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -265,7 +283,7 @@ class TestCodexSubcommand:
def test_runs_codex(self) -> None: def test_runs_codex(self) -> None:
runner = CliRunner() runner = CliRunner()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -287,7 +305,7 @@ class TestCodexSubcommand:
test_dir.mkdir() test_dir.mkdir()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -302,7 +320,7 @@ class TestCodexPassthrough:
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]: def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
runner = CliRunner() runner = CliRunner()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -328,7 +346,7 @@ class TestCodexResumeOptions:
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]: def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
runner = CliRunner() runner = CliRunner()
with ( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
mock_run.return_value = MagicMock(returncode=0) mock_run.return_value = MagicMock(returncode=0)
@@ -437,7 +455,7 @@ class TestHermesSubcommand:
runner = CliRunner() runner = CliRunner()
with ( with (
patch( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
@@ -461,7 +479,7 @@ class TestHermesSubcommand:
with ( with (
patch( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
@@ -478,7 +496,7 @@ class TestHermesPassthrough:
runner = CliRunner() runner = CliRunner()
with ( with (
patch( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
@@ -616,7 +634,7 @@ class TestCursorSubcommand:
runner = CliRunner() runner = CliRunner()
with ( with (
patch( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
@@ -664,7 +682,7 @@ class TestCursorSubcommand:
with ( with (
patch( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
@@ -681,7 +699,7 @@ class TestCursorPassthrough:
runner = CliRunner() runner = CliRunner()
with ( with (
patch( 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, patch("myagents.launcher.subprocess.run") as mock_run,
): ):
+57
View File
@@ -189,6 +189,63 @@ def _make_rows(claude_env, n: int):
return L._claude_session_rows(claude_env.cwd) 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: class TestPickerAdvance:
"""Windowed picker state machine (no tty needed).""" """Windowed picker state machine (no tty needed)."""