From 969fc9ac275543ba8db317473a2e16735a616b2b Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Sat, 15 Aug 2026 15:44:33 +0800 Subject: [PATCH] =?UTF-8?q?myclaude=20-r:=20=E4=BA=A4=E4=BA=92=E5=BC=8F=20?= =?UTF-8?q?resume=20picker=EF=BC=88=E6=96=B9=E5=90=91=E9=94=AE/=E7=BF=BB?= =?UTF-8?q?=E9=A1=B5/=E5=8D=B3=E6=97=B6=E8=BF=87=E6=BB=A4=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- myagents/launcher.py | 246 ++++++++++++++++++++++++++++---- tests/test_launcher_sessions.py | 148 ++++++++++++++++--- 2 files changed, 342 insertions(+), 52 deletions(-) diff --git a/myagents/launcher.py b/myagents/launcher.py index 200ac8c..00c879b 100644 --- a/myagents/launcher.py +++ b/myagents/launcher.py @@ -4,6 +4,7 @@ import hashlib import json import os import re +import select import shlex import shutil import sqlite3 @@ -16,8 +17,8 @@ from pathlib import Path import click from rich import box -from rich.console import Console -from rich.prompt import Prompt +from rich.console import Console, Group +from rich.live import Live from rich.table import Table from rich.text import Text @@ -810,48 +811,233 @@ def _list_sessions_claude(chat_cwd: Path) -> None: ) +_PICKER_WINDOW = 10 + + +@dataclass +class _PickerState: + """Mutable state for the interactive resume picker.""" + + rows: list[ClaudeSessionRow] + offset: int = 0 + cursor: int = 0 + buffer: str = "" + + +def _read_key(fd: int, timeout: float = 0.3) -> str: + """Read one keypress from a raw tty fd; returns a symbolic key name. + + Handles single bytes (q, digits, Enter) and multi-byte escape sequences + (arrow keys, PageUp/PageDown). Returns "timeout" when idle. + """ + readable, _, _ = select.select([fd], [], [], timeout) + if not readable: + return "timeout" + try: + first = os.read(fd, 1) + except OSError: + return "quit" + if first == b"\x1b": + seq = first + for _ in range(8): + readable, _, _ = select.select([fd], [], [], 0.1) + if not readable: + break + seq += os.read(fd, 1) + return { + b"\x1b[A": "up", + b"\x1b[B": "down", + b"\x1b[C": "right", + b"\x1b[D": "left", + b"\x1b[5~": "pgup", + b"\x1b[6~": "pgdown", + }.get(seq, "esc") + if first in (b"\r", b"\n"): + return "enter" + if first in (b"q", b"Q"): + return "q" + if first == b"\x03": + return "ctrl-c" + if first in (b"\x7f", b"\x08"): + return "backspace" + if first.isdigit(): + return first.decode() + return "other" + + +def _picker_advance( + state: _PickerState, key: str +) -> tuple[_PickerState, str, ClaudeSessionRow | str | None]: + """Apply one key to the picker state. + + Returns ``(state, action, payload)``: action is ``select`` (payload is the + chosen row), ``quit`` (payload None) or ``none`` (payload is an optional + message line). + """ + rows = state.rows + n = len(rows) + width = _PICKER_WINDOW + if state.offset > max(0, n - width): + state.offset = max(0, n - width) + visible = min(width, n - state.offset) + if state.cursor >= visible: + state.cursor = visible - 1 + + if key in ("j", "k"): + key = "down" if key == "j" else "up" + if key == "down": + state.buffer = "" + if state.cursor + 1 < visible: + state.cursor += 1 + elif state.offset + width < n: + state.offset += 1 + elif key == "up": + state.buffer = "" + if state.cursor > 0: + state.cursor -= 1 + elif state.offset > 0: + state.offset -= 1 + elif key == "pgdown": + state.buffer = "" + state.offset = min(state.offset + width, max(0, n - width)) + elif key == "pgup": + state.buffer = "" + state.offset = max(0, state.offset - width) + elif key.isdigit() and len(state.buffer) < 6: + state.buffer += key + elif key == "backspace": + state.buffer = state.buffer[:-1] + elif key == "enter": + if state.buffer: + index = int(state.buffer) + if not 1 <= index <= n: + return state, "none", f"range 1-{n}" + row = rows[index - 1] + else: + row = rows[state.offset + state.cursor] + if row.jsonl_path is None: + return state, "none", "client-only: no jsonl — continue in xiaohe client" + return state, "select", row + elif key in ("q", "esc", "ctrl-c"): + return state, "quit", None + return state, "none", None + + +def _render_picker_window( + state: _PickerState, chat_cwd: Path, message: str | None +) -> Group: + """Live renderable for the windowed picker (10 rows + status line).""" + 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, + ) + 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) + 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"), + ) + if message: + status = Text(message, style="yellow") + elif state.buffer: + status = Text.assemble( + ("select #", "cyan"), + (state.buffer, "bold cyan"), + (" · Enter confirm · ⌫ clear", "dim"), + ) + else: + status = Text( + "↑↓ scroll · Enter select · digits+Enter jump · PgUp/PgDn page · q quit", + style="dim", + ) + return Group(header, table, status) + + +def _pick_session( + rows: list[ClaudeSessionRow], chat_cwd: Path +) -> ClaudeSessionRow | None: + """Run the interactive windowed picker on a raw tty; None on quit.""" + import termios + import tty + + state = _PickerState(rows=rows) + fd = sys.stdin.fileno() + old = termios.tcgetattr(fd) + message: str | None = None + 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 + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, old) + + 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; selecting one resumes - the jsonl via ``claude --resume ``. + Lists jsonl + xiaohe client sessions for chat_cwd in a 10-row scrollable + window; selecting one resumes the jsonl via ``claude --resume ``. """ rows = _claude_session_rows(chat_cwd) if not rows: stderr_console.print(f"[yellow]No sessions for[/yellow] {chat_cwd}") raise SystemExit(0) - _render_claude_table(rows, chat_cwd, numbered=True) if not sys.stdin.isatty(): + _render_claude_table(rows, chat_cwd, numbered=True) stderr_console.print( "[yellow]Not a terminal — resume directly with[/yellow] " "[green]myclaude -r [/green]" ) raise SystemExit(0) - selectable = [r for r in rows if r.jsonl_path is not None] - while True: - answer = Prompt.ask( - "[bold]Choose session[/bold] " - "(Enter = latest, [cyan]q[/cyan] = quit)", - default="", - ) - if answer.strip().lower() == "q": - raise SystemExit(0) - if answer.strip() == "": - row = selectable[0] - break - if not answer.strip().isdigit(): - continue - n = int(answer.strip()) - if not 1 <= n <= len(rows): - continue - row = rows[n - 1] - if row.jsonl_path is None: - stderr_console.print( - "[red]Client-only session[/red] — no jsonl to resume in the " - "CLI; continue it in the xiaohe web/desktop client instead." - ) - continue - break + row = _pick_session(rows, chat_cwd) + if row is None: + raise SystemExit(0) console.print( f"Resuming [cyan]{row.cli_session_id}[/cyan] ({row.title[:40]})…" ) diff --git a/tests/test_launcher_sessions.py b/tests/test_launcher_sessions.py index 6232e11..c123391 100644 --- a/tests/test_launcher_sessions.py +++ b/tests/test_launcher_sessions.py @@ -2,6 +2,7 @@ import copy import json +import os import re import sys from pathlib import Path @@ -150,32 +151,135 @@ class TestResumePickerClaude: assert "Not a terminal" in err assert "resume directly with myclaude -r " in err - def test_selecting_number_launches_resume( - self, claude_env, monkeypatch - ) -> None: + def test_routing_select_launches(self, claude_env, monkeypatch) -> None: + """Picking a session resumes via ``claude --resume ``.""" _write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z") - _write_jsonl(claude_env.proj_dir, "bbb", "2026-08-15T03:00:00Z") - launched: list[list[str]] = [] + monkeypatch.setattr(sys.stdin, "isatty", lambda: True) + monkeypatch.setattr(L, "_pick_session", lambda rows, cwd: rows[0]) monkeypatch.setattr( L, "_launch", lambda *_a, **_k: launched.append(_a[2]) ) - monkeypatch.setattr(sys.stdin, "isatty", lambda: True) - monkeypatch.setattr(L.Prompt, "ask", lambda *a, **k: "2") - - L._resume_picker_claude(claude_env.cwd) - assert launched == [["--resume", "bbb"]] - - def test_enter_picks_latest(self, claude_env, monkeypatch) -> None: - _write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z") - _write_jsonl(claude_env.proj_dir, "bbb", "2026-08-15T03:00:00Z") - - launched: list[list[str]] = [] - monkeypatch.setattr( - L, "_launch", lambda *_a, **_k: launched.append(_a[2]) - ) - monkeypatch.setattr(sys.stdin, "isatty", lambda: True) - monkeypatch.setattr(L.Prompt, "ask", lambda *a, **k: "") - L._resume_picker_claude(claude_env.cwd) assert launched == [["--resume", "aaa"]] + + def test_routing_quit_exits(self, claude_env, monkeypatch) -> None: + """Quitting the picker exits 0 without launching.""" + _write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z") + launched: list[list[str]] = [] + monkeypatch.setattr(sys.stdin, "isatty", lambda: True) + monkeypatch.setattr(L, "_pick_session", lambda rows, cwd: None) + monkeypatch.setattr( + L, "_launch", lambda *_a, **_k: launched.append(_a[2]) + ) + with pytest.raises(SystemExit) as exc: + L._resume_picker_claude(claude_env.cwd) + assert exc.value.code == 0 + assert launched == [] + + +def _make_rows(claude_env, n: int): + """n sessions, s000 newest → rows[0] == s000.""" + for i in range(n): + _write_jsonl( + claude_env.proj_dir, + f"s{i:03d}", + f"2026-08-15T{23 - i:02d}:00:00Z", + ) + return L._claude_session_rows(claude_env.cwd) + + +class TestPickerAdvance: + """Windowed picker state machine (no tty needed).""" + + def test_enter_selects_latest(self, claude_env) -> None: + rows = _make_rows(claude_env, 3) + state = L._PickerState(rows=rows) + state, action, payload = L._picker_advance(state, "enter") + assert action == "select" + assert payload == rows[0] + + def test_number_selects_row(self, claude_env) -> None: + rows = _make_rows(claude_env, 3) + state = L._PickerState(rows=rows) + state, action, payload = L._picker_advance(state, "2") + assert action == "none" + state, action, payload = L._picker_advance(state, "enter") + assert action == "select" + assert payload == rows[1] + + def test_number_out_of_range(self, claude_env) -> None: + rows = _make_rows(claude_env, 3) + state = L._PickerState(rows=rows) + state, _, _ = L._picker_advance(state, "9") + state, action, payload = L._picker_advance(state, "enter") + assert action == "none" + assert isinstance(payload, str) and "range" in payload + + def test_scroll_window_advances(self, claude_env) -> None: + rows = _make_rows(claude_env, 15) + state = L._PickerState(rows=rows) + for _ in range(10): + state, action, payload = L._picker_advance(state, "down") + assert action == "none" + # cursor hits window bottom, then the window scrolls + assert state.offset == 1 + assert state.cursor == 9 + + def test_quit(self, claude_env) -> None: + rows = _make_rows(claude_env, 3) + state = L._PickerState(rows=rows) + state, action, payload = L._picker_advance(state, "q") + assert action == "quit" + assert payload is None + + def test_db_only_row_not_selectable(self, claude_env) -> None: + _write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z") + # hosted session with no jsonl on disk, dated newest (year 2033) + claude_env.index["nojsonl"] = XiaoheSession( + cli_session_id="nojsonl", + xiaohe_session_id="x", + title="Lost", + status="ready", + updated_at=2_000_000_000.0, + ) + rows = L._claude_session_rows(claude_env.cwd) + assert rows[0].jsonl_path is None + state = L._PickerState(rows=rows) + state, action, payload = L._picker_advance(state, "enter") + assert action == "none" + assert isinstance(payload, str) and "client-only" in payload + + +def _read_key_from_bytes(data: bytes) -> str: + """Feed bytes through a pipe into ``_read_key``.""" + r, w = os.pipe() + try: + os.write(w, data) + return L._read_key(r) + finally: + os.close(r) + os.close(w) + + +class TestReadKey: + def test_enter(self) -> None: + assert _read_key_from_bytes(b"\r") == "enter" + + def test_arrow_down(self) -> None: + assert _read_key_from_bytes(b"\x1b[B") == "down" + + def test_arrow_up(self) -> None: + assert _read_key_from_bytes(b"\x1b[A") == "up" + + def test_page_down(self) -> None: + assert _read_key_from_bytes(b"\x1b[6~") == "pgdown" + + def test_q(self) -> None: + assert _read_key_from_bytes(b"q") == "q" + + def test_digit(self) -> None: + assert _read_key_from_bytes(b"5") == "5" + + def test_escape(self) -> None: + assert _read_key_from_bytes(b"\x1b") == "esc"