transcript_parser.py

"""
Parse Claude Code JSONL transcripts into conversation turns for indexing.

Transcript format (one JSON object per line):
- type: "user"     → user message (content is str or list of tool_result)
- type: "assistant" → assistant message (content is list of text/tool_use/thinking blocks)
- type: "summary"  → compaction summary (short title string)
- type: "system", "file-history-snapshot", "progress", "queue-operation" → skip

Turn assembly:
1. A user entry with plain string content starts a new turn
2. Subsequent assistant text blocks are accumulated
3. Combined: "User: {user_text}\\n\\nAssistant: {assistant_text}"
4. Summary entries become standalone chunks

Incremental reading:
- index_state.json tracks last_byte_offset per transcript file
- On each invocation, seek to offset, read only new lines
"""

import hashlib
import json
import os
from pathlib import Path
from typing import List, Dict, Optional, Tuple


def parse_transcript(
    transcript_path: str,
    session_id: str,
    start_offset: int = 0,
    max_turn_chars: int = 8000,
) -> Tuple[List[Dict], int]:
    """Parse a Claude Code JSONL transcript into indexable turns.

    Args:
        transcript_path: Path to the .jsonl file
        session_id: Session UUID for this transcript
        start_offset: Byte offset to resume reading from
        max_turn_chars: Maximum characters per turn text (truncate longer)

    Returns:
        (turns, new_offset) where turns is a list of dicts ready for rag_engine.add_turns()
        and new_offset is the byte position after the last line read.
    """
    turns = []
    current_user_text = None
    current_user_start_byte = 0
    current_assistant_texts = []
    current_timestamp = ""
    current_git_branch = ""

    file_size = os.path.getsize(transcript_path)
    if start_offset >= file_size:
        return [], file_size

    transcript_file = os.path.basename(transcript_path)

    with open(transcript_path, "r", encoding="utf-8") as f:
        f.seek(start_offset)
        current_offset = start_offset

        for line in f:
            line_bytes = len(line.encode("utf-8"))
            current_offset += line_bytes

            line = line.strip()
            if not line:
                continue

            try:
                entry = json.loads(line)
            except json.JSONDecodeError:
                continue

            entry_type = entry.get("type", "")

            # Track git branch from any entry that has it
            entry_branch = entry.get("gitBranch", "")
            if entry_branch:
                current_git_branch = entry_branch

            # Track timestamp from any entry that has it
            entry_ts = entry.get("timestamp", "")
            if entry_ts:
                current_timestamp = entry_ts

            # Skip non-conversation entries
            if entry_type in ("file-history-snapshot", "progress", "system", "queue-operation"):
                continue

            # Handle summary entries (compaction summaries)
            if entry_type == "summary":
                # Flush any pending turn first
                if current_user_text is not None:
                    turn = _build_turn(
                        current_user_text, current_assistant_texts,
                        session_id, transcript_file,
                        current_user_start_byte,
                        current_timestamp, current_git_branch,
                        max_turn_chars, "turn",
                    )
                    if turn:
                        turns.append(turn)
                    current_user_text = None
                    current_assistant_texts = []

                summary_text = entry.get("summary", "")
                if summary_text:
                    summary_full = f"Session Summary: {summary_text}"
                    # Use current byte offset as turn_index for summaries
                    summary_byte = current_offset - len(line.encode("utf-8")) if line else current_offset
                    content_hash = hashlib.sha256(
                        f"{summary_byte}:{summary_full}".encode()
                    ).hexdigest()[:16]
                    turns.append({
                        "text": summary_full,
                        "doc_id": f"{session_id}::{content_hash}",
                        "session_id": session_id,
                        "transcript_file": transcript_file,
                        "turn_index": summary_byte,
                        "timestamp": current_timestamp,
                        "git_branch": current_git_branch,
                        "chunk_type": "summary",
                    })
                continue

            # Handle user messages
            if entry_type == "user":
                message = entry.get("message", {})
                content = message.get("content", "")

                # Skip tool_result messages (content is a list)
                if isinstance(content, list):
                    continue

                # Skip isMeta messages
                if entry.get("isMeta"):
                    continue

                # Skip empty content
                if not isinstance(content, str) or not content.strip():
                    continue

                # Flush previous turn if we have one
                if current_user_text is not None:
                    turn = _build_turn(
                        current_user_text, current_assistant_texts,
                        session_id, transcript_file,
                        current_user_start_byte,
                        current_timestamp, current_git_branch,
                        max_turn_chars, "turn",
                    )
                    if turn:
                        turns.append(turn)

                # Start new turn — record byte position for content-based doc_id
                current_user_text = content.strip()
                current_user_start_byte = current_offset - line_bytes
                current_assistant_texts = []
                continue

            # Handle assistant messages
            if entry_type == "assistant":
                message = entry.get("message", {})
                content = message.get("content", [])

                if not isinstance(content, list):
                    continue

                for block in content:
                    if not isinstance(block, dict):
                        continue
                    if block.get("type") == "text":
                        text = block.get("text", "").strip()
                        if text:
                            current_assistant_texts.append(text)

        # Flush final pending turn
        if current_user_text is not None:
            turn = _build_turn(
                current_user_text, current_assistant_texts,
                session_id, transcript_file,
                current_user_start_byte,
                current_timestamp, current_git_branch,
                max_turn_chars, "turn",
            )
            if turn:
                turns.append(turn)

    return turns, current_offset


def _build_turn(
    user_text: str,
    assistant_texts: List[str],
    session_id: str,
    transcript_file: str,
    start_byte: int,
    timestamp: str,
    git_branch: str,
    max_chars: int,
    chunk_type: str,
) -> Optional[Dict]:
    """Build a turn dict from user + assistant text.

    doc_id uses a content hash (SHA-256 of byte position + text) to guarantee
    uniqueness across incremental parses. This avoids the turn_index reset bug
    where subsequent parse batches would collide with earlier ones.

    turn_index uses the byte offset where the user message starts. This is
    naturally monotonic across incremental parses (later turns always have
    higher byte offsets), making get_turns context browsing work correctly.
    """
    parts = [f"User: {user_text}"]
    if assistant_texts:
        combined_assistant = "\n\n".join(assistant_texts)
        parts.append(f"Assistant: {combined_assistant}")

    text = "\n\n".join(parts)

    # Truncate if too long
    if len(text) > max_chars:
        text = text[:max_chars] + "\n\n[truncated]"

    # Skip very short turns (likely just whitespace or newlines)
    if len(text.strip()) < 20:
        return None

    # Content-addressed doc_id: hash of byte position + text.
    # - Byte position ensures identical text at different positions gets unique IDs
    # - Content ensures re-indexing the same position produces the same ID (idempotent)
    content_hash = hashlib.sha256(
        f"{start_byte}:{text}".encode()
    ).hexdigest()[:16]

    return {
        "text": text,
        "doc_id": f"{session_id}::{content_hash}",
        "session_id": session_id,
        "transcript_file": transcript_file,
        "turn_index": start_byte,
        "timestamp": timestamp,
        "git_branch": git_branch,
        "chunk_type": chunk_type,
    }


# --- Index state management ---

_STATE_DIR = Path.home() / ".session-rag"
_STATE_PATH = _STATE_DIR / "index_state.json"
_migrated = False


def _migrate_per_project_states(state: Dict):
    """One-time migration: merge per-project index_state.json files into the global state."""
    global _migrated
    if _migrated:
        return
    _migrated = True

    # Known per-project state locations
    claude_projects = Path.home() / ".claude" / "projects"
    if not claude_projects.is_dir():
        return

    # Scan for any .session-rag/index_state.json under common project roots
    home = Path.home()
    candidates = []
    for idea_dir in [home / "IdeaProjects", home / "git-repos"]:
        if idea_dir.is_dir():
            for project_dir in idea_dir.iterdir():
                state_file = project_dir / ".session-rag" / "index_state.json"
                if state_file.exists():
                    candidates.append((str(project_dir), state_file))

    if not candidates:
        return

    transcripts = state.setdefault("transcripts", {})
    merged_count = 0

    for project_root, state_file in candidates:
        try:
            with open(state_file) as f:
                old_state = json.load(f)
        except (json.JSONDecodeError, IOError):
            continue

        old_transcripts = old_state.get("transcripts", {})
        for tpath, tdata in old_transcripts.items():
            if tpath not in transcripts:
                # Add project_root to the migrated entry
                entry = dict(tdata)
                entry["project_root"] = project_root
                transcripts[tpath] = entry
                merged_count += 1

        # Preserve the latest expire check
        old_expire = old_state.get("last_expire_check", 0)
        if old_expire > state.get("last_expire_check", 0):
            state["last_expire_check"] = old_expire

    if merged_count:
        import sys
        print(f"[state] Migrated {merged_count} transcript entries from "
              f"{len(candidates)} per-project states", file=sys.stderr)


def load_index_state() -> Dict:
    """Load centralized index state from ~/.session-rag/index_state.json."""
    state = {}
    if _STATE_PATH.exists():
        try:
            with open(_STATE_PATH) as f:
                state = json.load(f)
        except (json.JSONDecodeError, IOError):
            state = {}

    _migrate_per_project_states(state)
    return state


def save_index_state(state: Dict):
    """Save centralized index state to ~/.session-rag/index_state.json."""
    _STATE_DIR.mkdir(parents=True, exist_ok=True)
    with open(_STATE_PATH, "w") as f:
        json.dump(state, f, indent=2)


def get_transcript_offset(state: Dict, transcript_path: str) -> int:
    """Get the last indexed byte offset for a transcript file."""
    return state.get("transcripts", {}).get(transcript_path, {}).get("last_byte_offset", 0)


def set_transcript_offset(state: Dict, transcript_path: str, offset: int,
                          project_root: str = ""):
    """Update the byte offset for a transcript file."""
    if "transcripts" not in state:
        state["transcripts"] = {}
    if transcript_path not in state["transcripts"]:
        state["transcripts"][transcript_path] = {}
    state["transcripts"][transcript_path]["last_byte_offset"] = offset
    if project_root:
        state["transcripts"][transcript_path]["project_root"] = project_root