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 <id> and
-c stay passthrough. Adds read-only xiaohe_sessions data layer + tests.

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 <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]"
+    )
+
+
+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 <cli_session_id>``.
+    """
+    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 <id>[/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 <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"
-        "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:

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

tests/test_entrypoints.py

@@ -37,6 +37,22 @@ class TestMyclaudeEntrypoint:
         assert "myclaude" in result.output
 
     def test_passthrough(self, tmp_path: Path) -> None:
+        """``--resume <id>`` 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:

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 <id>" 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"]]

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") == {}
+        )