"""Shared launcher logic for AI coding agent CLIs.""" import hashlib import json import os import platform import re import select import shlex import shutil import sqlite3 import subprocess import sys import unicodedata from dataclasses import dataclass from datetime import datetime from importlib.metadata import PackageNotFoundError, version from pathlib import Path import click from rich import box from rich.console import Console from rich.table import Table from rich.text import Text from myagents.project_root import get_workspace_root from myagents.xiaohe_sessions import session_index, workspace_ids_for_cwd stderr_console = Console(stderr=True) console = Console() _BACKENDS: dict[str, dict] = { "claude": { "binary": "claude", "env_bin": "CLAUDE_BIN", "sessions_root": lambda: Path.home() / ".claude" / "projects", "session_pattern": "*.jsonl", "default_args": ["--dangerously-skip-permissions"], "install_cmd": ["npm", "install", "-g", "@anthropic-ai/claude-code"], "npm_package": "@anthropic-ai/claude-code", "native_min_bytes": 1_000_000, "native_bin_rel": ("bin/claude.exe", "bin/claude"), "native_platform_packages": { "win32-x64": "@anthropic-ai/claude-code-win32-x64", "win32-arm64": "@anthropic-ai/claude-code-win32-arm64", "darwin-arm64": "@anthropic-ai/claude-code-darwin-arm64", "darwin-x64": "@anthropic-ai/claude-code-darwin-x64", "linux-x64": "@anthropic-ai/claude-code-linux-x64", "linux-arm64": "@anthropic-ai/claude-code-linux-arm64", }, "not_found_msg": ( "[red]claude CLI not found in PATH.[/red] Install Claude Code or set " "[cyan]CLAUDE_BIN[/cyan]." ), }, "kimi": { "binary": "kimi", "env_bin": "KIMI_BIN", "sessions_root": lambda: Path.home() / ".kimi" / "sessions", "session_pattern": "*", "default_args": [], "not_found_msg": ( "[red]kimi CLI not found in PATH.[/red] Install Kimi Code CLI or set " "[cyan]KIMI_BIN[/cyan]." ), }, "codex": { "binary": "codex", "env_bin": "CODEX_BIN", "sessions_root": lambda: Path.home() / ".codex" / "sessions", "session_pattern": "**/*.jsonl", "default_args": ["--dangerously-bypass-approvals-and-sandbox"], "install_cmd": ["npm", "install", "-g", "@openai/codex"], "npm_package": "@openai/codex", "native_min_bytes": 1_000_000, "native_bin_rel": ( "codex.exe", "bin/codex.exe", "bin/codex", "vendor/codex.exe", ), "native_platform_packages": { "win32-x64": "@openai/codex-win32-x64", "win32-arm64": "@openai/codex-win32-arm64", "darwin-arm64": "@openai/codex-darwin-arm64", "darwin-x64": "@openai/codex-darwin-x64", "linux-x64": "@openai/codex-linux-x64", "linux-arm64": "@openai/codex-linux-arm64", }, "not_found_msg": ( "[red]codex CLI not found in PATH.[/red] Install with " "[cyan]npm install -g @openai/codex[/cyan] or set [cyan]CODEX_BIN[/cyan]." ), }, "hermes": { "binary": "hermes", "env_bin": "HERMES_BIN", # Hermes keeps sessions in a global SQLite store, not per-cwd files. "state_db": lambda: Path.home() / ".hermes" / "state.db", "default_args": ["--yolo"], "not_found_msg": ( "[red]hermes CLI not found in PATH.[/red] Install via the official " "Hermes app or [cyan]hermes-agent.nousresearch.com/install.sh[/cyan], " "or set [cyan]HERMES_BIN[/cyan]." ), }, "cursor": { # Cursor Agent CLI ships as ``agent`` (also linked as ``cursor-agent``). "binary": "agent", "alt_binaries": ("cursor-agent",), "env_bin": "CURSOR_BIN", "sessions_root": lambda: Path.home() / ".cursor" / "chats", "session_pattern": "*", "default_args": ["--force"], "not_found_msg": ( "[red]Cursor Agent CLI (agent) not found in PATH.[/red] Install " "from [cyan]https://cursor.com/docs/cli[/cyan] or set " "[cyan]CURSOR_BIN[/cyan]." ), }, "dsh": { # DeepSeek Harness: map myclaude-style ``-r`` to ``--resume``. "binary": "dsh", "env_bin": "DSH_BIN", "sessions_root": lambda: Path( os.environ.get("DSH_HOME") or (Path.home() / ".local/share/dsh") ).expanduser() / "sessions", "session_pattern": "*/*/session.jsonl.zstd", "default_args": [], "install_cmd": ["npm", "install", "-g", "@deepseek-ai/dsh"], "npm_package": "@deepseek-ai/dsh", "not_found_msg": ( "[red]dsh CLI not found in PATH.[/red] Install DeepSeek Harness " "with [cyan]npm install -g @deepseek-ai/dsh[/cyan] or set " "[cyan]DSH_BIN[/cyan]." ), }, } _NPM_OFFICIAL_REGISTRY = "https://registry.npmjs.org/" _NPM_INSTALL_FLAGS = ( "--include=optional", "--foreground-scripts", f"--registry={_NPM_OFFICIAL_REGISTRY}", ) def _resolve_chat_cwd(cwd: str | None) -> Path: """Working directory: --cwd if given, else workspace/.""" if cwd is None: return get_workspace_root() chat_cwd = Path(cwd).resolve() if not chat_cwd.is_dir(): stderr_console.print(f"[red]Not a directory:[/red] {cwd}") raise SystemExit(1) return chat_cwd def _translate_codex_extra(extra: list[str]) -> list[str]: """Map myagents-style resume options to native codex subcommands. Codex CLI uses ``resume`` / ``continue`` subcommands rather than flags. This keeps ``mycodex`` UX consistent with ``myclaude`` / ``mykimi``. """ if not extra: return extra head = extra[0] tail = extra[1:] if head in ("-r", "--resume"): if not tail: return ["resume"] return ["resume", tail[0], *tail[1:]] if head == "--continue": return ["continue", *tail] if head == "--last": return ["resume", "--last", *tail] if head == "--all": return ["resume", "--all", *tail] return extra def _translate_hermes_extra(extra: list[str]) -> list[str]: """Map myagents-style resume options to native hermes syntax. Hermes ``-r``/``--resume`` requires a session id (no bare picker), has no ``-c`` shorthand, and restores the session's original cwd by default. This keeps ``myhermes`` UX consistent with ``myclaude``: bare ``-r`` opens the interactive picker, ``-c`` continues the latest session, and resumed sessions stay in the launch cwd. """ if not extra: return extra head = extra[0] tail = extra[1:] if head in ("-r", "--resume"): if not tail or tail[0].startswith("-"): return ["sessions", "browse", *tail] return ["--resume", tail[0], "--no-restore-cwd", *tail[1:]] if head in ("-c", "--continue", "--last"): return ["--continue", *tail, "--no-restore-cwd"] return extra def _translate_cursor_extra(extra: list[str]) -> list[str]: """Map myagents-style resume shortcuts to Cursor Agent CLI flags. Cursor Agent accepts ``--resume [chatId]`` and ``--continue``, but not ``-r`` / ``-c`` / ``--last``. Keep ``mycursor`` UX aligned with the others. """ if not extra: return extra head = extra[0] tail = extra[1:] if head == "-r": if not tail or tail[0].startswith("-"): return ["--resume", *tail] return ["--resume", tail[0], *tail[1:]] if head in ("-c", "--last"): return ["--continue", *tail] return extra def _translate_dsh_extra(extra: list[str]) -> list[str]: """Map myclaude-style ``-r`` onto dsh ``--resume``; leave the rest as-is.""" if not extra: return extra head = extra[0] tail = extra[1:] if head == "-r": if not tail or tail[0].startswith("-"): return ["--resume", *tail] return ["--resume", tail[0], *tail[1:]] return extra def agent_entrypoints() -> dict[str, str]: """Map ``myclaude`` / ``mydsh`` / … to backend names.""" return {f"my{name}": name for name in _BACKENDS} def _resolve_backend_binary(config: dict) -> str | None: """Resolve backend CLI: env override → PATH → npm global prefix.""" env = (os.environ.get(config["env_bin"]) or "").strip() if env: return env binary = shutil.which(config["binary"]) if not binary: for alt in config.get("alt_binaries", ()): found = shutil.which(alt) if found: binary = found break if not binary: binary = _resolve_from_npm_global(config) if not binary: return None if not _native_binary_ok(config): return None return binary def _npm_global_bin() -> Path | None: """Directory where ``npm install -g`` places binaries, if resolvable.""" npm = shutil.which("npm") if not npm: return None try: proc = subprocess.run( [npm, "prefix", "-g"], capture_output=True, text=True, timeout=30, check=False, ) except (OSError, subprocess.TimeoutExpired): return None prefix = (proc.stdout or "").strip() if proc.returncode != 0 or not prefix: return None root = Path(prefix) # Unix: /bin; Windows npm usually puts shims in itself. if sys.platform == "win32": return root return root / "bin" def _npm_global_modules() -> Path | None: """``node_modules`` root for ``npm install -g``, if present.""" npm_bin = _npm_global_bin() if not npm_bin: return None if sys.platform == "win32": cand = npm_bin / "node_modules" else: cand = npm_bin.parent / "lib" / "node_modules" if not cand.is_dir(): alt = npm_bin.parent / "node_modules" cand = alt if alt.is_dir() else cand return cand if cand.is_dir() else None def _npm_platform_key() -> str: machine = platform.machine().lower() if machine in ("x86_64", "amd64"): cpu = "x64" elif machine in ("aarch64", "arm64"): cpu = "arm64" else: cpu = machine if sys.platform == "win32": os_name = "win32" elif sys.platform == "darwin": os_name = "darwin" else: os_name = "linux" return f"{os_name}-{cpu}" def _native_magic_ok(path: Path) -> bool: """True when ``path`` looks like a native binary for this OS.""" try: with path.open("rb") as fh: magic = fh.read(4) except OSError: return False if len(magic) < 2: return False if sys.platform == "win32": return magic[:2] == b"MZ" if sys.platform == "darwin": return magic in ( b"\xcf\xfa\xed\xfe", b"\xfe\xed\xfa\xcf", b"\xca\xfe\xba\xbe", b"\xbe\xba\xfe\xca", b"\xce\xfa\xed\xfe", b"\xfe\xed\xfa\xce", ) return magic[:4] == b"\x7fELF" def _looks_like_native_bin(path: Path, *, min_bytes: int) -> bool: try: if path.stat().st_size < min_bytes: return False except OSError: return False return _native_magic_ok(path) def _native_candidate_paths(config: dict) -> list[Path]: rels = tuple(config.get("native_bin_rel") or ()) if not rels: return [] modules = _npm_global_modules() if modules is None: return [] out: list[Path] = [] plat_pkgs = config.get("native_platform_packages") or {} plat_pkg = plat_pkgs.get(_npm_platform_key()) names = (config.get("binary"), *config.get("alt_binaries", ())) if plat_pkg: pkg_dir = modules / plat_pkg for rel in rels: out.append(pkg_dir / rel) for name in names: if not name: continue out.append(pkg_dir / name) if sys.platform == "win32": out.append(pkg_dir / f"{name}.exe") main_pkg = config.get("npm_package") if main_pkg: pkg_dir = modules / main_pkg for rel in rels: out.append(pkg_dir / rel) return out def _native_binary_ok(config: dict) -> bool: """False when a known native package is present but is a stub/wrong OS. Unknown layouts (no candidate files) are treated as OK so PATH-only installs are not blocked. """ min_bytes = int(config.get("native_min_bytes") or 0) if min_bytes <= 0: return True found = [p for p in _native_candidate_paths(config) if p.is_file()] if not found: return True return any(_looks_like_native_bin(p, min_bytes=min_bytes) for p in found) def _resolve_from_npm_global(config: dict) -> str | None: """Look up binary under ``npm prefix -g`` (even when that dir is not on PATH).""" names = (config["binary"], *config.get("alt_binaries", ())) npm_bin = _npm_global_bin() if not npm_bin or not npm_bin.is_dir(): return None for name in names: candidates = [npm_bin / name] if sys.platform == "win32": candidates.extend( [npm_bin / f"{name}.cmd", npm_bin / f"{name}.exe"] ) for cand in candidates: if cand.is_file(): return str(cand) return None def _is_interactive() -> bool: """True when we can safely prompt the user (TTY, not CI).""" if os.environ.get("CI", "").lower() in ("1", "true", "yes"): return False if os.environ.get("XIAOHE_ASSUME_YES", "").lower() in ("1", "true", "yes"): return False try: return bool(sys.stdin.isatty() and sys.stderr.isatty()) except Exception: # noqa: BLE001 — treat broken streams as non-interactive return False _NPM_BIN_ORIGINALLY_ON_PATH: dict[str, bool] = {} def _npm_bin_on_path(npm_bin: Path) -> bool: """True if ``npm_bin`` already appears as a PATH entry.""" want = str(npm_bin.resolve()) if npm_bin.exists() else str(npm_bin) for part in os.environ.get("PATH", "").split(os.pathsep): if not part: continue try: if str(Path(part).resolve()) == want: return True except OSError: if part.rstrip("/\\") == str(npm_bin).rstrip("/\\"): return True return False def ensure_npm_bin_on_path(*, persist: bool | None = None) -> Path | None: """Prepend npm global bin to ``PATH`` for this process. When ``persist`` is True, or None on an interactive TTY when the bin was not on PATH at first sight, offer to append an export line to the shell rc so future shells find ``claude`` via ``which``. Returns the npm bin directory when known. """ npm_bin = _npm_global_bin() if not npm_bin: return None key = str(npm_bin) if key not in _NPM_BIN_ORIGINALLY_ON_PATH: _NPM_BIN_ORIGINALLY_ON_PATH[key] = _npm_bin_on_path(npm_bin) originally = _NPM_BIN_ORIGINALLY_ON_PATH[key] if not _npm_bin_on_path(npm_bin): os.environ["PATH"] = str(npm_bin) + os.pathsep + os.environ.get( "PATH", "" ) # Quiet when silently refreshing PATH for this process (persist=False). if persist is not False: stderr_console.print( f"[dim]Prepended to PATH (this process):[/dim] {npm_bin}" ) do_persist = persist if do_persist is None: do_persist = (not originally) and _is_interactive() if do_persist and not originally: _offer_persist_npm_bin(npm_bin) return npm_bin def _shell_rc_path() -> Path: shell = os.environ.get("SHELL") or "" if "zsh" in shell: return Path.home() / ".zshrc" if "fish" in shell: return Path.home() / ".config" / "fish" / "config.fish" return Path.home() / ".bashrc" def _offer_persist_npm_bin(npm_bin: Path) -> None: """Ask to add npm global bin to the shell rc (guided, default Yes).""" rc = _shell_rc_path() if "fish" in str(rc): line = f'set -gx PATH "{npm_bin}" $PATH # myagents / npm global' else: line = f'export PATH="{npm_bin}:$PATH" # myagents / npm global' if rc.is_file(): try: if line.split("#")[0].strip() in rc.read_text(encoding="utf-8"): return # Also skip if the directory is already exported somehow. if str(npm_bin) in rc.read_text(encoding="utf-8"): return except OSError: pass stderr_console.print( f"[yellow]{npm_bin}[/yellow] is not on your login PATH.\n" f" Add it to [cyan]{rc}[/cyan] so new terminals find " f"[cyan]claude[/cyan] / [cyan]dsh[/cyan] / [cyan]codex[/cyan]?" ) try: answer = click.prompt( "Add to shell rc? [Y/n]", default="y", show_default=False, err=True ) except (click.Abort, EOFError): return if answer.strip().lower() not in ("y", "yes", ""): stderr_console.print( f"[dim]Skipped.[/dim] Later, add:\n {line}" ) return rc.parent.mkdir(parents=True, exist_ok=True) with rc.open("a", encoding="utf-8") as fh: fh.write(f"\n{line}\n") stderr_console.print( f"[green]Added[/green] to {rc} — open a new terminal " f"(or [cyan]source {rc}[/cyan])." ) def _resolve_installed_binary(config: dict) -> str | None: """Resolve binary after install (PATH + npm global).""" ensure_npm_bin_on_path(persist=False) return _resolve_backend_binary(config) def _tmux_session_name(backend: str, chat_cwd: Path) -> str: """Deterministic per-backend, per-directory tmux session name. Re-running ``myclaude --tmux`` in the same directory reattaches to the same session; the digest keeps same-named directories from colliding. """ slug = re.sub(r"[^A-Za-z0-9_-]", "-", chat_cwd.name).strip("-") or "root" digest = hashlib.md5(str(chat_cwd).encode("utf-8")).hexdigest()[:6] # noqa: S324 return f"my{backend}-{slug}-{digest}" def _exec_tmux(backend: str, chat_cwd: Path, cmd: list[str]) -> None: """Run cmd inside an attachable tmux session. Never returns.""" if os.environ.get("TMUX"): stderr_console.print( "[yellow]Already inside tmux; running without a nested session.[/yellow]" ) proc = subprocess.run(cmd, cwd=str(chat_cwd), check=False) raise SystemExit(proc.returncode) tmux = shutil.which("tmux") if not tmux: stderr_console.print( "[red]tmux not found in PATH.[/red] Install it (e.g. " "[cyan]sudo apt install tmux[/cyan] / [cyan]brew install tmux[/cyan]) " "or drop [cyan]--tmux[/cyan]." ) raise SystemExit(127) name = _tmux_session_name(backend, chat_cwd) stderr_console.print( f"[dim]tmux session[/dim] [cyan]{name}[/cyan] " "[dim](detach: C-b d, reattach: same command)[/dim]" ) proc = subprocess.run( [ tmux, "new-session", "-A", "-s", name, "-c", str(chat_cwd), shlex.join(cmd), ], check=False, ) raise SystemExit(proc.returncode) def _ensure_agent_hint() -> str: """Prefer product-facing xiaohe when installed; else myagents.""" return ( "xiaohe ensure-agent" if shutil.which("xiaohe") else "myagents ensure-agent" ) def _npm_install_argv(install_cmd: list[str]) -> list[str]: """``npm install -g `` plus flags so platform optional deps resolve.""" argv = list(install_cmd) if argv and argv[0] == "npm": argv.extend(_NPM_INSTALL_FLAGS) return argv def _run_install(config: dict) -> str | None: """Run ``install_cmd`` and return the resolved binary path, or None.""" install_cmd = config.get("install_cmd") if not install_cmd: return None argv = _npm_install_argv(list(install_cmd)) installer = shutil.which(argv[0]) if not installer: hint = _ensure_agent_hint() stderr_console.print( f"[red]{argv[0]} not found[/red] — cannot auto-install " f"{config['binary']}.\n" " Run: [cyan]brew install node[/cyan] (or install Node.js), then " f"[cyan]{hint}[/cyan]." ) return None stderr_console.print( f"[dim]Installing {config['binary']} via[/dim] " f"[cyan]{shlex.join(argv)}[/cyan]" ) stderr_console.print( "[dim] (npm global install — may take a minute)[/dim]" ) proc = subprocess.run([installer, *argv[1:]], check=False) if proc.returncode != 0: hint = _ensure_agent_hint() stderr_console.print( f"[red]Install failed (exit {proc.returncode}).[/red]\n" f" Run: [cyan]{hint}[/cyan]." ) return None # Refresh PATH for this process; offer to persist npm bin into shell rc. ensure_npm_bin_on_path(persist=None) resolved = _resolve_backend_binary(config) if resolved: return resolved pkg = str(config.get("npm_package") or config["binary"]) stderr_console.print( "[red]Installed CLI is not a valid native binary for this OS[/red] " "(npm mirrors often ship a 1KB stub or the wrong platform).\n" f" [cyan]npm uninstall -g {pkg}[/cyan]\n" f" [cyan]npm install -g {pkg} --registry={_NPM_OFFICIAL_REGISTRY} " "--include=optional --foreground-scripts[/cyan]" ) return None def _offer_install(config: dict, *, yes: bool | None = None) -> str | None: """Install a missing backend CLI (guided prompt, or auto when non-interactive). Only backends with a known one-shot installer (``install_cmd``) are offered; others fall back to the manual ``not_found_msg``. ``yes``: True = install without asking; False = always prompt when possible; None = auto (no prompt when non-interactive / CI). Interactive default is Yes when the installer binary is on PATH. """ install_cmd = config.get("install_cmd") if not install_cmd: return None auto = bool(yes) if yes is not None else not _is_interactive() if not auto: stderr_console.print( f"[yellow]{config['binary']} CLI is missing[/yellow] " "(needed to launch this agent).\n" f" Will run: [cyan]{shlex.join(_npm_install_argv(list(install_cmd)))}[/cyan]" ) default = "y" if shutil.which(install_cmd[0]) else "n" prompt = ( "Install now? [Y/n]" if default == "y" else "Install now? [y/N]" ) try: answer = click.prompt( prompt, default=default, show_default=False, err=True ) except (click.Abort, EOFError): return None if answer.strip().lower() not in ("y", "yes", ""): hint = _ensure_agent_hint() stderr_console.print( f"[dim]Skipped. Later:[/dim] [cyan]{hint}[/cyan]" ) return None return _run_install(config) def resolve_backend_binary(backend: str) -> str | None: """Resolve absolute path to a backend CLI (env → PATH → npm global).""" cfg = _BACKENDS.get(backend) if not cfg: return None ensure_npm_bin_on_path(persist=False) return _resolve_backend_binary(cfg) def backend_available(backend: str) -> bool: """True if the backend CLI resolves (PATH, env override, or npm global).""" return resolve_backend_binary(backend) is not None def ensure_backend(backend: str, *, yes: bool | None = None) -> bool: """Ensure a backend CLI is available, installing it when missing. Guided by default: prompts on a TTY (default Yes when npm is available). Non-interactive contexts (install scripts, CI, piped stdin) install automatically when an ``install_cmd`` exists. Pass ``yes=True`` only for explicit scripting. Backends without an installer (e.g. hermes) cannot be auto-installed. Returns True if the CLI is available afterwards. """ ensure_npm_bin_on_path(persist=False) cfg = _BACKENDS.get(backend) if not cfg: return False if _resolve_backend_binary(cfg): return True return _offer_install(cfg, yes=yes) is not None def _launch( backend: str, chat_cwd: Path, extra: list[str], use_tmux: bool = False, offer_install: bool = False, ) -> None: """Run backend CLI in chat_cwd, forwarding extra args. Never returns.""" ensure_npm_bin_on_path(persist=False) config = _BACKENDS[backend] binary = _resolve_backend_binary(config) if not binary and offer_install: binary = _offer_install(config, yes=False) if not binary: stderr_console.print(config["not_found_msg"]) raise SystemExit(127) if backend == "codex": extra = _translate_codex_extra(extra) elif backend == "hermes": extra = _translate_hermes_extra(extra) elif backend == "cursor": extra = _translate_cursor_extra(extra) elif backend == "dsh": extra = _translate_dsh_extra(extra) cmd = [binary, *config["default_args"], *extra] if use_tmux: _exec_tmux(backend, chat_cwd, cmd) proc = subprocess.run(cmd, cwd=str(chat_cwd), check=False) raise SystemExit(proc.returncode) _SESSION_ID_UUID_RE = re.compile( r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.IGNORECASE, ) def _sessions_dir(backend: str, chat_cwd: Path) -> Path: """Directory where the backend stores sessions for chat_cwd.""" config = _BACKENDS[backend] cwd_str = str(chat_cwd) if backend in ("kimi", "cursor"): # Kimi / Cursor Agent hash the cwd with md5. munged = hashlib.md5(cwd_str.encode("utf-8")).hexdigest() # noqa: S324 elif backend in ("codex", "dsh"): # Codex stores all sessions under a single dated tree; dsh stores them # under // (cwd is in each session header). return config["sessions_root"]() else: munged = re.sub(r"[^A-Za-z0-9]", "-", cwd_str) return config["sessions_root"]() / munged # Claude Code local-command protocol rows (not human chat); skip when picking # the first-prompt fallback title. _LOCAL_COMMAND_NOISE_RE = re.compile( r"^\s*<(?:local-command-caveat|local-command-stdout|command-name)\b", re.IGNORECASE, ) # Only the head of each jsonl is scanned for a title: aiTitle/customTitle are # written near the start of a session and the first user prompt is the first # user row, so a bounded read covers the overwhelming majority of sessions # without paying a full-file parse per `-l` row. _TITLE_SCAN_LINES = 512 # Title cache keyed by (path, mtime_ns, size); a `-l` call can scan hundreds of # jsonl files, and unchanged files must not be re-parsed between polls. _TITLE_CACHE_MAX = 2000 _TITLE_CACHE: dict[tuple[str, int, int], str] = {} def _claude_user_text(entry: dict) -> str: """Plain-text of a claude user entry (text blocks joined).""" content = entry.get("message", {}).get("content") if isinstance(content, list): return " ".join( block.get("text", "") for block in content if isinstance(block, dict) and block.get("type") == "text" ) return content if isinstance(content, str) else "" def _session_title_claude(session_file: Path) -> str: """Claude jsonl title matching Claude Code SDK priority. customTitle (Ctrl+R) > aiTitle (auto-generated) > first real user prompt. Same source of truth the xiaohe client sidebar uses, so `myclaude -l` and the UI show the same label for the same session. """ try: st = session_file.stat() except OSError: return "" key = (str(session_file), st.st_mtime_ns, st.st_size) if key in _TITLE_CACHE: return _TITLE_CACHE[key] custom_title = ai_title = first_prompt = "" first_locked = False try: with session_file.open(encoding="utf-8", errors="ignore") as fh: for _ in range(_TITLE_SCAN_LINES): line = fh.readline() if not line: break try: entry = json.loads(line) except json.JSONDecodeError: continue ct = entry.get("customTitle") if isinstance(ct, str) and ct: custom_title = ct at = entry.get("aiTitle") if isinstance(at, str) and at: ai_title = at if not first_locked and entry.get("type") == "user": if ( entry.get("isMeta") is True or entry.get("isCompactSummary") is True ): continue text = _claude_user_text(entry).strip() if text and not _LOCAL_COMMAND_NOISE_RE.match(text): first_prompt = " ".join(text.split())[:200] first_locked = True except OSError: return "" result = custom_title or ai_title or first_prompt if len(_TITLE_CACHE) >= _TITLE_CACHE_MAX: _TITLE_CACHE.clear() _TITLE_CACHE[key] = result return result # Claude jsonl tail read for real-activity time (bounded scan keeps list cheap). _TAIL_SCAN_LINES = 256 def _iso_to_epoch(value: str) -> float | None: """Parse a Claude ``timestamp`` (ISO-8601, may end in Z) to epoch seconds.""" try: dt = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None return dt.timestamp() def _jsonl_content_updated_at(session_file: Path) -> float: """Newest message timestamp in a Claude jsonl (content truth, not mtime). The xiaohe client re-touches jsonl files during projection, so file mtime overstates recency; the last message timestamp is the real activity time. """ lines: list[str] = [] try: with session_file.open(encoding="utf-8", errors="ignore") as fh: for line in fh: lines.append(line) if len(lines) > _TAIL_SCAN_LINES: lines.pop(0) except OSError: return 0.0 newest = 0.0 for line in lines: try: entry = json.loads(line) except json.JSONDecodeError: continue ts = entry.get("timestamp") if not ts: msg = entry.get("message") if isinstance(msg, dict): ts = msg.get("timestamp") if isinstance(ts, str) and ts: parsed = _iso_to_epoch(ts) if parsed and parsed > newest: newest = parsed return newest @dataclass(frozen=True) class ClaudeSessionRow: """One resumable Claude session: jsonl (source of truth) + xiaohe index.""" cli_session_id: str title: str origin: str # "xiaohe" (client) or "cli" (plain Claude Code) status: str | None updated_at: float # real activity: max(content ts, db ts, mtime) jsonl_path: Path | None # None → client-only, not CLI-resumable xiaohe_session_id: str | None def _claude_session_rows(chat_cwd: Path) -> list[ClaudeSessionRow]: """Union of jsonl + xiaohe hosted sessions for chat_cwd, dedup by cli id. jsonl files are the source of truth; the xiaohe hosted-session index adds the client origin, status, and its own activity timestamp. Sessions known only to the index (jsonl deleted/absent) are kept as client-only rows. """ index = session_index(workspace_ids_for_cwd(chat_cwd)) rows: list[ClaudeSessionRow] = [] seen: set[str] = set() for session_file in _session_files("claude", chat_cwd): cid = session_file.stem seen.add(cid) xh = index.get(cid) db_ts = xh.updated_at if xh else None content_ts = _jsonl_content_updated_at(session_file) # Content/db timestamps are the real activity; file mtime is only a # fallback because the xiaohe projection re-touches jsonl files. activity = [t for t in (content_ts, db_ts) if t and t > 0] updated_at = max(activity) if activity else session_file.stat().st_mtime rows.append( ClaudeSessionRow( cli_session_id=cid, title=_session_title_claude(session_file), origin="xiaohe" if xh else "cli", status=xh.status if xh else None, updated_at=updated_at, jsonl_path=session_file, xiaohe_session_id=xh.xiaohe_session_id if xh else None, ) ) for cid, xh in index.items(): if cid in seen: continue rows.append( ClaudeSessionRow( cli_session_id=cid, title=xh.title or "(client-only)", origin="xiaohe", status=xh.status, updated_at=xh.updated_at or 0.0, jsonl_path=None, xiaohe_session_id=xh.xiaohe_session_id, ) ) rows.sort(key=lambda r: r.updated_at, reverse=True) return rows def _render_claude_table( rows: list[ClaudeSessionRow], chat_cwd: Path, numbered: bool ) -> None: """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, show_header=True, header_style="dim", expand=True, pad_edge=False, collapse_padding=True, ) if numbered: 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) # Picker selects by number; the full id column is only useful for `-r `. if not numbered: table.add_column("ID", overflow="ellipsis", no_wrap=True, style="cyan", min_width=36) for i, row in enumerate(rows, 1): cells: list[Text] = [] if numbered: cells.append(Text(str(i))) title_style = "dim" if not row.title else "" if row.jsonl_path is None: title_style = "dim italic" cells.append(Text(row.title or "(empty)", style=title_style)) src_style = "magenta" if row.origin == "xiaohe" else "dim" cells.append(Text(row.origin, style=src_style)) cells.append(Text(row.status or "")) cells.append(Text(datetime.fromtimestamp(row.updated_at).strftime("%Y-%m-%d %H:%M"))) if not numbered: cells.append(Text(row.cli_session_id)) table.add_row(*cells) console.print(table) def _list_sessions_claude(chat_cwd: Path) -> None: """List unified jsonl + xiaohe sessions for the claude backend.""" rows = _claude_session_rows(chat_cwd) if not rows: stderr_console.print(f"[yellow]No sessions for[/yellow] {chat_cwd}") return _render_claude_table(rows, chat_cwd, numbered=False) console.print( "\n[dim]Resume with[/dim] [green]myclaude -r[/green] [dim](picker), " "[green]myclaude -r [/green] [dim](direct), " "[green]myclaude -c[/green] [dim](continue last).[/dim]" ) _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. Escape sequences are consumed greedily but only up to the terminating byte (``[A``/``[B``… or ``[5~``/``[6~``) — any further bytes belong to the next keypress and are left in the buffer. This keeps rapid arrow-key repeats from being merged into one unknown sequence (which would read as ESC and quit the picker). """ readable, _, _ = select.select([fd], [], [], timeout) if not readable: return "timeout" try: first = os.read(fd, 1) except OSError: return "quit" if first == b"\x1b": return _read_escape_seq(fd) 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 _read_escape_seq(fd: int) -> str: """Read one CSI escape sequence (``ESC [`` …), returning a symbolic key. Reads only the bytes of one sequence: arrow keys are ``ESC [ A-D`` (3 bytes), PgUp/PgDn are ``ESC [ 5~`` / ``ESC [ 6~`` (4 bytes). It never reads past the terminator, so rapid back-to-back arrow keys stay distinct and don't merge into one unknown sequence (which would read as ESC). """ bracket = _read_byte_or_none(fd, 0.05) if bracket != b"[": return "esc" # bare ESC or unknown prefix nxt = _read_byte_or_none(fd, 0.05) if nxt is None: return "esc" if nxt in (b"A", b"B", b"C", b"D"): return {"A": "up", "B": "down", "C": "right", "D": "left"}[nxt.decode()] if nxt in (b"5", b"6"): term = _read_byte_or_none(fd, 0.05) if term == b"~": return "pgup" if nxt == b"5" else "pgdown" return "esc" return "esc" def _read_byte_or_none(fd: int, timeout: float) -> bytes | None: """Read one byte within ``timeout``, or None if nothing arrives.""" readable, _, _ = select.select([fd], [], [], timeout) if not readable: return None try: return os.read(fd, 1) except OSError: return None 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 _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 ) -> 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) # 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}" ) lines = [header] for i in range(start, end): lines.append( _picker_line(rows[i], i + 1, selected=(i - start == state.cursor), title_width=title_width) ) if message: status = f"\x1b[33m{message}\x1b[0m" elif state.buffer: status = f"select #{state.buffer} · Enter confirm · ⌫ clear" else: 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: """Delete the picker block lines before exiting so no residue remains. Uses Delete Line (``CSI n M``) rather than blanking lines, so the block disappears entirely instead of leaving N empty rows below the prompt. Ends with ``\r`` so the next writer (``console.print`` of the Resuming line, or the shell prompt) starts at column 0, not wherever the block's last row left the cursor. """ if prev_height: sys.stdout.write(f"\x1b[{prev_height}A") sys.stdout.write(f"\x1b[{prev_height}M") sys.stdout.write("\r") sys.stdout.flush() def _pick_session( rows: list[ClaudeSessionRow], chat_cwd: Path ) -> ClaudeSessionRow | None: """Run the interactive 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 prev_height = 0 try: tty.setraw(fd) sys.stdout.write("\x1b[?25l") # hide cursor while the picker owns the block sys.stdout.flush() 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: sys.stdout.write("\x1b[?25h") # restore cursor on the way out sys.stdout.flush() 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 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: stderr_console.print(f"[yellow]No sessions for[/yellow] {chat_cwd}") raise SystemExit(0) 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) 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]})…" ) _launch( "claude", chat_cwd, ["--resume", row.cli_session_id], use_tmux=use_tmux, ) def _first_prompt_kimi(session_file: Path) -> str: """Best-effort snippet of the first human prompt in a kimi context.jsonl.""" try: with session_file.open(encoding="utf-8", errors="ignore") as fh: for line in fh: try: entry = json.loads(line) except json.JSONDecodeError: continue if entry.get("role") != "user": continue content = entry.get("content") if isinstance(content, str) and content.strip(): return " ".join(content.split())[:80] except OSError: pass return "" def _codex_session_cwd(session_file: Path) -> str | None: """Read cwd from a codex session's session_meta record.""" try: with session_file.open(encoding="utf-8", errors="ignore") as fh: for line in fh: if not line.strip(): continue try: entry = json.loads(line) except json.JSONDecodeError: continue if entry.get("type") == "session_meta": return entry.get("payload", {}).get("cwd") except OSError: pass return None def _first_prompt_codex(session_file: Path) -> str: """Best-effort snippet of the first human prompt in a codex rollout jsonl.""" try: with session_file.open(encoding="utf-8", errors="ignore") as fh: for line in fh: try: entry = json.loads(line) except json.JSONDecodeError: continue if entry.get("type") != "user_message": continue payload = entry.get("payload", {}) if payload.get("type") != "user_message": continue content = payload.get("message") if isinstance(content, str) and content.strip(): return " ".join(content.split())[:80] except OSError: pass return "" def _dsh_matches_cwd(session_file: Path, chat_cwd: Path) -> bool: """True when a dsh session's header cwd matches ``chat_cwd``.""" stored = _dsh_session_cwd(session_file) if stored is None: return False return stored == str(chat_cwd) or Path(stored).resolve() == chat_cwd def _dsh_session_cwd(session_file: Path) -> str | None: """Read the ``cwd`` from a dsh ``session.jsonl.zstd`` header line.""" try: import io import zstandard as zstd except ImportError: return None try: with session_file.open("rb") as fh: with zstd.ZstdDecompressor().stream_reader(fh) as reader: text = io.TextIOWrapper(reader, encoding="utf-8", errors="ignore") for line in text: line = line.strip() if not line: continue try: entry = json.loads(line) except json.JSONDecodeError: continue if not isinstance(entry, dict) or entry.get("type") != "session": return None cwd = entry.get("cwd") return cwd if isinstance(cwd, str) else None except (OSError, zstd.ZstdError): return None return None def _first_prompt_dsh(session_file: Path) -> str: """Best-effort label from a dsh ``session.jsonl.zstd``. Prefers the ``session/title`` event, falling back to the first user text. """ try: import io import zstandard as zstd except ImportError: return "" try: with session_file.open("rb") as fh: with zstd.ZstdDecompressor().stream_reader(fh) as reader: text = io.TextIOWrapper(reader, encoding="utf-8", errors="ignore") for line in text: line = line.strip() if not line: continue try: entry = json.loads(line) except json.JSONDecodeError: continue if not isinstance(entry, dict): continue etype = entry.get("type") data = entry.get("data") if etype == "session/title" and isinstance(data, dict): title = data.get("title") if isinstance(title, str) and title.strip(): return " ".join(title.split())[:80] if etype == "user/message" and isinstance(data, dict): content = data.get("content") if isinstance(content, list): for block in content: if ( isinstance(block, dict) and block.get("type") == "text" ): t = block.get("text") if isinstance(t, str) and t.strip(): return " ".join(t.split())[:80] except (OSError, zstd.ZstdError): return "" return "" _USER_QUERY_RE = re.compile( r"\s*(.*?)\s*", re.DOTALL | re.IGNORECASE ) def _decode_cursor_meta_value(value: str) -> dict | None: """Decode Cursor store.db meta values (hex-encoded JSON or plain JSON).""" candidates = [value] if len(value) % 2 == 0 and all(c in "0123456789abcdefABCDEF" for c in value): try: candidates.insert(0, bytes.fromhex(value).decode("utf-8")) except (ValueError, UnicodeDecodeError): pass for raw in candidates: try: obj = json.loads(raw) except json.JSONDecodeError: continue if isinstance(obj, dict): return obj return None def _first_prompt_cursor(session_file: Path) -> str: """Best-effort snippet from a Cursor Agent ``store.db`` session.""" try: conn = sqlite3.connect(f"file:{session_file}?mode=ro", uri=True) except sqlite3.Error: return "" try: for _key, value in conn.execute("SELECT key, value FROM meta"): meta = _decode_cursor_meta_value(str(value)) if not meta: continue name = str(meta.get("name") or "").strip() if name and name.lower() not in ("new agent", "new chat", "untitled"): return " ".join(name.split())[:80] for _blob_id, data in conn.execute("SELECT id, data FROM blobs"): try: if isinstance(data, memoryview): data = data.tobytes() if isinstance(data, bytes): entry = json.loads(data.decode("utf-8", errors="ignore")) else: entry = json.loads(data) except (TypeError, json.JSONDecodeError, UnicodeDecodeError): continue if entry.get("role") != "user": continue content = entry.get("content") if not isinstance(content, str) or not content.strip(): continue match = _USER_QUERY_RE.search(content) text = match.group(1) if match else content text = " ".join(text.split()) if text: return text[:80] except sqlite3.Error: return "" finally: conn.close() return "" def _codex_session_id(session_file: Path) -> str: """Extract the UUID session id from a codex rollout filename.""" match = _SESSION_ID_UUID_RE.search(session_file.stem) return match.group(0) if match else session_file.stem def _session_files(backend: str, chat_cwd: Path) -> list[Path]: """Return session files/directories sorted by newest first.""" config = _BACKENDS[backend] sessions_dir = _sessions_dir(backend, chat_cwd) if backend == "codex": files = [p for p in sessions_dir.glob(config["session_pattern"]) if p.is_file()] cwd_str = str(chat_cwd) def _matches_cwd(p: Path) -> bool: stored = _codex_session_cwd(p) if stored is None: return False return stored == cwd_str or Path(stored).resolve() == chat_cwd files = [p for p in files if _matches_cwd(p)] files.sort(key=lambda p: p.stat().st_mtime, reverse=True) return files if backend == "dsh": # Sessions live under //session.jsonl.zstd; # filter by the cwd recorded in each session header. files = [ p for p in sessions_dir.glob(config["session_pattern"]) if p.is_file() and _dsh_matches_cwd(p, chat_cwd) ] files.sort(key=lambda p: p.stat().st_mtime, reverse=True) return files if backend == "claude": files = sorted( sessions_dir.glob(config["session_pattern"]), key=lambda p: p.stat().st_mtime, reverse=True, ) return [p for p in files if p.is_file()] if backend == "cursor": # Cursor Agent: ~/.cursor/chats///store.db files: list[Path] = [] for session_dir in sessions_dir.glob(config["session_pattern"]): if not session_dir.is_dir(): continue store = session_dir / "store.db" if store.is_file(): files.append(store) files.sort(key=lambda p: p.stat().st_mtime, reverse=True) return files # Kimi stores sessions as //context.jsonl files: list[Path] = [] for session_dir in sessions_dir.glob(config["session_pattern"]): if not session_dir.is_dir(): continue context_file = session_dir / "context.jsonl" if context_file.is_file(): files.append(context_file) files.sort(key=lambda p: p.stat().st_mtime, reverse=True) return files def _session_title(backend: str, session_file: Path) -> str: """Session label for the list: backend-native title when available, else the first human prompt.""" if backend == "claude": return _session_title_claude(session_file) if backend == "codex": return _first_prompt_codex(session_file) if backend == "cursor": return _first_prompt_cursor(session_file) if backend == "dsh": return _first_prompt_dsh(session_file) return _first_prompt_kimi(session_file) def _resume_syntax(backend: str) -> str: """Return the backend's resume syntax for use in help text.""" if backend == "kimi": return "--session, -S " if backend == "codex": return "-r, --resume or resume " if backend == "hermes": return "-r, --resume (bare -r opens the session picker)" if backend == "cursor": return "-r, --resume " if backend == "dsh": return "--resume, -r " return "--resume, -r " _HERMES_SESSIONS_SQL = """ SELECT s.id, s.started_at, s.title, (SELECT m.content FROM messages m WHERE m.session_id = s.id AND m.role = 'user' ORDER BY m.timestamp ASC LIMIT 1) AS first_message FROM sessions s ORDER BY s.started_at DESC LIMIT 20 """ def _list_sessions_hermes(chat_cwd: Path) -> None: """List recent hermes sessions from the global SQLite store. Hermes sessions are not tied to a cwd, so the newest sessions are shown regardless of chat_cwd. Snippet is the session title, falling back to the first user message. """ db_path = _BACKENDS["hermes"]["state_db"]() rows: list[tuple] = [] if db_path.is_file(): try: conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) try: rows = conn.execute(_HERMES_SESSIONS_SQL).fetchall() finally: conn.close() except sqlite3.Error as exc: stderr_console.print( f"[red]Failed to read hermes session store:[/red] {exc}" ) return if not rows: stderr_console.print(f"[yellow]No sessions for[/yellow] {chat_cwd}") return console.print( f"[bold]Sessions in[/bold] {chat_cwd} " "[dim](hermes sessions are global)[/dim]" ) for session_id, started_at, title, first_message in rows: mtime = datetime.fromtimestamp(started_at) snippet = " ".join(str(title or first_message or "").split())[:80] console.print( f" [cyan]{session_id}[/cyan] " f"[dim]{mtime:%Y-%m-%d %H:%M}[/dim] {snippet or '[dim](empty)[/dim]'}" ) console.print( "\n[dim]Resume with[/dim] [green]myhermes -r [/green] " "[dim](bare myhermes -r opens the picker)[/dim]" ) def _list_sessions(backend: str, chat_cwd: Path) -> None: """Print resumable sessions for chat_cwd, newest first.""" if backend == "hermes": _list_sessions_hermes(chat_cwd) return if backend == "claude": _list_sessions_claude(chat_cwd) return files = _session_files(backend, chat_cwd) if not files: stderr_console.print(f"[yellow]No sessions for[/yellow] {chat_cwd}") return console.print(f"[bold]Sessions in[/bold] {chat_cwd}") table = Table( box=box.SIMPLE_HEAD, show_header=True, header_style="dim", expand=True, pad_edge=False, collapse_padding=True, ) table.add_column( "Title", overflow="ellipsis", no_wrap=True, ratio=1, min_width=20 ) table.add_column( "ID", overflow="ellipsis", no_wrap=True, style="cyan", min_width=36 ) table.add_column( "Updated", justify="right", no_wrap=True, style="dim", min_width=16 ) for session_file in files: mtime = session_file.stat().st_mtime title = _session_title(backend, session_file) if backend in ("kimi", "cursor", "dsh"): # Session id is the parent directory name (//). session_id = session_file.parent.name elif backend == "codex": session_id = _codex_session_id(session_file) else: session_id = session_file.stem table.add_row( Text(title or "(empty)", style="dim" if not title else ""), session_id, datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M"), ) console.print(table) resume_prog = f"my{backend}" console.print( f"\n[dim]Resume with[/dim] [green]{resume_prog} {_resume_syntax(backend)}[/green] " "[dim](add --cwd if not workspace).[/dim]" ) class LaunchGroup(click.Group): """Group that forwards any non-subcommand invocation to the backend CLI. Unknown leading tokens (resume flags, ``--continue``, etc.) would otherwise make click raise "No such command". Instead we stash the raw tokens and route them to the hidden ``__run__`` command, which launches the backend with them as passthrough. """ def resolve_command(self, ctx, args): # type: ignore[override] if args and not args[0].startswith("-") and args[0] in self.commands: return super().resolve_command(ctx, args) ctx.meta["passthrough"] = list(args) run = self.get_command(ctx, "__run__") assert run is not None return run.name, run, [] def build_cli( backend: str, prog_name: str | None = None, offer_install: bool = False, ) -> click.Group: """Build a click CLI that wraps ``backend`` (claude, kimi, or codex). ``prog_name`` is used in --version output. When omitted it defaults to ``myagents `` (suitable for use as a subcommand of ``myagents``). With ``offer_install`` a missing backend binary triggers an interactive install prompt instead of a plain error. """ backend_title = backend.capitalize() if backend == "codex": backend_title = "OpenAI Codex" elif backend == "cursor": backend_title = "Cursor Agent" elif backend == "dsh": backend_title = "DeepSeek Harness" @click.group( cls=LaunchGroup, invoke_without_command=True, context_settings={ "ignore_unknown_options": True, "allow_extra_args": True, }, ) @click.option( "--cwd", "-C", is_flag=False, flag_value=".", default=None, type=click.Path(dir_okay=True, file_okay=False), help="Use specified path as working directory (default: workspace/).", ) @click.option( "--list", "-l", "list_sessions", is_flag=True, default=False, help=f"List resumable {backend} sessions for the working directory and exit.", ) @click.option( "--tmux", "-t", is_flag=True, default=False, help=( "Run inside tmux: reattaches to a per-directory session that " "survives SSH drops (detach with C-b d)." ), ) @click.pass_context def cli( ctx: click.Context, cwd: str | None, list_sessions: bool, tmux: bool ) -> None: """Launcher entrypoint; full help is set on the group below.""" if ctx.invoked_subcommand not in (None, "__run__"): return chat_cwd = _resolve_chat_cwd(cwd) if list_sessions: _list_sessions(backend, chat_cwd) raise SystemExit(0) if ctx.invoked_subcommand is None: _launch(backend, chat_cwd, [], use_tmux=tmux, offer_install=offer_install) @cli.command(name="__run__", hidden=True) @click.pass_context def _run(ctx: click.Context) -> None: """Hidden passthrough target: launch backend with stashed raw args.""" parent = ctx.parent assert parent is not None chat_cwd = _resolve_chat_cwd(parent.params.get("cwd")) extra = list(ctx.meta.get("passthrough", [])) if backend == "claude" and extra in (["-r"], ["--resume"]): # Bare resume opens the unified picker (jsonl ∪ xiaohe client). _resume_picker_claude( chat_cwd, use_tmux=bool(parent.params.get("tmux")) ) raise SystemExit(0) _launch( backend, chat_cwd, extra, use_tmux=bool(parent.params.get("tmux")), offer_install=offer_install, ) resume_note = "" if backend == "claude": resume_note = ( "Bare `-r`/`--resume` opens the unified picker (jsonl + xiaohe " "client sessions); `-r ` resumes directly.\n\n" ) cli.help = ( f"Launch {backend_title} in workspace/.\n\n" f"Unknown arguments ({_resume_syntax(backend)}, --continue, …) pass through to {backend}.\n\n" + resume_note + "With --tmux/-t the agent runs in a per-directory tmux session that " "survives SSH disconnects; re-run the same command to reattach." ) try: pkg_version = version("myagents") except PackageNotFoundError: pkg_version = "0.0.0" click.version_option( version=pkg_version, prog_name=prog_name or f"myagents {backend}", )(cli) return cli