From 88170bdf57d40216fd056890fd583a32af4c9034 Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Sat, 15 Aug 2026 13:42:53 +0800 Subject: [PATCH] myclaude -r: unify resume picker with xiaohe client sessions Bare -r/--resume now opens a numbered picker merging jsonl (source of truth) with the xiaohe hosted-session index, deduped by cli_session_id and sorted by real content activity (not jsonl mtime, which the xiaohe projection re-touches). -l gains a Src column (xiaohe/cli); -r and -c stay passthrough. Adds read-only xiaohe_sessions data layer + tests. --- myagents/launcher.py | 236 +++++++++++++++++++++++++++++++- myagents/xiaohe_sessions.py | 111 +++++++++++++++ tests/test_entrypoints.py | 18 ++- tests/test_launcher_sessions.py | 181 ++++++++++++++++++++++++ tests/test_xiaohe_sessions.py | 90 ++++++++++++ 5 files changed, 634 insertions(+), 2 deletions(-) create mode 100644 myagents/xiaohe_sessions.py create mode 100644 tests/test_launcher_sessions.py create mode 100644 tests/test_xiaohe_sessions.py diff --git a/myagents/launcher.py b/myagents/launcher.py index 2522776..200ac8c 100644 --- a/myagents/launcher.py +++ b/myagents/launcher.py @@ -9,6 +9,7 @@ import shutil import sqlite3 import subprocess import sys +from dataclasses import dataclass from datetime import datetime from importlib.metadata import PackageNotFoundError, version from pathlib import Path @@ -16,10 +17,12 @@ from pathlib import Path import click from rich import box from rich.console import Console +from rich.prompt import Prompt 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() @@ -645,6 +648,221 @@ def _session_title_claude(session_file: Path) -> str: 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: + """Shared rich table for ``myclaude -l`` and the resume picker.""" + 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]" + ) + + +def _resume_picker_claude(chat_cwd: Path, use_tmux: bool = False) -> None: + """Interactive unified resume picker for the claude backend. + + Lists jsonl + xiaohe client sessions for chat_cwd; selecting one resumes + the jsonl via ``claude --resume ``. + """ + rows = _claude_session_rows(chat_cwd) + if not rows: + stderr_console.print(f"[yellow]No sessions for[/yellow] {chat_cwd}") + raise SystemExit(0) + _render_claude_table(rows, chat_cwd, numbered=True) + if not sys.stdin.isatty(): + stderr_console.print( + "[yellow]Not a terminal — resume directly with[/yellow] " + "[green]myclaude -r [/green]" + ) + raise SystemExit(0) + selectable = [r for r in rows if r.jsonl_path is not None] + while True: + answer = Prompt.ask( + "[bold]Choose session[/bold] " + "(Enter = latest, [cyan]q[/cyan] = quit)", + default="", + ) + if answer.strip().lower() == "q": + raise SystemExit(0) + if answer.strip() == "": + row = selectable[0] + break + if not answer.strip().isdigit(): + continue + n = int(answer.strip()) + if not 1 <= n <= len(rows): + continue + row = rows[n - 1] + if row.jsonl_path is None: + stderr_console.print( + "[red]Client-only session[/red] — no jsonl to resume in the " + "CLI; continue it in the xiaohe web/desktop client instead." + ) + continue + break + 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: @@ -909,6 +1127,9 @@ def _list_sessions(backend: str, chat_cwd: Path) -> None: 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}") @@ -1051,6 +1272,12 @@ def build_cli( 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, @@ -1059,10 +1286,17 @@ def build_cli( 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" - "With --tmux/-t the agent runs in a per-directory tmux session that " + + 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: diff --git a/myagents/xiaohe_sessions.py b/myagents/xiaohe_sessions.py new file mode 100644 index 0000000..34051b6 --- /dev/null +++ b/myagents/xiaohe_sessions.py @@ -0,0 +1,111 @@ +"""Read-only access to xiaohe hosted-session SQLite stores. + +myagents runs against Claude Code CLI sessions (``~/.claude/projects``), but the +xiaohe client also drives Claude Code and keeps a hosted-session index for the +same jsonl transcripts. ``myclaude -r`` unifies both sources so client sessions +and plain CLI sessions appear in one resumable list. + +Every accessor takes explicit path parameters and degrades to an empty result +when a store is missing or unreadable — a machine without xiaohe data behaves +exactly as before. +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass +from pathlib import Path + +_SESSIONS_DB = Path.home() / ".xiaohe" / "data" / "sessions-claude.db" +_WORKSPACES_DB = Path.home() / ".xiaohe" / "data" / "workspaces.db" + + +def _connect_readonly(path: Path) -> sqlite3.Connection | None: + """Open a SQLite store read-only, or ``None`` when unavailable.""" + try: + return sqlite3.connect(f"file:{path}?mode=ro", uri=True) + except (sqlite3.Error, OSError): + return None + + +def workspace_ids_for_cwd( + chat_cwd: Path, workspaces_db: Path = _WORKSPACES_DB +) -> list[str]: + """Workspace ids whose bind path resolves to ``chat_cwd`` (e.g. ``primary``). + + Empty when the store is missing or no workspace binds to the cwd. + """ + conn = _connect_readonly(workspaces_db) + if conn is None: + return [] + try: + rows = conn.execute( + "SELECT workspace_id, workspace_path FROM workspaces " + "WHERE workspace_path IS NOT NULL" + ).fetchall() + except sqlite3.Error: + return [] + finally: + conn.close() + + target = chat_cwd.resolve() + return [ + str(rid) + for rid, raw_path in rows + if raw_path and Path(str(raw_path)).expanduser().resolve() == target + ] + + +@dataclass(frozen=True) +class XiaoheSession: + """Hosted-session index row for one cli_session_id (jsonl UUID).""" + + cli_session_id: str + xiaohe_session_id: str + title: str | None + status: str | None + updated_at: float | None + + +def session_index( + workspace_ids: list[str], sessions_db: Path = _SESSIONS_DB +) -> dict[str, XiaoheSession]: + """Map ``cli_session_id`` → hosted-session row for the given workspaces.""" + if not workspace_ids: + return {} + conn = _connect_readonly(sessions_db) + if conn is None: + return {} + placeholders = ",".join("?" for _ in workspace_ids) + query = ( + "SELECT cli_session_id, session_id, title, status, updated_at " + "FROM sessions " + f"WHERE workspace_id IN ({placeholders}) AND cli_session_id IS NOT NULL" + ) + try: + rows = conn.execute(query, workspace_ids).fetchall() + except sqlite3.Error: + return {} + finally: + conn.close() + + index: dict[str, XiaoheSession] = {} + for cli_id, xh_id, title, status, ts in rows: + index[str(cli_id)] = XiaoheSession( + cli_session_id=str(cli_id), + xiaohe_session_id=str(xh_id), + title=str(title) if title else None, + status=str(status) if status else None, + updated_at=float(ts) if isinstance(ts, (int, float)) else None, + ) + return index + + +def default_sessions_db() -> Path: + """Default hosted-sessions store path (kept importable for tests).""" + return _SESSIONS_DB + + +def default_workspaces_db() -> Path: + """Default workspace store path (kept importable for tests).""" + return _WORKSPACES_DB diff --git a/tests/test_entrypoints.py b/tests/test_entrypoints.py index e544425..3ea1f8d 100644 --- a/tests/test_entrypoints.py +++ b/tests/test_entrypoints.py @@ -37,6 +37,22 @@ class TestMyclaudeEntrypoint: assert "myclaude" in result.output def test_passthrough(self, tmp_path: Path) -> None: + """``--resume `` still passes through to the native CLI.""" + runner = CliRunner() + with ( + patch("myagents.launcher.shutil.which", return_value="/usr/bin/claude"), + patch("myagents.launcher.subprocess.run") as mock_run, + ): + mock_run.return_value = MagicMock(returncode=0) + result = runner.invoke( + claude_cli, ["--cwd", str(tmp_path), "--resume", "abc123"] + ) + assert result.exit_code == 0 + args = mock_run.call_args[0][0] + assert args[-2:] == ["--resume", "abc123"] + + def test_bare_resume_opens_picker(self, tmp_path: Path) -> None: + """Bare ``--resume`` opens the unified picker instead of the CLI.""" runner = CliRunner() with ( patch("myagents.launcher.shutil.which", return_value="/usr/bin/claude"), @@ -45,7 +61,7 @@ class TestMyclaudeEntrypoint: mock_run.return_value = MagicMock(returncode=0) result = runner.invoke(claude_cli, ["--cwd", str(tmp_path), "--resume"]) assert result.exit_code == 0 - assert "--resume" in mock_run.call_args[0][0] + mock_run.assert_not_called() class TestMykimiEntrypoint: diff --git a/tests/test_launcher_sessions.py b/tests/test_launcher_sessions.py new file mode 100644 index 0000000..6232e11 --- /dev/null +++ b/tests/test_launcher_sessions.py @@ -0,0 +1,181 @@ +"""Tests for unified claude session listing/resume (jsonl ∪ xiaohe index).""" + +import copy +import json +import re +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import myagents.launcher as L +from myagents.xiaohe_sessions import XiaoheSession + + +def _slug(path: Path) -> str: + return re.sub(r"[^A-Za-z0-9]", "-", str(path.resolve())) + + +def _write_jsonl( + proj_dir: Path, cid: str, ts: str, *, ai_title: str | None = None +) -> Path: + """Write a minimal Claude Code jsonl with a message timestamp + optional aiTitle.""" + lines: list[dict[str, object]] = [ + {"type": "mode", "mode": "normal", "sessionId": cid}, + { + "type": "permission-mode", + "permissionMode": "bypassPermissions", + "sessionId": cid, + }, + ] + if ai_title: + lines.append( + {"type": "ai-title", "aiTitle": ai_title, "sessionId": cid} + ) + lines.append( + { + "type": "user", + "sessionId": cid, + "message": { + "id": f"m-{cid}", + "role": "user", + "content": [{"type": "text", "text": f"first prompt {cid}"}], + "timestamp": ts, + }, + } + ) + path = proj_dir / f"{cid}.jsonl" + path.write_text( + "\n".join(json.dumps(line) for line in lines) + "\n", encoding="utf-8" + ) + return path + + +@pytest.fixture +def claude_env(tmp_path: Path, monkeypatch): + """Point the claude session root + xiaohe index at temp fixtures.""" + projects = tmp_path / "projects" + cwd = tmp_path / "cwd" + cwd.mkdir() + proj_dir = projects / _slug(cwd) + proj_dir.mkdir(parents=True) + + index: dict[str, XiaoheSession] = {} + + def fake_workspace_ids(_chat_cwd): + return ["primary"] + + def fake_session_index(workspace_ids): + return index + + monkeypatch.setattr(L, "_BACKENDS", copy.deepcopy(L._BACKENDS)) + L._BACKENDS["claude"]["sessions_root"] = lambda: projects + monkeypatch.setattr(L, "workspace_ids_for_cwd", fake_workspace_ids) + monkeypatch.setattr(L, "session_index", fake_session_index) + return SimpleNamespace(cwd=cwd, proj_dir=proj_dir, index=index) + + +class TestClaudeSessionRows: + def test_merges_jsonl_with_xiaohe_index(self, claude_env) -> None: + _write_jsonl( + claude_env.proj_dir, + "aaa", + "2026-08-15T04:00:00Z", + ai_title="Client A", + ) + _write_jsonl(claude_env.proj_dir, "bbb", "2026-08-15T03:00:00Z") + claude_env.index["aaa"] = XiaoheSession( + cli_session_id="aaa", + xiaohe_session_id="x-1", + title="Client A", + status="ready", + updated_at=1000.0, + ) + + rows = L._claude_session_rows(claude_env.cwd) + by_id = {r.cli_session_id: r for r in rows} + + assert by_id["aaa"].origin == "xiaohe" + assert by_id["aaa"].status == "ready" + assert by_id["aaa"].xiaohe_session_id == "x-1" + assert by_id["aaa"].title == "Client A" + assert by_id["bbb"].origin == "cli" + assert by_id["bbb"].status is None + + def test_sorts_by_content_activity_desc(self, claude_env) -> None: + _write_jsonl(claude_env.proj_dir, "old", "2026-08-13T04:00:00Z") + _write_jsonl(claude_env.proj_dir, "mid", "2026-08-14T04:00:00Z") + _write_jsonl(claude_env.proj_dir, "new", "2026-08-15T04:00:00Z") + + rows = L._claude_session_rows(claude_env.cwd) + assert [r.cli_session_id for r in rows] == ["new", "mid", "old"] + + def test_db_only_session_is_client_only(self, claude_env) -> None: + _write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z") + claude_env.index["nojsonl"] = XiaoheSession( + cli_session_id="nojsonl", + xiaohe_session_id="x-2", + title="Lost", + status="ready", + updated_at=2000.0, + ) + + rows = L._claude_session_rows(claude_env.cwd) + by_id = {r.cli_session_id: r for r in rows} + assert by_id["nojsonl"].jsonl_path is None + assert by_id["nojsonl"].origin == "xiaohe" + + def test_dedup_by_cli_session_id(self, claude_env) -> None: + _write_jsonl(claude_env.proj_dir, "dup", "2026-08-15T04:00:00Z") + claude_env.index["dup"] = XiaoheSession( + cli_session_id="dup", + xiaohe_session_id="x-3", + title="Dup", + status="ready", + updated_at=1.0, + ) + rows = L._claude_session_rows(claude_env.cwd) + ids = [r.cli_session_id for r in rows] + assert ids.count("dup") == 1 + + +class TestResumePickerClaude: + def test_non_tty_prints_and_exits(self, claude_env, capsys) -> None: + _write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z") + with pytest.raises(SystemExit) as exc: + L._resume_picker_claude(claude_env.cwd) + assert exc.value.code == 0 + err = capsys.readouterr().err + assert "Not a terminal" in err + assert "resume directly with myclaude -r " in err + + def test_selecting_number_launches_resume( + self, claude_env, monkeypatch + ) -> None: + _write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z") + _write_jsonl(claude_env.proj_dir, "bbb", "2026-08-15T03:00:00Z") + + launched: list[list[str]] = [] + monkeypatch.setattr( + L, "_launch", lambda *_a, **_k: launched.append(_a[2]) + ) + monkeypatch.setattr(sys.stdin, "isatty", lambda: True) + monkeypatch.setattr(L.Prompt, "ask", lambda *a, **k: "2") + + L._resume_picker_claude(claude_env.cwd) + assert launched == [["--resume", "bbb"]] + + def test_enter_picks_latest(self, claude_env, monkeypatch) -> None: + _write_jsonl(claude_env.proj_dir, "aaa", "2026-08-15T04:00:00Z") + _write_jsonl(claude_env.proj_dir, "bbb", "2026-08-15T03:00:00Z") + + launched: list[list[str]] = [] + monkeypatch.setattr( + L, "_launch", lambda *_a, **_k: launched.append(_a[2]) + ) + monkeypatch.setattr(sys.stdin, "isatty", lambda: True) + monkeypatch.setattr(L.Prompt, "ask", lambda *a, **k: "") + + L._resume_picker_claude(claude_env.cwd) + assert launched == [["--resume", "aaa"]] diff --git a/tests/test_xiaohe_sessions.py b/tests/test_xiaohe_sessions.py new file mode 100644 index 0000000..aca918a --- /dev/null +++ b/tests/test_xiaohe_sessions.py @@ -0,0 +1,90 @@ +"""Tests for the read-only xiaohe hosted-session data layer.""" + +import sqlite3 +from pathlib import Path + +from myagents.xiaohe_sessions import session_index, workspace_ids_for_cwd + + +def _make_workspaces_db(path: Path, rows: list[tuple[str, str | None]]) -> None: + conn = sqlite3.connect(path) + conn.execute( + "CREATE TABLE workspaces (workspace_id TEXT, workspace_path TEXT)" + ) + conn.executemany("INSERT INTO workspaces VALUES (?, ?)", rows) + conn.commit() + conn.close() + + +def _make_sessions_db(path: Path, rows: list[tuple]) -> None: + conn = sqlite3.connect(path) + conn.execute( + "CREATE TABLE sessions (" + "session_id TEXT PRIMARY KEY, workspace_id TEXT, cli_session_id TEXT, " + "title TEXT, status TEXT, updated_at REAL)" + ) + conn.executemany("INSERT INTO sessions VALUES (?, ?, ?, ?, ?, ?)", rows) + conn.commit() + conn.close() + + +class TestWorkspaceIdsForCwd: + def test_returns_matching_bind_workspaces(self, tmp_path: Path) -> None: + """Bind workspaces whose path resolves to the cwd are returned.""" + db = tmp_path / "workspaces.db" + cwd = tmp_path / "ws" + cwd.mkdir() + _make_workspaces_db( + db, + [ + ("primary", str(cwd)), + ("bot-f", str(cwd)), + ("other", str(tmp_path / "elsewhere")), + ("session-ws", None), + ], + ) + result = workspace_ids_for_cwd(cwd, workspaces_db=db) + assert result == ["primary", "bot-f"] + + def test_no_match_returns_empty(self, tmp_path: Path) -> None: + db = tmp_path / "workspaces.db" + cwd = tmp_path / "ws" + _make_workspaces_db(db, [("primary", str(tmp_path / "other"))]) + assert workspace_ids_for_cwd(cwd, workspaces_db=db) == [] + + def test_missing_store_returns_empty(self, tmp_path: Path) -> None: + assert workspace_ids_for_cwd(tmp_path / "ws") == [] + + +class TestSessionIndex: + def test_maps_cli_ids_for_matching_workspaces(self, tmp_path: Path) -> None: + db = tmp_path / "sessions-claude.db" + _make_sessions_db( + db, + [ + ("s1", "primary", "c1", "Title A", "ready", 1000.0), + ("s2", "primary", "c2", None, "error", 2000.0), + ("s3", "other", "c3", "Title B", "ready", 3000.0), + ], + ) + index = session_index(["primary"], sessions_db=db) + assert set(index) == {"c1", "c2"} + assert index["c1"].xiaohe_session_id == "s1" + assert index["c1"].status == "ready" + assert index["c2"].title is None + assert index["c2"].updated_at == 2000.0 + + def test_skips_null_cli_session_id(self, tmp_path: Path) -> None: + db = tmp_path / "sessions-claude.db" + _make_sessions_db(db, [("s1", "primary", None, "T", "ready", 1.0)]) + assert session_index(["primary"], sessions_db=db) == {} + + def test_empty_workspace_ids_skips_query(self, tmp_path: Path) -> None: + db = tmp_path / "sessions-claude.db" + _make_sessions_db(db, [("s1", "primary", "c1", "T", "ready", 1.0)]) + assert session_index([], sessions_db=db) == {} + + def test_missing_store_returns_empty(self, tmp_path: Path) -> None: + assert ( + session_index(["primary"], sessions_db=tmp_path / "nope.db") == {} + )