session-tools/bin/_claude-projects

#!/usr/bin/env python3
# Internal helper for rclaude — enumerates Claude Code state from
# ~/.claude/projects/. Two modes:
#
#   (default)        one tab-separated row per project directory
#                    columns: <mtime_epoch>\t<cwd>\t<session_count>
#
#   --sessions       one tab-separated row per session jsonl
#                    columns: <mtime_epoch>\t<session_uuid>\t<cwd>\t<snippet>
#
# Used by `rclaude list` and `rclaude resume` to discover sessions that exist
# on disk but have no live tmux session attached. Run locally or invoked
# remotely via ssh.

import json
import os
import re
import signal
import sys
from pathlib import Path

# Don't dump a traceback when piped into `head`.
signal.signal(signal.SIGPIPE, signal.SIG_DFL)

ROOT = Path.home() / ".claude" / "projects"

# Reading every jsonl on a host with thousands of past sessions is slow. Cap
# the per-session listing to the N most recently modified jsonls.
SESSION_LIMIT = int(os.environ.get("CLAUDE_PROJECTS_LIMIT", "500"))

# Lines that look system-injected rather than user-typed. When picking the
# "name" of a session for fuzzy matching we want the first real user line.
SYSTEM_PREFIXES = (
    "<command-name>", "<command-message>", "<system-reminder>",
    "<local-command-", "<bash-input>", "<bash-stdout>",
    "Caveat: The messages below",
    "[task-persistence]", "[tts-state]", "[VERIFY:",
    "This session is being continued", "Please continue",
    "[result]", "<task-notification>", "[Request interrupted",
)


def looks_system(text: str) -> bool:
    s = (text or "").lstrip()
    if not s:
        return True
    return s.startswith(SYSTEM_PREFIXES)


def message_text(entry) -> str:
    msg = entry.get("message") or {}
    content = msg.get("content")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for block in content:
            if isinstance(block, dict) and block.get("type") == "text":
                parts.append(block.get("text", ""))
        return " ".join(parts)
    return ""


def first_user_snippet(jsonl: Path) -> str:
    """Return a one-line snippet of the first non-system user message."""
    try:
        with jsonl.open("r", encoding="utf-8", errors="replace") as f:
            for line in f:
                try:
                    entry = json.loads(line)
                except json.JSONDecodeError:
                    continue
                role = (entry.get("message") or {}).get("role") or entry.get("type")
                if role != "user":
                    continue
                text = message_text(entry)
                if looks_system(text):
                    continue
                snippet = re.sub(r"\s+", " ", text).strip()
                return snippet[:120]
    except OSError:
        pass
    return ""


def session_cwd(jsonl: Path) -> str | None:
    try:
        with jsonl.open("r", encoding="utf-8", errors="replace") as f:
            for line in f:
                try:
                    entry = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if entry.get("cwd"):
                    return entry["cwd"]
    except OSError:
        return None
    return None


def list_projects():
    if not ROOT.is_dir():
        return
    rows = []
    for project_dir in ROOT.iterdir():
        if not project_dir.is_dir():
            continue
        jsonls = sorted(
            project_dir.glob("*.jsonl"),
            key=lambda p: p.stat().st_mtime,
            reverse=True,
        )
        if not jsonls:
            continue
        cwd = session_cwd(jsonls[0])
        if not cwd:
            continue
        mtime = int(jsonls[0].stat().st_mtime)
        rows.append((mtime, cwd, len(jsonls)))
    rows.sort(reverse=True)
    for mtime, cwd, count in rows:
        print(f"{mtime}\t{cwd}\t{count}")


def list_sessions():
    if not ROOT.is_dir():
        return
    # First pass: cheap stat-only collection, then keep the N most-recent
    # to bound the second (expensive) jsonl-parse pass.
    candidates = []
    for project_dir in ROOT.iterdir():
        if not project_dir.is_dir():
            continue
        for jsonl in project_dir.glob("*.jsonl"):
            try:
                mtime = int(jsonl.stat().st_mtime)
            except OSError:
                continue
            candidates.append((mtime, jsonl))
    candidates.sort(key=lambda x: x[0], reverse=True)
    for mtime, jsonl in candidates[:SESSION_LIMIT]:
        uuid = jsonl.stem
        cwd = session_cwd(jsonl) or ""
        if not cwd:
            continue
        snippet = first_user_snippet(jsonl).replace("\t", " ")
        print(f"{mtime}\t{uuid}\t{cwd}\t{snippet}")


def main():
    args = sys.argv[1:]
    if args and args[0] == "--sessions":
        list_sessions()
    else:
        list_projects()


if __name__ == "__main__":
    main()