path: root/claude_logbook/sessions.py

"""Parsing of the .jsonl files Claude Code leaves in ~/.claude/projects/.

Each conversation is a JSON Lines file: one line per event. From it comes one
record per session with one-letter keys, because that same record travels
embedded inside the HTML and long names are paid once per session.

Schema of the record returned by `read_session`:

    id   session uuid (the file name)
    p    project cwd
    b    git branch
    t    title
    ai   True if Claude generated the title, False if it is the first message
    n    True if it looks like a non-interactive `claude -p`
    e    True if the session has no message at all
    i    True if `p` was inferred from another session of the same project
    f/l  timestamp of the first and last event (ISO 8601)
    d    duration in minutes
    u/a  number of messages from you / from Claude
    k    size of the .jsonl in KB
    v    Claude Code version
    c    transcript: [{"r": "u" | "a" | "t" | "c" | "d" | "i" | "v", "x": text}]
         (t: tool call summary, c: compaction summary, d: the working
         directory changed to x, i: an image you attached, v: an image a
         tool gave Claude). Image blocks carry the picture itself in "src",
         as a data: URL, only when read with `images=True`
    h    how many blocks at the start of `c` were copied from the session
         this one branched from (/branch); 0 if it is not a branch
    o    id of that session, or None
    ot   its title, if it is still on disk (filled in by `_fill_gaps`)

A branch starts as a copy of the original's history. Only what came after
counts towards `u`, `a`, `d` and `f`; the copy stays in `c`, first.

`project_dir` and `mtime` are internal and never leave the module:
`public_records()` drops them before the record is serialized.
"""

import glob
import json
import os
import re
from datetime import datetime, timezone

# Bump it when the record schema changes: it invalidates old caches instead of
# reading records with the previous shape.
CACHE_VERSION = 5

EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)

INTERNAL_KEYS = ("project_dir", "mtime")


class SessionError(Exception):
    """Usage error that the CLI turns into a message and an exit code."""


# ──────────────────────────────── locations ──────────────────────────────────

def default_root():
    """~/.claude/projects, or its equivalent if CLAUDE_CONFIG_DIR is set."""
    base = os.environ.get("CLAUDE_CONFIG_DIR") or os.path.join(
        os.path.expanduser("~"), ".claude")
    return os.path.join(base, "projects")


def default_cache_path():
    base = os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache")
    return os.path.join(base, "claude-logbook", "cache.json")


def session_path(s, root=None):
    """Path of the .jsonl. The file name is the UUID and the parent directory
    name is what we keep in project_dir, so it can be rebuilt."""
    return os.path.join(root or default_root(),
                        s["project_dir"], s["id"] + ".jsonl")


# ─────────────────────────── parsing the .jsonl files ───────────────────────

TAG_RE = re.compile(r"<[^>]+>")
REMINDER_RE = re.compile(r"<system-reminder>.*?</system-reminder>", re.S)

# A message starting with any of these is not user text: it is a
# block the CLI itself generates when running a local command.
SKIP_PREFIXES = (
    "<local-command-caveat", "<command-name", "<command-message",
    "<command-args", "<local-command-stdout", "<system-reminder",
)

TITLE_MAX = 160
TOOL_ARG_MAX = 140

# Threshold of the `claude -p` heuristic: a single message longer than this, with
# no back and forth, is a pipe on stdin and not a conversation.
NONINTERACTIVE_CHARS = 1500

# Active time: a pause between two messages longer than this is the session
# left open, not work, and does not count towards its duration.
IDLE_GAP_MINUTES = 30

# For each tool, the parameter that best summarizes what it did.
TOOL_KEY = {
    "Bash": "command", "Read": "file_path", "Edit": "file_path",
    "Write": "file_path", "NotebookEdit": "notebook_path", "Glob": "pattern",
    "Grep": "pattern", "WebFetch": "url", "WebSearch": "query",
    "Task": "description", "Agent": "description", "Skill": "skill",
}


def clean_text(s):
    """Returns readable user text, or None if it is harness noise."""
    if not isinstance(s, str):
        return None
    s = s.strip()
    if not s or s.startswith(SKIP_PREFIXES):
        return None
    s = REMINDER_RE.sub(" ", s)
    s = TAG_RE.sub(" ", s)
    s = re.sub(r"\s+", " ", s).strip()
    return s if len(s) >= 3 else None


def tool_summary(block):
    """A line like 'Bash: git status' for a tool call."""
    name = block.get("name") or "tool"
    args = block.get("input") or {}
    if not isinstance(args, dict):
        return name
    val = args.get(TOOL_KEY.get(name, ""))
    if val is None:
        val = next((v for v in args.values() if isinstance(v, str)), None)
    if not isinstance(val, str):
        return name
    val = re.sub(r"\s+", " ", val).strip()
    if len(val) > TOOL_ARG_MAX:
        val = val[:TOOL_ARG_MAX] + "…"
    return f"{name}: {val}" if val else name


# The formats a browser shows that cannot carry a script (so no SVG).
IMAGE_TYPES = ("image/png", "image/jpeg", "image/gif", "image/webp")


def image_src(block):
    """data: URL of an image block, or None if it is not one we can show."""
    src = block.get("source")
    if not isinstance(src, dict) or src.get("type") != "base64":
        return None
    kind, data = src.get("media_type"), src.get("data")
    if kind not in IMAGE_TYPES or not isinstance(data, str) or not data:
        return None
    return f"data:{kind};base64,{data}"


def image_block(role, block, images):
    out = {"r": role, "x": "[image]"}
    if images:
        src = image_src(block)
        if src:
            out["src"] = src
    return out


def blocks_of(message):
    content = message.get("content")
    if isinstance(content, str):
        return [{"type": "text", "text": content}]
    return content if isinstance(content, list) else []


def parse_ts(ts):
    """ISO 8601 → timezone-aware datetime, or None if it cannot be read."""
    if not ts:
        return None
    try:
        return datetime.fromisoformat(ts.replace("Z", "+00:00"))
    except (ValueError, AttributeError):
        return None


def active_minutes(stamps):
    """Minutes of work: the gaps between consecutive messages, leaving out
    the pauses longer than IDLE_GAP_MINUTES. None without messages."""
    times = sorted(t for t in map(parse_ts, stamps) if t)
    if not times:
        return None
    total = 0.0
    for a, b in zip(times, times[1:]):
        gap = (b - a).total_seconds() / 60
        if gap <= IDLE_GAP_MINUTES:
            total += gap
    return round(total)


def read_session(path, images=False):
    """Parses a whole .jsonl and returns that session's record. With images
    the image blocks bring the pictures, which can weigh megabytes: that
    record is for a single export, never for the cache."""
    session_id = os.path.basename(path)[:-6]  # without .jsonl
    first_ts = last_ts = cwd = here = git_branch = version = None
    origin = None  # the session /branch copied the history from
    inherited = None  # blocks of c that came with that copy
    old_first_ts = None  # first timestamp of the copied rows
    ai_title = custom_title = fallback_title = None
    stamps = []  # of the messages, for the active time
    user_msgs = assistant_msgs = 0
    convo = []

    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue  # line truncated by a session that is still writing
            if not isinstance(obj, dict):
                continue

            kind = obj.get("type")

            if kind == "ai-title":
                if obj.get("aiTitle"):
                    ai_title = obj["aiTitle"]  # keep the most recent one
                continue
            if kind == "custom-title":
                if obj.get("customTitle"):
                    custom_title = obj["customTitle"]  # /rename, wins over the AI one
                continue

            # /branch copies the original rows into the new file, each one
            # marked with where it came from. They come first.
            forked = obj.get("forkedFrom")
            copied = isinstance(forked, dict)
            if copied and forked.get("sessionId"):
                origin = forked["sessionId"]

            ts = obj.get("timestamp")
            if ts:
                if copied:
                    old_first_ts = old_first_ts or ts
                elif first_ts is None:
                    first_ts = ts
                last_ts = ts
            if cwd is None and obj.get("cwd"):
                cwd = here = obj["cwd"]
            if git_branch is None and obj.get("gitBranch"):
                git_branch = obj["gitBranch"]
            if obj.get("version"):
                version = obj["version"]

            if kind not in ("user", "assistant") or obj.get("isSidechain"):
                continue

            message = obj.get("message")
            if not isinstance(message, dict):
                continue
            if not copied:
                if origin and inherited is None:
                    inherited = len(convo)  # the branch's own part starts here
                if ts:
                    stamps.append(ts)
            # A `cd` that sticks, /cd or a worktree: from here on the session
            # works somewhere else. Only messages count, so a subagent or a
            # hook running elsewhere does not move it.
            if obj.get("cwd") and obj["cwd"] != here:
                here = obj["cwd"]
                convo.append({"r": "d", "x": here})

            if kind == "user":
                if obj.get("isMeta"):
                    continue
                # What /compact (or running out of context) left in place of
                # the earlier conversation: written by Claude, not by you.
                if obj.get("isCompactSummary"):
                    text = "\n".join(
                        (b.get("text") or "").strip() for b in blocks_of(message)
                        if isinstance(b, dict) and b.get("type") == "text").strip()
                    convo.append({"r": "c", "x": text})
                    continue
                for b in blocks_of(message):
                    if not isinstance(b, dict):
                        continue
                    if b.get("type") == "text":
                        text = clean_text(b.get("text"))
                        if text:
                            user_msgs += not copied
                            if fallback_title is None:
                                fallback_title = text[:TITLE_MAX