Function parse_jsonl_file

scanner.py:317–449

Parse a JSONL file and return (session_metas, turns, agents, line_count). Deduplicates streaming events by message.id — Claude Code logs multiple JSONL records per API response, all sharing the same message.id. Only the last record per message_id is kept (it has the final usage tallies)

(filepath)

def parse_jsonl_file(filepath):
    """Parse a JSONL file and return (session_metas, turns, agents, line_count).

    Deduplicates streaming events by message.id — Claude Code logs multiple
    JSONL records per API response, all sharing the same message.id. Only the
    last record per message_id is kept (it has the final usage tallies).
    """
    seen_messages = {}  # message_id -> turn dict (dedup streaming records)
    turns_no_id = []    # turns without a message_id (kept as-is)
    session_meta = {}   # session_id -> dict
    agents = {}         # agent_id -> dispatch dict
    line_count = 0

    try:
        with open(filepath, encoding="utf-8", errors="replace") as f:
            for line_count, line in enumerate(f, 1):
                line = line.strip()
                if not line:
                    continue
                try:
                    record = json.loads(line)
                except json.JSONDecodeError:
                    continue

                rtype = record.get("type")
                if rtype not in ("assistant", "user", "custom-title", "ai-title"):
                    continue

                session_id = record.get("sessionId")
                if not session_id:
                    continue

                # Extract session title from title records
                title = _extract_title(record)
                if title:
                    if session_id not in session_meta:
                        session_meta[session_id] = {
                            "session_id": session_id,
                            "project_name": "unknown",
                            "first_timestamp": "",
                            "last_timestamp": "",
                            "git_branch": "",
                            "model": None,
                            "topic": None,
                        }
                    meta = session_meta[session_id]
                    # custom-title always wins; ai-title only if no custom-title set
                    if rtype == "custom-title":
                        meta["topic"] = title
                    elif rtype == "ai-title" and not meta.get("topic"):
                        meta["topic"] = title
                    continue

                if rtype == "user":
                    dispatch = extract_agent_dispatch(record)
                    if dispatch is not None:
                        agents[dispatch["agent_id"]] = dispatch