Zhengshou Lai
88170bdf57
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.
112 lines
3.4 KiB
Python
112 lines
3.4 KiB
Python
"""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
|