Claude Session Manager

interface:
  display_name: "Claude Session Manager"
  short_description: "List and export Claude Code JSONL sessions"
  default_prompt: "Use $claude-session-manager to list, inspect, or export Claude Code session JSONL transcripts as readable Markdown."

#!/usr/bin/env python3
"""List and export Claude Code JSONL session transcripts."""

from __future__ import annotations

import argparse
import datetime as dt
import json
from pathlib import Path
from typing import Any


DEFAULT_SOURCE = Path("~/.claude/projects").expanduser()
DEFAULT_OUTPUT = Path("~/.claude/session-markdown").expanduser()

# Cap any single tool input/result payload written to the details file. The raw,
# untruncated payload always remains in the source JSONL (linked from the digest).
TOOL_DETAIL_LIMIT = 8000


def main() -> int:
    parser = argparse.ArgumentParser(
        description="List and export Claude Code JSONL sessions to organized Markdown."
    )
    parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE)
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument(
        "--mode",
        choices=("full", "specific"),
        default="full",
        help="full exports all matching sessions; specific lists or exports a chosen session.",
    )
    parser.add_argument(
        "--list-candidates",
        action="store_true",
        help="Print a numbered candidate list instead of exporting.",
    )
    parser.add_argument(
        "--pick",
        type=int,
        help="In specific mode, export the Nth candidate from the numbered candidate list.",
    )
    parser.add_argument("--project", help="Only export project keys containing this text.")
    parser.add_argument("--session", help="Only export session ids or filenames containing this text.")
    parser.add_argument("--since", help="Only export files modified on or after YYYY-MM-DD.")
    parser.add_argument("--limit", type=int, help="Export the N most recently modified matches.")
    parser.add_argument(
        "--include-tool-details-inline",
        action="store_true",
        help="Embed full tool inputs and results in the main Markdown instead of compact details.",
    )
    args = parser.parse_args()

    source = args.source.expanduser().resolve()
    output = args.output.expanduser().resolve()
    since = parse_since(args.since)

    if not source.exists():
        raise SystemExit(f"Source folder does not exist: {source}")

    sessions = find_sessions(source, args.project, args.session, since)
    sessions = sorted(sessions, key=lambda item: item.stat().st_mtime, reverse=True)
    if args.limit:
        sessions = sessions[: args.limit]

    if args.mode == "specific" and (args.list_candidates or not args.pick):
        print_candidates(sessions, source)
        return 0

    if args.mode == "specific":
        if args.pick < 1 or args.pick > len(sessions):
            raise SystemExit(f"--pick must be between 1 and {len(sessions)}")
        sessions = [sessions[args.pick - 1]]
    else:
        sessions = sorted(sessions)

    output.mkdir(parents=True, exist_ok=True)
    exports: list[dict[str, Any]] = []
    warnings: list[str] = []

    for session_file in sessions:
        try:
            export = export_session(session_file, source, output, args.include_tool_details_inline)
            exports.append(export)
        except Exception as exc:  # Keep batch exports useful even if one transcript is odd.
            warnings.append(f"{session_file}: {exc}")

    write_root_index(output, exports, source)

    print(f"Source: {source}")
    print(f"Output: {output}")
    print(f"Exported sessions: {len(exports)}")
    if exports:
        # Print the actual file paths — the folder name is sanitized (leading
        # dashes stripped), so it is not always derivable from the project key.
        print("Files:")
        for export in exports[:50]:
            print(f"- {export['path']}")
        if len(exports) > 50:
            print(f"- ... {len(exports) - 50} more")
    if warnings:
        print(f"Warnings: {len(warnings)}")
        for warning in warnings[:20]:
            print(f"- {warning}")
        if len(warnings) > 20:
            print(f"- ... {len(warnings) - 20} more")
    print(f"Index: {output / 'index.md'}")
    return 0


def print_candidates(sessions: list[Path], source: Path) -> None:
    print(f"Source: {source}")
    print(f"Candidate sessions: {len(sessions)}")
    print("")
    print("| # | Modified | Project | Session | CWD | First user prompt |")
    print("|---:|---|---|---|---|---|")
    for index, path in enumerate(sessions, start=1):
        events, _warnings = read_jsonl(path)
        rel = path.relative_to(source)
        project_key = rel.parts[0] if rel.parts else path.parent.name
        session_id = detect_session_id(events, path)
        modified = dt.datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds")
        cwd = first_value(events, "cwd") or "unknown"
        prompt = first_user_prompt(events) or "unknown"
        print(
            f"| {index} | `{modified}` | {code_cell(project_key)} | "
            f"{code_cell(short_session_id(session_id))} | {code_cell(cwd)} | "
            f"{markdown_cell(prompt)} |"
        )


def parse_since(value: str | None) -> float | None:
    if not value:
        return None
    day = dt.datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=dt.timezone.utc)
    return day.timestamp()


def find_sessions(
    source: Path, project_filter: str | None, session_filter: str | None, since: float | None
) -> list[Path]:
    files = [path for path in source.rglob("*.jsonl") if path.is_file()]
    matches = []
    for path in files:
        rel = path.relative_to(source)
        project_key = rel.parts[0] if rel.parts else path.parent.name
        session_id = session_id_from_path(path)
        if project_filter and project_filter.lower() not in project_key.lower():
            continue
        if session_filter:
            text = f"{session_id} {path.name}".lower()
            if session_filter.lower() not in text:
                continue
        if since and path.stat().st_mtime < since:
            continue
        matches.append(path)
    return matches


def export_session(session_file: Path, source: Path, output: Path, inline_tools: bool) -> dict[str, Any]:
    events, warnings = read_jsonl(session_file)
    rel = session_file.relative_to(source)
    project_key = rel.parts[0] if rel.parts else session_file.parent.name
    session_id = detect_session_id(events, session_file)

    project_dir = output / safe_name(project_key)
    tool_dir = project_dir / "tool-details"
    project_dir.mkdir(parents=True, exist_ok=True)
    if not inline_tools:
        tool_dir.mkdir(parents=True, exist_ok=True)

    session_md = project_dir / f"{safe_name(session_id)}.md"
    tool_md = tool_dir / f"{safe_name(session_id)}.tools.md"

    # Where an inline `<tool_call_…>` reference should link for full details.
    # Sidecar mode points at the details file; inline mode points at a same-doc anchor.
    details_href = "" if inline_tools else f"tool-details/{tool_md.name}"

    rendered, tool_sections, stats = render_events(events, inline_tools, details_href)
    first_prompt = first_user_prompt(events)
    timestamps = [extract_timestamp(event) for event in events]
    timestamps = [stamp for stamp in timestamps if stamp]
    cwd = first_value(events, "cwd")

    digest = [
        f"# Claude Session {session_id}",
        "",
        "## Digest",
        "",
        f"- Source: `{session_file}`",
        f"- Project key: `{project_key}`",
        f"- Session id: `{session_id}`",
        f"- CWD: `{cwd or 'unknown'}`",
        f"- Started: `{min(timestamps) if timestamps else 'unknown'}`",
        f"- Ended: `{max(timestamps) if timestamps else 'unknown'}`",
        f"- Events: `{len(events)}`",
        f"- Messages: `{stats['messages']}`",
        f"- Tool uses: `{stats['tool_uses']}`",
        f"- Tool results: `{stats['tool_results']}`",
        f"- First user prompt excerpt: `{inline_code(first_prompt) if first_prompt else 'unknown'}`",
    ]
    if warnings:
        digest.append(f"- Parse warnings: `{len(warnings)}`")
    if tool_sections and not inline_tools:
        digest.append(f"- Tool details: [tool-details/{tool_md.name}](tool-details/{tool_md.name})")

    if stats.get("skipped_noise"):
        digest.append(f"- Collapsed metadata/noise events: `{stats['skipped_noise']}`")

    # In inline mode the details ride along in collapsible <details> next to each
    # call, so there is no separate section or sidecar; in sidecar mode the compact
    # references link out to tool_md.
    body = digest + ["", "## Conversation", ""] + rendered
    session_md.write_text("\n".join(body).rstrip() + "\n", encoding="utf-8")

    if tool_sections and not inline_tools:
        tool_md.write_text(
            "\n".join(
                [
                    f"# Tool Details for {session_id}",
                    "",
                    f"- Main transcript: [../{session_md.name}](../{session_md.name})",
                    "",
                ]
                + tool_sections
            ).rstrip()
            + "\n",
            encoding="utf-8",
        )

    write_project_index(project_dir, project_key)

    return {
        "project_key": project_key,
        "session_id": session_id,
        "path": session_md,
        "started": min(timestamps) if timestamps else "",
        "ended": max(timestamps) if timestamps else "",
        "mtime": session_file.stat().st_mtime,
        "tool_uses": stats["tool_uses"],
        "warnings": len(warnings),
    }


def read_jsonl(path: Path) -> tuple[list[dict[str, Any]], list[str]]:
    events = []
    warnings = []
    with path.open("r", encoding="utf-8", errors="replace") as handle:
        for line_no, line in enumerate(handle, start=1):
            if not line.strip():
                continue
            try:
                value = json.loads(line)
            except json.JSONDecodeError as exc:
                warnings.append(f"line {line_no}: {exc}")
                continue
            if isinstance(value, dict):
                events.append(value)
            else:
                warnings.append(f"line {line_no}: expected object, got {type(value).__name__}")
    return events, warnings


def render_events(
    events: list[dict[str, Any]], inline_tools: bool = False, details_href: str = ""
) -> tuple[list[str], list[str], dict[str, int]]:
    turns: list[dict[str, Any]] = []
    tool_sections: list[str] = []
    stats = {"messages": 0, "tool_uses": 0, "tool_results": 0, "skipped_noise": 0}
    pending_tool_results = collect_tool_results(events)

    for idx, event in enumerate(events, start=1):
        message = event.get("message") if isinstance(event.get("message"), dict) else {}
        role = message.get("role") or event.get("role") or event.get("type") or "event"
        timestamp = extract_timestamp(event)

        if role == "tool_use":
            stats["tool_uses"] += 1
            tool_ref = tool_ref_name(stats["tool_uses"])
            tool_id = event.get("id", f"tool-{idx}-{stats['tool_uses']}")
            name = event.get("name", "unknown_tool")
            result = pending_tool_results.get(str(tool_id))
            append_turn(turns, "Claude", timestamp, format_tool_reference(tool_ref, name, event, result, inline_tools, details_href))
            if not inline_tools:
                tool_sections.extend(tool_detail_section(tool_ref, idx, tool_id, name, event, result))
            continue

        if role == "tool_result":
            stats["tool_results"] += 1
            tool_id = event.get("tool_use_id") or event.get("id") or f"tool-result-{idx}"
            pending_tool_results[str(tool_id)] = {"block": event, "structured": event.get("toolUseResult")}
            continue

        if role not in {"user", "assistant"}:
            stats["skipped_noise"] += 1
            continue

        content = message.get("content", event.get("content", ""))
        chunks = content if isinstance(content, list) else [content]
        event_lines: list[str] = []

        for block in chunks:
            if isinstance(block, str):
                if is_noise_text(block):
                    stats["skipped_noise"] += 1
                elif block.strip():
                    event_lines.append(block)
                continue
            if not isinstance(block, dict):
                event_lines.append(format_scalar(block))
                continue

            block_type = block.get("type", "object")
            if block_type == "text":
                text = str(block.get("text", ""))
                if is_noise_text(text):
                    stats["skipped_noise"] += 1
                else:
                    event_lines.append(text)
            elif block_type == "tool_use":
                stats["tool_uses"] += 1
                tool_ref = tool_ref_name(stats["tool_uses"])
                tool_id = block.get("id", f"tool-{idx}-{stats['tool_uses']}")
                name = block.get("name", "unknown_tool")
                result = pending_tool_results.get(str(tool_id))
                event_lines.extend(format_tool_reference(tool_ref, name, block, result, inline_tools, details_href))
                if not inline_tools:
                    tool_sections.extend(tool_detail_section(tool_ref, idx, tool_id, name, block, result))
            elif block_type == "tool_result":
                stats["tool_results"] += 1
            elif block_type in {"thinking", "redacted_thinking"}:
                stats["skipped_noise"] += 1
            else:
                stats["skipped_noise"] += 1

        text = "\n\n".join(part.strip() for part in event_lines if part and part.strip())
        if role == "user" and message_has_only_tool_results(chunks):
            stats["skipped_noise"] += 1
            continue
        if text:
            speaker = "User" if role == "user" else "Claude"
            append_turn(turns, speaker, timestamp, [text])
        else:
            stats["skipped_noise"] += 1

    rendered: list[str] = []
    for turn in turns:
        rendered.extend(render_message_header(turn["speaker"], turn["timestamp"]))
        rendered.append("\n\n".join(turn["parts"]))
        rendered.append("")

    stats["messages"] = len(turns)
    return rendered, tool_sections, stats


def append_turn(turns: list[dict[str, Any]], speaker: str, timestamp: str, parts: list[str]) -> None:
    clean_parts = [normalize_message_flow_markdown(part.strip()) for part in parts if part and part.strip()]
    if not clean_parts:
        return
    if turns and turns[-1]["speaker"] == speaker:
        turns[-1]["parts"].extend(clean_parts)
        if not turns[-1]["timestamp"] and timestamp:
            turns[-1]["timestamp"] = timestamp
        return
    turns.append({"speaker": speaker, "timestamp": timestamp, "parts": clean_parts})


def render_message_header(speaker: str, timestamp: str) -> list[str]:
    prefix = format_local_timestamp(timestamp) if timestamp else "unknown-time"
    return [f"### {prefix} {speaker}:", ""]


def format_local_timestamp(timestamp: str) -> str:
    try:
        value = timestamp.replace("Z", "+00:00")
        parsed = dt.datetime.fromisoformat(value)
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=dt.datetime.now().astimezone().tzinfo)
        local = parsed.astimezone()
        return local.strftime("%m-%d %H:%M:%S")
    except ValueError:
        return timestamp


def normalize_message_flow_markdown(text: str) -> str:
    """Keep transcript message flow readable in common Markdown renderers."""
    lines = []
    in_fence = False
    for line in text.splitlines():
        stripped = line.lstrip()
        if stripped.startswith("```"):
            in_fence = not in_fence
            lines.append(line)
            continue
        if not in_fence and stripped.startswith("#"):
            marker = stripped.split(maxsplit=1)
            if marker and set(marker[0]) == {"#"} and len(marker[0]) <= 6:
                indent = line[: len(line) - len(stripped)]
                heading_text = marker[1].strip() if len(marker) > 1 else ""
                lines.append(f"{indent}**{heading_text}**" if heading_text else line)
                continue
        lines.append(line)
    return "\n".join(lines)


def collect_tool_results(events: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    """Map tool_use_id -> {"block": <flat tool_result>, "structured": <toolUseResult>}.

    The flat block is what the model saw (a string). The structured `toolUseResult`,
    when present on the same event, carries richer per-tool fields (Bash stdout/stderr,
    Read line counts, AskUserQuestion answers) that make a better rendering.
    """
    results: dict[str, dict[str, Any]] = {}
    for event in events:
        structured = event.get("toolUseResult")
        role = event.get("role") or event.get("type")
        if role == "tool_result":
            tool_id = event.get("tool_use_id") or event.get("id")
            if tool_id:
                results[str(tool_id)] = {"block": event, "structured": structured}
        message = event.get("message") if isinstance(event.get("message"), dict) else {}
        content = message.get("content", event.get("content", ""))
        chunks = content if isinstance(content, list) else [content]
        for block in chunks:
            if isinstance(block, dict) and block.get("type") == "tool_result":
                tool_id = block.get("tool_use_id") or block.get("id")
                if tool_id:
                    results[str(tool_id)] = {"block": block, "structured": structured}
    return results


def tool_ref_name(index: int) -> str:
    return f"tool_call_{index:06d}"


def format_tool_reference(
    ref: str,
    name: str,
    tool_payload: dict[str, Any],
    result_record: dict[str, Any] | None,
    inline: bool,
    details_href: str = "",
) -> list[str]:
    """One scannable line: what tool ran, what it did, what came back, where the rest is."""
    label = f"`<{ref}>` {name}"
    summary = tool_label(name, tool_payload)
    if summary:
        label += f" · {summary}"
    label += f" → {inline_result_signal(name, result_record)}"
    if details_href:
        label += f" · [details]({details_href}#{ref})"

    if not inline:
        return [label]
    return [label] + format_inline_tool_details(name, tool_payload, result_record)


def tool_label(name: str, payload: dict[str, Any]) -> str:
    """A single human-readable label for the call — the most telling input field."""
    tool_input = payload.get("input")
    if not isinstance(tool_input, dict):
        return ""
    # Per-tool preference order: the field a reader most wants to see inline.
    for key in ("description", "skill", "command", "file_path", "path", "pattern", "query", "url"):
        value = tool_input.get(key)
        if value in (None, "", [], {}):
            continue
        if key in ("file_path", "path"):
            return Path(str(value)).name
        return shorten_summary(value, 90)
    questions = tool_input.get("questions")
    if isinstance(questions, list) and questions and isinstance(questions[0], dict):
        return shorten_summary(questions[0].get("question", ""), 90)
    if tool_input:
        key = next(iter(tool_input))
        return shorten_summary(tool_input[key], 90)
    return ""


def shorten_summary(value: Any, limit: int = 120) -> str:
    if isinstance(value, (dict, list)):
        text = json.dumps(value, ensure_ascii=False, sort_keys=True)
    else:
        text = str(value)
    return shorten(text, limit)


def inline_result_signal(name: str, record: dict[str, Any] | None) -> str:
    """Smallest useful fact about the result: status, line count, error, or chosen answer."""
    if record is None:
        return "pending"
    block = record.get("block") or {}
    structured = record.get("structured")
    is_error = bool(block.get("is_error"))
    flat = block_text(block)

    # AskUserQuestion: the answer the user picked is the point.
    if isinstance(structured, dict) and isinstance(structured.get("answers"), dict):
        picked = "; ".join(str(v) for v in structured["answers"].values() if v)
        if picked:
            return shorten(picked, 120)

    # Bash: stdout/stderr split out.
    if isinstance(structured, dict) and ("stdout" in structured or "stderr" in structured):
        if structured.get("interrupted"):
            return "interrupted"
        stderr = (structured.get("stderr") or "").strip()
        stdout = structured.get("stdout") or ""
        if is_error:
            return f"error: {first_line(stderr or flat)}"
        n = count_lines(stdout)
        return f"ok · {n} line{'s' if n != 1 else ''}" if n else "ok"

    # Read: line counts.
    if isinstance(structured, dict) and isinstance(structured.get("file"), dict):
        total = structured["file"].get("totalLines") or structured["file"].get("numLines")
        return f"{total} lines" if total else "ok"

    if is_error:
        return f"error: {first_line(flat)}"

    n = count_lines(flat)
    if n <= 1:
        return shorten(flat, 80) or "ok"
    return f"ok · {n} lines"


def first_line(text: str, limit: int = 100) -> str:
    stripped = (text or "").strip()
    if not stripped:
        return ""
    return shorten(stripped.splitlines()[0], limit)


def count_lines(text: str) -> int:
    if not text:
        return 0
    return text.count("\n") + (0 if text.endswith("\n") else 1)


def render_result_detail(record: dict[str, Any] | None, hpre: str = "### ", hsuf: str = "") -> list[str]:
    """Full result body for the details file / inline block, using structured fields when present."""
    def heading(text: str) -> str:
        return f"{hpre}{text}{hsuf}"

    if record is None:
        return [heading("Result"), "", "_Result pending or not captured._", ""]

    block = record.get("block") or {}
    structured = record.get("structured")
    is_error = bool(block.get("is_error"))

    # Bash: separate stdout/stderr and surface interruption.
    if isinstance(structured, dict) and ("stdout" in structured or "stderr" in structured):
        out = [heading("Error" if is_error else "Result"), ""]
        if structured.get("interrupted"):
            out.append("- Interrupted: `true`")
            out.append("")
        stdout = structured.get("stdout") or ""
        stderr = structured.get("stderr") or ""
        out += ["**stdout**", "```", truncate_preserve_lines(stdout, TOOL_DETAIL_LIMIT) if stdout.strip() else "(empty)", "```", ""]
        if stderr.strip():
            out += ["**stderr**", "```", truncate_preserve_lines(stderr, TOOL_DETAIL_LIMIT), "```", ""]
        return out

    # Read: file metadata + content.
    if isinstance(structured, dict) and isinstance(structured.get("file"), dict):
        fobj = structured["file"]
        out = [heading("Result"), ""]
        if fobj.get("filePath"):
            out.append(f"- File: `{fobj['filePath']}`")
        if fobj.get("totalLines") is not None:
            out.append(
                f"- Lines: `{fobj.get('numLines', '?')}` shown of `{fobj['totalLines']}` "
                f"total (from line `{fobj.get('startLine', 1)}`)"
            )
        out.append("")
        content = fobj.get("content") or block_text(block)
        out += ["```", truncate_preserve_lines(content, TOOL_DETAIL_LIMIT), "```", ""]
        return out

    # AskUserQuestion: list each question and the chosen answer.
    if isinstance(structured, dict) and isinstance(structured.get("answers"), dict):
        out = [heading("Result"), ""]
        for question, answer in structured["answers"].items():
            out.append(f"- **{question}** → {answer}")
        out.append("")
        return out

    # Fallback: the flat string the model saw.
    return [
        heading("Error" if is_error else "Result"),
        "",
        "```",
        truncate_preserve_lines(block_text(block), TOOL_DETAIL_LIMIT),
        "```",
        "",
    ]


def format_inline_tool_details(name: str, tool_payload: dict[str, Any], result_record: dict[str, Any] | None) -> list[str]:
    lines = [
        "<details>",
        f"<summary><strong>Tool:</strong> {name}</summary>",
        "",
    ]
    tool_input = tool_payload.get("input")
    if tool_input not in (None, {}, ""):
        lines.extend(
            [
                "**Input:**",
                "```json",
                truncate_preserve_lines(json.dumps(tool_input, ensure_ascii=False, indent=2, sort_keys=True), TOOL_DETAIL_LIMIT),
                "```",
                "",
            ]
        )
    lines.extend(render_result_detail(result_record, hpre="**", hsuf="**"))
    lines.extend(["</details>", ""])
    return lines


def message_has_only_tool_results(chunks: list[Any]) -> bool:
    if not chunks:
        return False
    return all(isinstance(block, dict) and block.get("type") == "tool_result" for block in chunks)


def tool_detail_section(
    ref: str,
    event_index: int,
    tool_id: Any,
    name: str,
    payload: dict[str, Any],
    result_record: dict[str, Any] | None,
) -> list[str]:
    lines = [
        f'<a id="{ref}"></a>',
        f"## `<{ref}>` {name}",
        "",
        f"- Event: `{event_index}`",
        f"- Id: `{tool_id or 'unknown'}`",
        f"- Tool: `{name}`",
        "",
        "### Input",
        "",
        "```json",
        truncate_preserve_lines(json.dumps(payload.get("input", payload), ensure_ascii=False, indent=2, sort_keys=True), TOOL_DETAIL_LIMIT),
        "```",
        "",
    ]
    lines.extend(render_result_detail(result_record))
    return lines


def block_text(block: dict[str, Any]) -> str:
    value = block.get("content", block.get("text", block.get("input", block)))
    if isinstance(value, str):
        return value
    return json.dumps(value, ensure_ascii=False, sort_keys=True)


def first_user_prompt(events: list[dict[str, Any]]) -> str:
    for event in events:
        message = event.get("message") if isinstance(event.get("message"), dict) else {}
        role = message.get("role") or event.get("role") or event.get("type")
        if role != "user":
            continue
        content = message.get("content", event.get("content", ""))
        text = extract_text(content)
        if text and not is_noise_text(text):
            return shorten(text, 500)
    return ""


def extract_text(content: Any) -> str:
    if isinstance(content, str):
        return content.strip()
    if isinstance(content, list):
        parts = []
        for block in content:
            if isinstance(block, str):
                parts.append(block)
            elif isinstance(block, dict) and block.get("type") == "text":
                parts.append(str(block.get("text", "")))
        return "\n".join(part.strip() for part in parts if part and part.strip())
    return ""


def is_noise_text(text: str) -> bool:
    stripped = text.strip()
    if not stripped:
        return True
    noise_prefixes = (
        "<local-command-caveat>",
        "<command-name>",
        "<local-command-stdout>",
        "<local-command-stderr>",
    )
    if stripped.startswith(noise_prefixes):
        return True
    if stripped.startswith("Base directory for this skill:"):
        return True
    if stripped.startswith("The file ") and " has been updated successfully" in stripped:
        return True
    if stripped.startswith("Launching skill:"):
        return True
    return False


def extract_timestamp(event: dict[str, Any]) -> str:
    value = event.get("timestamp") or event.get("created_at") or event.get("createdAt")
    return str(value) if value else ""


def first_value(events: list[dict[str, Any]], key: str) -> Any:
    for event in events:
        if event.get(key):
            return event[key]
    return None


def detect_session_id(events: list[dict[str, Any]], path: Path) -> str:
    for event in events:
        for key in ("sessionId", "session_id", "sessionID"):
            if event.get(key):
                return str(event[key])
    return session_id_from_path(path)


def session_id_from_path(path: Path) -> str:
    if path.name == "transcript.jsonl":
        return path.parent.name
    return path.stem


def write_project_index(project_dir: Path, project_key: str) -> None:
    sessions = sorted(project_dir.glob("*.md"))
    rows = [f"# {project_key}", "", "| Session | Modified |", "|---|---|"]
    for path in sessions:
        if path.name == "index.md":
            continue
        modified = dt.datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds")
        rows.append(f"| [{path.stem}]({path.name}) | `{modified}` |")
    (project_dir / "index.md").write_text("\n".join(rows).rstrip() + "\n", encoding="utf-8")


def write_root_index(output: Path, exports: list[dict[str, Any]], source: Path) -> None:
    grouped: dict[str, list[dict[str, Any]]] = {}
    for export in exports:
        grouped.setdefault(export["project_key"], []).append(export)

    rows = [
        "# Claude Session Markdown Export",
        "",
        f"- Source: `{source}`",
        f"- Generated: `{dt.datetime.now().isoformat(timespec='seconds')}`",
        f"- Sessions exported: `{len(exports)}`",
        "",
        "## Projects",
        "",
        "| Project | Sessions | Latest exported session |",
        "|---|---:|---|",
    ]
    for project_key in sorted(grouped):
        items = sorted(grouped[project_key], key=lambda item: item.get("ended") or item.get("started") or "")
        latest = items[-1]
        project_folder = safe_name(project_key)
        latest_link = f"{project_folder}/{safe_name(latest['session_id'])}.md"
        rows.append(
            f"| [{project_key}]({project_folder}/index.md) | {len(items)} | "
            f"[{latest['session_id']}]({latest_link}) |"
        )
    (output / "index.md").write_text("\n".join(rows).rstrip() + "\n", encoding="utf-8")


def inline_code(text: str) -> str:
    return shorten(text, 500).replace("`", "'")


def markdown_cell(text: str) -> str:
    cleaned = shorten(text, 180).replace("|", "\\|").replace("\n", " ")
    return cleaned or "unknown"


def code_cell(text: str) -> str:
    return f"`{markdown_cell(inline_code(text))}`"


def short_session_id(session_id: str) -> str:
    if len(session_id) <= 18:
        return session_id
    return f"{session_id[:8]}...{session_id[-6:]}"


def shorten(text: str, limit: int) -> str:
    collapsed = " ".join(str(text).split())
    if len(collapsed) <= limit:
        return collapsed
    return collapsed[: limit - 1].rstrip() + "..."


def truncate_preserve_lines(text: str, limit: int) -> str:
    if len(text) <= limit:
        return text
    kept = text[:limit].rstrip()
    return f"{kept}\n... (truncated {len(text) - limit} of {len(text)} chars — full payload in source JSONL)"


def format_scalar(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, (int, float, bool)):
        return str(value)
    return json.dumps(value, ensure_ascii=False, sort_keys=True)


def safe_name(value: Any) -> str:
    text = str(value).strip() or "unknown"
    safe = []
    for char in text:
        if char.isalnum() or char in "-_.":
            safe.append(char)
        else:
            safe.append("-")
    result = "".join(safe).strip(".-")
    return result[:180] or "unknown"


if __name__ == "__main__":
    raise SystemExit(main())