Zhengshou Lai
75a72a8b24
- _read_key 改按需读 CSI 序列（方向键读到终止符即返回），连按 \x1b[B 不再被吞进同一序列误判为 ESC 退出
- picker 全程隐藏光标（\x1b[?25l），退出恢复（\x1b[?25h），末尾不再突兀闪现光标
- 退出改用 Delete Line（\x1b[{n}M）删除块区域，替代逐行清空，不再留下 N 个空行
- tests: 新增快速连按方向键回归测试
1573 lines
54 KiB
Python
1573 lines
54 KiB
Python
"""Shared launcher logic for AI coding agent CLIs."""
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
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"],
|
||
"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"],
|
||
"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]."
|
||
),
|
||
},
|
||
}
|
||
|
||
|
||
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 _resolve_backend_binary(config: dict) -> str | None:
|
||
"""Resolve backend CLI: env override → PATH → npm global prefix."""
|
||
binary = os.environ.get(config["env_bin"]) or shutil.which(config["binary"])
|
||
if binary:
|
||
return binary
|
||
for alt in config.get("alt_binaries", ()):
|
||
found = shutil.which(alt)
|
||
if found:
|
||
return found
|
||
return _resolve_from_npm_global(config)
|
||
|
||
|
||
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: <prefix>/bin; Windows npm usually puts shims in <prefix> itself.
|
||
if sys.platform == "win32":
|
||
return root
|
||
return root / "bin"
|
||
|
||
|
||
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]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 _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
|
||
installer = shutil.which(install_cmd[0])
|
||
if not installer:
|
||
hint = _ensure_agent_hint()
|
||
stderr_console.print(
|
||
f"[red]{install_cmd[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(install_cmd)}[/cyan]"
|
||
)
|
||
stderr_console.print(
|
||
"[dim] (npm global install — may take a minute)[/dim]"
|
||
)
|
||
proc = subprocess.run([installer, *install_cmd[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)
|
||
return _resolve_backend_binary(config)
|
||
|
||
|
||
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(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)
|
||
|
||
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 == "codex":
|
||
# Codex stores all sessions under a single dated tree; cwd is in metadata.
|
||
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 <id>`.
|
||
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 <id>[/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.
|
||
"""
|
||
if prev_height:
|
||
sys.stdout.write(f"\x1b[{prev_height}A")
|
||
sys.stdout.write(f"\x1b[{prev_height}M")
|
||
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 <id>``.
|
||
"""
|
||
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 <id>[/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 ""
|
||
|
||
|
||
_USER_QUERY_RE = re.compile(
|
||
r"<user_query>\s*(.*?)\s*</user_query>", 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 == "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/<cwd-md5>/<chat-id>/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 <cwd-md5>/<session-id>/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)
|
||
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 <id>"
|
||
if backend == "codex":
|
||
return "-r, --resume <id> or resume <id>"
|
||
if backend == "hermes":
|
||
return "-r, --resume <id> (bare -r opens the session picker)"
|
||
if backend == "cursor":
|
||
return "-r, --resume <id>"
|
||
return "--resume, -r <id>"
|
||
|
||
|
||
_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 <id>[/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"):
|
||
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 <backend>`` (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"
|
||
|
||
@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 <id>` 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
|