daymade__claude-code-skills/daymade-claude-code/read-claude-code-history/scripts/analyze_sessions.py

#!/usr/bin/env python3
"""
Analyze Claude Code session files to find relevant sessions and statistics.

This script helps locate sessions containing specific keywords, analyze
session activity, and generate reports about session content.

By default, history is searched across every active Claude config home plus
every long-term archive registered in ~/.claude/history-sources.json. Searching
only ~/.claude or only the current active tree can silently miss a real session.
Conversation dates come from internal JSONL records, never file mtime.

Three opt-in widenings exist because "not found" is the expensive answer:
--all-projects sweeps every project when the project is a guess, --codex
also searches Codex rollout history (~/.codex), and --kimi also searches
Kimi CLI sessions (~/.kimi-code) — different stores that the Claude registry
never covers.
"""

import hashlib
import json
import os
import re
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
from collections import Counter, defaultdict


# Multi-home discovery lives in the bundled `_core` package — the single source
# of truth is daymade-claude-code/_conversation_core/, copied here into
# scripts/_core/ by sync_core.py so this skill stays self-contained. Make this
# script's own dir importable regardless of how it is invoked, then import.
sys.path.insert(0, str(Path(__file__).resolve().parent))
from _core.claude import peek_final_claude_session_id, scan_claude_session  # noqa: E402
from _core.codex import codex_meta_from_rollout, codex_session_id  # noqa: E402
from _core.kimi import (  # noqa: E402
    iter_kimi_session_dirs,
    kimi_wire_files,
    kimi_wire_time_range,
    load_kimi_state,
    resolve_kimi_home,
    scan_kimi_session,
)  # noqa: E402
from _core.homes import home_label  # noqa: E402
from _core.parse import (  # noqa: E402
    TimestampRange,
    format_timestamp,
    parse_date_boundary,
    parse_timestamp,
    range_overlaps_window,
    timestamp_in_window,
    workspace_matches,
)
from _core.sources import (  # noqa: E402
    HistorySource,
    HistorySourceConfigError,
    discover_claude_sources,
)
from _core.text import (  # noqa: E402
    SearchSegment,
    extract_text,
    files_possibly_matching,
    is_automated_title,
    is_claude_agent_prompt_record,
    iter_jsonl,
    keywords_are_raw_byte_safe,
    searchable_segments,
)


CODEX_EXACT_SESSION_ID_RE = re.compile(
    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
    re.IGNORECASE,
)
DEFAULT_CODEX_SCAN_BUDGET_SECONDS = 300.0
CODEX_PROGRESS_INTERVAL_SECONDS = 15.0


class CodexScanBudgetExceeded(RuntimeError):
    """Raised instead of silently returning an incomplete broad-search result."""

    def __init__(
        self, scanned: int, total: Optional[int], elapsed: float, stage: str
    ):
        self.scanned = scanned
        self.total = total
        self.elapsed = elapsed
        self.stage = stage
        total_text = "?" if total is None else str(total)
        super().__init__(
            f"Codex scan exceeded its time budget during {stage}: "
            f"{scanned}/{total_text} rollouts inspected in {elapsed:.1f}s"
        )


class CodexScanIncomplete(RuntimeError):
    """One or more rollout files could not be parsed completely."""

    def __init__(self, paths: List[Path]):
        self.paths = list(dict.fromkeys(paths))
        super().__init__(
            f"{len(self.paths)} Codex rollout(s) could not be read completely"
        )


def _record_identity(record: Dict[str, Any]) -> str:
    """Return a stable identity for record-level union across session copies."""
    canonical = json.dumps(
        record,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return "sha256:" + hashlib.sha256(canonical).hexdigest()


# ---------------------------------------------------------------------------
# Session tail classification (`triage` command) — see references/
# session_file_format.md "Detect Session Interruption" and "Tool Use / Tool
# Result Ordering" for the reasoning this implements.
# ---------------------------------------------------------------------------

# Structural terminal states. This is deliberately NOT a judgment about
# whether a human reply is expected — "done" only means the last assistant
# turn produced a normal text block, which is equally true of a session that
# fully wrapped up and one that surfaced a finding and is waiting for a
# response. Reading last_assistant_text is the caller's job.
TAIL_INTERRUPTED_EXPLICIT = "interrupted_explicit"
TAIL_NET_ERROR = "net_error"
TAIL_STUCK_NO_RESULT = "stuck_no_result"
TAIL_DONE = "done"
TAIL_EMPTY = "empty"

_INTERRUPTED_MARKER = "[Request interrupted by user"
_NET_ERROR_PREFIX = "API Error"


@dataclass
class SessionTail:
    """Structural classification of how a session's final turn ended."""

    kind: str
    last_user_text: str
    last_assistant_kind: str  # "text" | "tool_use" | "thinking_only" | "none"
    last_assistant_text: str
    last_assistant_timestamp: Optional[float]


def classify_session_tail(path: Path) -> SessionTail:
    """Classify a session's ending state by streaming its records once.

    Resolves tool_use/tool_result across the *whole* file as a true
    set-difference, not an incremental add/discard in file order: a
    tool_result can be written before the tool_use record it answers (see
    "Tool Use / Tool Result Ordering" in references/session_file_format.md).
    A single-pass ``discard-then-add`` was tried first and is NOT actually
    order-independent — ``discard()`` on an id not yet seen is a silent
    no-op, so a tool_result appearing before its tool_use left the id
    "pending" even though it was genuinely resolved (verified against real
    session data, 2026-08: 11/14 files hitting this ordering had their
    final `kind` flipped). Accumulating two never-mutated sets and diffing
    them once at the end is immune to this, because neither operation can
    ever silently miss the other regardless of which came first in the file.

    Classification is also computed from the RAW content of the LAST
    assistant record only, not from state that could carry over from an
    earlier turn: a final turn that produces no text and no tool_use (e.g.
    thinking-only) previously left `last_assistant_kind`/`last_assistant_text`
    holding an earlier turn's already-answered reply, misreporting a session
    that crashed before responding to its latest question as `done`.

    The interruption marker is checked the same way: `tail_is_interrupt` is
    reset by any later user or assistant record, so it only survives to the
    end of the loop when the marker is the LAST relevant record in the file.
    A mid-session Ctrl+C that the conversation continued past is not a tail
    interruption — treating "marker appears anywhere" as equivalent to "the
    session ended on interruption" was tried first and false-positived on
    exactly that shape (verified against real session data, 2026-08).
    """
    all_tool_use_ids: set = set()
    all_resolved_tool_use_ids: set = set()
    last_user_text = ""
    last_assistant_content: Any = None
    last_assistant_timestamp: Optional[float] = None
    tail_is_interrupt = False

    for record in iter_jsonl(path):
        record_type = record.get("type")
        message = record.get("message")
        content = message.get("content") if isinstance(message, dict) else None

        if record_type == "user" and not record.get("isMeta"):
            if isinstance(content, str) and _INTERRUPTED_MARKER in content:
                tail_is_interrupt = True
                continue
            tail_is_interrupt = False
            if isinstance(content, str):
                last_user_text = content
            elif isinstance(content, list):
                is_tool_result_only = bool(content) and all(
                    isinstance(block, dict) and block.get("type") == "tool_result"
                    for block in content
                )
                for block in content:
                    if not isinstance(block, dict):
                        continue
                    if block.get("type") == "tool_result":
                        tool_use_id = block.get("tool_use_id")
                        if tool_use_id is not None:
                            all_resolved_tool_use_ids.add(tool_use_id)
                if not is_tool_result_only:
                    text = extract_text(content)
                    if text:
                        last_user_text = text

        elif record_type == "assistant":
            tail_is_interrupt = False
            parsed_ts = parse_timestamp(record.get("timestamp"))
            if parsed_ts is not None:
                last_assistant_timestamp = parsed_ts
            last_assistant_content = content
            if isinstance(content, list):
                for block in content:
                    if isinstance(block, dict) and block.get("type") == "tool_use":
                        tool_use_id = block.get("id")
                        if tool_use_id is not None:
                            all_tool_use_ids.add(tool_use_id)

    pending_tool_use_ids = all_tool_use_ids - all_resolved_tool_use_ids

    last_assistant_kind = "none"
    last_assistant_text = ""
    final_turn_has_pending_tool = False
    if isinstance(last_assistant_content, list):
        text_blocks = [
            block.get("text", "")
            for block in last_assistant_content
            if isinstance(block, dict) and block.get("type") == "text"
        ]
        tool_blocks = [
            block
            for block in last_assistant_content
            if isinstance(block, dict) and block.get("type") == "tool_use"
        ]
        joined_text = "\n".join(part for part in text_blocks if part)
        final_turn_has_pending_tool = any(
            block.get("id") in pending_tool_use_ids for block in tool_blocks
        )
        if final_turn_has_pending_tool:
            last_assistant_kind = "tool_use"
            pending_names = [
                block.get("name", "?")
                for block in tool_blocks
                if block.get("id") in pending_tool_use_ids
            ]
            last_assistant_text = f"[tool_use:{pending_names[-1]}]"
        elif joined_text:
            last_assistant_kind = "text"
            last_assistant_text = joined_text
        elif tool_blocks:
            # Every tool_use in this turn already has a matching result
            # elsewhere in the file, but no further assistant reply followed
            # it — the harness likely stopped between the tool result
            # landing and the model's next turn being captured. Distinct
            # from `final_turn_has_pending_tool`: nothing is unresolved, but
            # nothing was said either, so this is not a normal `done` reply.
            last_assistant_kind = "tool_use"
            last_assistant_text = f"[tool_use:{tool_blocks[-1].get('name', '?')}] (resolved, no further reply)"
        else:
            last_assistant_kind = "thinking_only"
            last_assistant_text = (
                "(final assistant turn has no text or tool_use block — "
                "thinking-only or empty content)"
            )
    elif isinstance(last_assistant_content, str) and last_assistant_content:
        last_assistant_kind = "text"
        last_assistant_text = last_assistant_content

    if tail_is_interrupt:
        kind = TAIL_INTERRUPTED_EXPLICIT
    elif last_assistant_kind == "none":
        kind = TAIL_EMPTY
    elif last_assistant_kind == "text" and last_assistant_text.startswith(
        _NET_ERROR_PREFIX
    ):
        kind = TAIL_NET_ERROR
    elif last_assistant_kind == "text":
        kind = TAIL_DONE
    else:
        # tool_use (pending or just-resolved-with-no-followup) and
        # thinking_only all mean the same thing for triage purposes: the
        # final turn produced no textual reply, so the model was still
        # working when the file stopped.
        kind = TAIL_STUCK_NO_RESULT

    return SessionTail(
        kind=kind,
        last_user_text=last_user_text,
        last_assistant_kind=last_assistant_kind,
        last_assistant_text=last_assistant_text,
        last_assistant_timestamp=last_assistant_timestamp,
    )


# ---------------------------------------------------------------------------
# Codex rollout search (--codex)
#
# Codex stores conversations outside the Claude history registry: rollout
# JSONL files under <CODEX_HOME>/sessions/<YYYY>/<MM>/<DD>/ plus
# <CODEX_HOME>/archived_sessions/. Their record schema is NOT the Claude one
# (response_item/event_msg/session_meta, not user/assistant/queue-operation),
# so searchable_segments() does not apply. The extractor below covers the
# user-visible payload of each response_item variant. event_msg user/agent
# message records are deliberate strict mirrors of response_item message text
# (verified 2026-07-16: 26/26 and 104/104 subset on a real rollout), so they
# are skipped to avoid double-counting.
# ---------------------------------------------------------------------------


def _flatten_strings(value: Any) -> List[str]:
    """Flatten nested str/list/dict content into plain strings."""
    if isinstance(value, str):
        return [value]
    if isinstance(value, list):
        return [part for item in value for part in _flatten_strings(item)]
    if isinstance(value, dict):
        return [part for item in value.values() for part in _flatten_strings(item)]
    return []


def codex_searchable_segments(record: Dict[str, Any]) -> List[SearchSegment]:
    """Extract searchable text fields from one Codex rollout record."""
    segments: List[SearchSegment] = []

    def add(source: str, value: Any) -> None:
        for text_value in _flatten_strings(value):
            segments.append(SearchSegment(source=source, text=text_value))

    if record.get("type") == "compacted":
        # Compaction records carry a summary of earlier conversation content.
        payload = record.get("payload")
        if isinstance(payload, dict):
            add("summary", payload.get("message"))
        return list(dict.fromkeys(segments))

    if record.get("type") != "response_item":
        return segments
    payload = record.get("payload")
    if not isinstance(payload, dict):
        return segments
    payload_type = payload.get("type")
    if payload_type == "message":
        for block in payload.get("content") or []:
            if isinstance(block, dict) and block.get("type") in {
                "input_text",
                "output_text",
            }:
                add("message", block.get("text"))
    elif payload_type == "reasoning":
        for block in payload.get("summary") or []:
            if isinstance(block, dict) and block.get("type") == "summary_text":
                add("thinking", block.get("text"))
    elif payload_type in {"function_call", "custom_tool_call"}:
        name = payload.get("name")
        source = (
            f"tool_input:{name}"
            if isinstance(name, str) and name
            else "tool_input"
        )
        add(source, payload.get("arguments"))
        add(source, payload.get("input"))
    elif payload_type in {"function_call_output", "custom_tool_call_output"}:
        add("tool_result", payload.get("output"))
    return list(dict.fromkeys(segments))


def discover_codex_rollouts(
    codex_home: Path,
    *,
    on_progress: Optional[Callable[[int], None]] = None,
) -> List[Path]:
    """Enumerate Codex rollout files under a Codex home (sessions + archived)."""
    rollouts: List[Path] = []
    sessions_dir = codex_home / "sessions"
    if sessions_dir.is_dir():
        for path in sessions_dir.rglob("*.jsonl"):
            rollouts.append(path)
            if on_progress is not None and len(rollouts) % 256 == 0:
                on_progress(len(rollouts))
    archived_dir = codex_home / "archived_sessions"
    if archived_dir.is_dir():
        for path in archived_dir.glob("*.jsonl"):
            rollouts.append(path)
            if on_progress is not None and len(rollouts) % 256 == 0:
                on_progress(len(rollouts))
    if on_progress is not None:
        on_progress(len(rollouts))
    return sorted(rollouts)


def locate_codex_rollouts(codex_home: Path, session_id: str) -> List[Dict[str, Any]]:
    """Locate one exact Codex session without parsing the rollout corpus.

    Current Codex rollout filenames carry the session UUID. Globbing that
    exact UUID touches directory entries only, then each small candidate is
    validated against the authoritative ``session_meta.id`` before it is
    returned. A UUID is metadata, not conversation text; scanning multi-GB
    rollout bodies for metadata is the failure mode this function prevents.
    """
    normalized = session_id.strip().lower()
    if not CODEX_EXACT_SESSION_ID_RE.fullmatch(normalized):
        raise ValueError(f"invalid Codex session id: {session_id!r}")

    candidates: set[Path] = set()
    sessions_dir = codex_home / "sessions"
    if sessions_dir.is_dir():
        candidates.update(
            sessions_dir.glob(f"*/*/*/rollout-*-{normalized}.jsonl")
        )
    archived_dir = codex_home / "archived_sessions"
    if archived_dir.is_dir():
        candidates.update(archived_dir.glob(f"rollout-*-{normalized}.jsonl"))

    located: List[Dict[str, Any]] = []
    for path in sorted(candidates):
        meta = codex_meta_from_rollout(path, strict=True)
        if not isinstance(meta, dict):
            continue
        authoritative_id = meta.get("id")
        if (
            not isinstance(authoritative_id, str)
            or not authoritative_id.strip()
            or authoritative_id.lower() != normalized
        ):
            continue
        located.append(
            {
                "session_id": authoritative_id,
                "path": path,
                "cwd": meta.get("cwd"),
                "timestamp": parse_timestamp(meta.get("timestamp")),
                "storage": "archived" if archived_dir in path.parents else "active",
            }
        )
    located.sort(key=lambda item: (item["storage"] == "archived", str(item["path"])))
    return located


def _print_codex_locations(codex_home: Path, session_id: str) -> int:
    try:
        locations = locate_codex_rollouts(codex_home, session_id)
    except (ValueError, OSError, UnicodeError, json.JSONDecodeError) as error:
        print(f"Codex session lookup was incomplete: {error}", file=sys.stderr)
        return 2
    if not locations:
        print(
            f"No Codex rollout with session_meta.id={session_id} under {codex_home}.",
            file=sys.stderr,
        )
        return 1
    print(
        f"Codex session {locations[0]['session_id']} "
        f"({len(locations)} physical copy/copies):"
    )
    for item in locations:
        print(f"  - Storage: {item['storage']}")
        if isinstance(item.get("cwd"), str) and item["cwd"]:
            print(f"    cwd: {item['cwd']}")
        if item.get("timestamp") is not None:
            print(f"    Created: {format_timestamp(item['timestamp'])}")
        print(f"    Path: {item['path']}")
    return 0


def file_content_digest(path: Path) -> Optional[str]:
    """SHA-256 over a file's bytes, or ``None`` when it cannot be read.

    ``None`` means "no information" and must make the caller fall through to
    the normal full parse — never to treating two files as identical.
    """
    digest = hashlib.sha256()
    try:
        with open(path, "rb") as handle:
            for block in iter(lambda: handle.read(1 << 20), b""):
                digest.update(block)
    except (OSError, ValueError):
        return None
    return digest.hexdigest()


def repeated_copy_sizes(copies: List[Dict[str, Any]]) -> set:
    """Byte sizes that occur more than once among a session's copies.

    Only these are worth hashing: a size that appears once cannot have a
    byte-identical twin, so single-copy sessions pay nothing for the
    duplicate-copy optimisation.
    """
    sizes = Counter()
    for item in copies:
        try:
            sizes[item["path"].stat().st_size] += 1
        except OSError:
            continue
    return {size for size, count in sizes.items() if count > 1}


def search_codex_rollouts(
    rollouts: List[Path],
    keywords: List[str],
    case_sensitive: bool = False,
    from_timestamp: Optional[float] = None,
    to_timestamp: Optional[float] = None,
    project_path: Optional[str] = None,
    exclude_ids: Optional[set] = None,
    use_prefilter: bool = True,
    max_scan_seconds: Optional[float] = DEFAULT_CODEX_SCAN_BUDGET_SECONDS,
    progress_interval_seconds: float = CODEX_PROGRESS_INTERVAL_SECONDS,
    clock: Callable[[], float] = time.monotonic,
    progress: Optional[Callable[[str], None]] = None,
    started_at: Optional[float] = None,
) -> List[Dict[str, Any]]:
    """Search Codex rollouts for keywords, one match dict per session.

    ``project_path`` filters by the rollout's session_meta cwd (recursive
    workspace match). Rollouts are de-duplicated by session id — a copy left
    in both sessions/ and archived_sessions/ is reported once.

    ``use_prefilter`` skips per-file and per-line json.loads() work a raw byte
    scan already proves cannot match (see ``files_possibly_matching`` and
    ``iter_jsonl``'s ``line_keywords``). The FILE-level skip is safe
    unconditionally — every counter this function reports (``excluded_untimed``,
    ``session_range``, ``match_range``) is scoped to a single rollout file and
    only surfaces when that same file has ``total_mentions > 0``, so a file
    ruled out entirely never had anything to report.

    Codex does NOT do line-level skipping at all — not even outside a date
    window. Two separate accountings need to see every record: ``session_range``
    (which would otherwise collapse onto ``match_range``) and, under a date
    window, ``excluded_untimed``. See the comment at the ``line_keywords``
    assignment below; the disabling is unconditional and a test enforces it.
    """
    search_keywords = [
        (keyword, keyword if case_sensitive else keyword.casefold())
        for keyword in keywords
    ]
    exclude = exclude_ids or set()
    matches: List[Dict[str, Any]] = []
    seen_ids: set[str] = set()
    incomplete_paths: List[Path] = []

    started = clock() if started_at is None else started_at
    next_progress = started + progress_interval_seconds
    progress_sink = progress or (
        lambda message: print(message, file=sys.stderr, flush=True)
    )
    scanned_files = 0

    def check_liveness(stage: str) -> None:
        nonlocal next_progress
        now = clock()
        elapsed = now - started
        if (
            max_scan_seconds is not None
            and max_scan_seconds > 0
            and elapsed > max_scan_seconds
        ):
            raise CodexScanBudgetExceeded(scanned_files, len(rollouts), elapsed, stage)
        if progress_interval_seconds > 0 and now >= next_progress:
            progress_sink(
                "Codex scan progress: "
                f"stage={stage}, {scanned_files}/{len(rollouts)} rollouts, "
                f"matches={len(matches)}, elapsed={elapsed:.1f}s"
            )
            while next_progress <= now:
                next_progress += progress_interval_seconds

    matched_files: Optional[set[Path]] = None
    if use_prefilter:
        check_liveness("prefilter-start")
        deadline = None
        if max_scan_seconds is not None and max_scan_seconds > 0:
            deadline = started + max_scan_seconds
        matched_files = files_possibly_matching(
            rollouts,
            keywords,
            case_sensitive=case_sensitive,
            deadline=deadline,
            clock=clock,
            progress_interval_seconds=progress_interval_seconds,
            on_progress=lambda: check_liveness("prefilter"),
        )
        check_liveness("prefilter-complete")
    # Codex gets FILE-level pre-filtering only, never line-level.
    #
    # ``session_range.observe()`` has to see every record — including the
    # ``session_meta`` first line — to report the conversation's time span.
    # Skipping lines that lack the keyword bytes collapses ``Internal range``
    # onto ``Match range``: measured on a 4-line rollout, ``01-01 .. 01-20``
    # became ``01-10 .. 01-10``. That is not "one less field", it is a *wrong
    # value* in a field callers quote, and it fires on the plain no-date-window
    # search — the most common invocation there is.
    #
    # A file the file-level filter rules out entirely reports no range at all,
    # so that layer stays safe and stays on.
    line_keywords = None

    for path_index, path in enumerate(rollouts, start=1):
        scanned_files = path_index - 1
        check_liveness("rollout-open")
        try:
            meta = codex_meta_from_rollout(path, strict=True)
        except (OSError, UnicodeError, json.JSONDecodeError) as e:
            print(f"Warning: Error reading {path}: {e}", file=sys.stderr)
            incomplete_paths.append(path)
            continue
        sid = codex_session_id(meta, path) if meta else None
        if not sid:
            sid = path.stem
        # Dedup and exclusion happen BEFORE the pre-filter check below, on
        # purpose: they must claim/reject this path exactly as before, in the
        # same traversal order, regardless of whether it turns out to have a
        # keyword match. Moving the pre-filter check earlier would change
        # which physical copy "wins" the session id when two copies of one
        # session exist (sessions/ vs archived_sessions/) — a real behavior
        # change this performance fix has no business making silently.
        if sid in seen_ids:
            continue
        seen_ids.add(sid)
        if sid in exclude or path.stem in exclude:
            continue
        if matched_files is not None and path not in matched_files:
            continue
        cwd = meta.get("cwd") if isinstance(meta, dict) else None
        if project_path is not None:
            if not isinstance(cwd, str) or not workspace_matches(
                cwd, project_path, recursive=True
            ):
                continue

        keyword_counts: Dict[str, int] = defaultdict(int)
        total_mentions = 0
        match_sources: set[str] = set()
        session_range = TimestampRange()
        match_range = TimestampRange()
        excluded_untimed = 0
        try:
            for record_index, record in enumerate(
                iter_jsonl(path, line_keywords=line_keywords, strict=True)
            ):
                if record_index % 512 == 0:
                    check_liveness("rollout-parse")
                record_timestamp = parse_timestamp(record.get("timestamp"))
                if record_timestamp is not None:
                    session_range.observe(record_timestamp)
                if record.get("type") == "session_meta" and isinstance(
                    record.get("payload"), dict
                ):
                    meta_timestamp = parse_timestamp(
                        record["payload"].get("timestamp")
                    )
                    if meta_timestamp is not None:
                        session_range.observe(meta_timestamp)
                if from_timestamp is not None or to_timestamp is not None:
                    if record_timestamp is None:
                        excluded_untimed += 1
                        continue
                    if not timestamp_in_window(
                        record_timestamp, from_timestamp, to_timestamp
                    ):
                        continue
                record_counts: Dict[str, int] = defaultdict(int)
                record_sources: set[str] = set()
                for segment in codex_searchable_segments(record):
                    search_text = (
                        segment.text if case_sensitive else segment.text.casefold()
                    )
                    for keyword, search_keyword in search_keywords:
                        count = search_text.count(search_keyword)
                        if count > 0:
                            record_counts[keyword] += count
                            record_sources.add(segment.source)
                record_mentions = sum(record_counts.values())
                if not record_mentions:
                    continue
                for keyword, count in record_counts.items():
                    keyword_counts[keyword] += count
                total_mentions += record_mentions
                match_sources.update(record_sources)
                if record_timestamp is not None:
                    match_range.observe(record_timestamp)
        except (OSError, UnicodeError, json.JSONDecodeError) as e:
            print(f"Warning: Error processing {path}: {e}", file=sys.stderr)
            incomplete_paths.append(path)
            continue

        scanned_files = path_index
        check_liveness("rollout-complete")

        if total_mentions > 0:
            matches.append(
                {
                    "session_id": sid,
                    "path": path,
                    "cwd": cwd,
                    "total_mentions": total_mentions,
                    "keyword_counts": dict(keyword_counts),
                    "match_sources": sorted(match_sources),
                    "created_at": session_range.earliest,
                    "updated_at": session_range.latest,
                    "match_created_at": match_range.earliest,
                    "match_updated_at": match_range.latest,
                    "excluded_untimed_records": excluded_untimed,
                }
            )

    if incomplete_paths:
        raise CodexScanIncomplete(incomplete_paths)

    matches.sort(
        key=lambda match: (
            match["total_mentions"],
            match["match_updated_at"]
            if match["match_updated_at"] is not None
            else float("-inf"),
        ),
        reverse=True,
    )
    return matches


# ---------------------------------------------------------------------------
# Kimi CLI session search (--kimi)
#
# Kimi CLI (kimi-code) stores conversations under <KIMI_HOME>/sessions/
# wd_<workspace>_<hash>/session_<uuid>/agents/<agent>/wire.jsonl, with a
# per-session state.json (id/cwd/title/createdAt/updatedAt in ms). The wire
# record schema is NOT the Claude or Codex one (turn.prompt,
# context.append_message, context.append_loop_event, all timestamped by a
# `time` epoch-ms field), so neither searchable_segments() nor
# codex_searchable_segments() applies. Layout and record shapes were verified
# against Kimi CLI 0.38.0 (wire protocol 1.5) on a real store; see
# _core/kimi.py's module docstring for the full format contract.
#
# Searchable coverage is the conversation itself: user prompts (turn.prompt /
# turn.steer), appended messages (user/assistant text and attachments), and
# loop events (assistant content parts, tool calls, tool results). Static
# boilerplate — config.update / profile.bind system prompts, tool snapshots,
# usage/token metrics — is deliberately NOT indexed: a keyword that only
# appears in a shared system prompt would match every session, and "not found"
# is the answer this tool is trusted to give about conversation content.
# ---------------------------------------------------------------------------


def _kimi_flatten_payload(value: Any) -> List[str]:
    """Flatten nested Kimi payload content, dropping structural identifier keys.

    Same principle as text.py's searchable_segments excluding id/signature
    keys: UUID-class fields (turnId, toolCallId, …) are unique per record, so
    indexing them only manufactures false-positive hits and dilutes the match
    source attribution (independent review, 2026-08).
    """
    if isinstance(value, str):
        return [value]
    if isinstance(value, list):
        return [part for item in value for part in _kimi_flatten_payload(item)]
    if isinstance(value, dict):
        return [
            part
            for key, item in value.items()
            if key not in _KIMI_STRUCTURAL_KEYS
            for part in _kimi_flatten_payload(item)
        ]
    return []


_KIMI_STRUCTURAL_KEYS = frozenset(
    {
        "type",
        "id",
        "uuid",
        "stepUuid",
        "turnId",
        "toolCallId",
        "parentUuid",
        "callId",
    }
)


def kimi_searchable_segments(record: Dict[str, Any]) -> List[SearchSegment]:
    """Extract searchable text fields from one Kimi wire.jsonl record."""
    segments: List[SearchSegment] = []

    def add(source: str, value: Any) -> None:
        for text_value in _kimi_flatten_payload(value):
            segments.append(SearchSegment(source=source, text=text_value))

    record_type = record.get("type")
    if record_type in {"turn.prompt", "turn.steer"}:
        add("prompt", record.get("input"))
    elif record_type == "context.append_message":
        message = record.get("message")
        if isinstance(message, dict):
            add("message", message.get("content"))
            add("tool_input", message.get("toolCalls"))
    elif record_type == "context.append_loop_event":
        event = record.get("event")
        if isinstance(event, dict):
            event_type = event.get("type")
            if event_type == "content.part":
                add("message", event)
            elif event_type == "tool.call":
                add("tool_input", event)
            elif event_type == "tool.result":
                add("tool_result", event)
            # step.begin / step.end carry no conversation text.
    elif record_type == "plugin.session_start":
        add("plugin", record.get("content"))
    return list(dict.fromkeys(segments))


def discover_kimi_wires(kimi_home: Path) -> List[tuple]:
    """Enumerate (session_dir, agent_name, wire_path) triples under a Kimi home."""
    wires: List[tuple] = []
    for session_dir in iter_kimi_session_dirs(kimi_home):
        main_wire, subagent_wires = kimi_wire_files(session_dir)
        for path in ([main_wire] if main_wire else []) + subagent_wires:
            if path is not None:
                wires.append((session_dir, path.parent.name, path))
    return wires


def search_kimi_wires(
    wires: List[tuple],
    keywords: List[str],
    case_sensitive: bool = False,
    from_timestamp: Optional[float] = None,
    to_timestamp: Optional[float] = None,
    project_path: Optional[str] = None,
    exclude_ids: Optional[set] = None,
    use_prefilter: bool = True,
) -> List[Dict[str, Any]]:
    """Search Kimi wire files for keywords, one match dict per SESSION.

    A session's subagent wires (agents/agent-N/) are runs of the same
    conversation, so matches aggregate at session level with the agent name
    prefixed into each match source (e.g. "main:message", "agent-0:tool_input").

    Like the Codex search, Kimi gets FILE-level pre-filtering only, never
    line-level: the session's internal range must observe every record's
    `time`, and collapsing it onto the match range would report a wrong value
    in a field callers quote. One Kimi-specific subtlety: the pre-filter rules
    out whole wires, but a session's range spans ALL its wires — so when any
    wire of a session matched, the remaining wires are re-read for their time
    ranges only (no keyword work), keeping the displayed range exact.
    """
    search_keywords = [
        (keyword, keyword if case_sensitive else keyword.casefold())
        for keyword in keywords
    ]
    exclude = exclude_ids or set()

    matched_files: Optional[set] = None
    if use_prefilter:
        matched_files = files_possibly_matching(
            [wire_path for _, _, wire_path in wires],
            keywords,
            case_sensitive=case_sensitive,
        )
    line_keywords = None  # see docstring — never line-level for Kimi either

    # Group wires by session, preserving discovery order.
    sessions: Dict[Path, List[tuple]] = {}
    for session_dir, agent_name, wire_path in wires:
        sessions.setdefault(session_dir, []).append((agent_name, wire_path))

    matches: List[Dict[str, Any]] = []
    for session_dir, agent_wires in sessions.items():
        state = load_kimi_state(session_dir) or {}
        session_id = state.get("id") if isinstance(state.get("id"), str) else None
        if not session_id:
            session_id = session_dir.name
        if session_id in exclude or session_dir.name in exclude:
            continue
        if isinstance(state.get("cwd"), str) and state["cwd"].strip():
            cwd: Optional[str] = state["cwd"].strip()
        else:
            cwd = None
        if project_path is not None:
            if not cwd or not workspace_matches(cwd, project_path, recursive=True):
                continue

        keyword_counts: Dict[str, int] = defaultdict(int)
        total_mentions = 0
        match_sources: set = set()
        session_range = TimestampRange()
        match_range = TimestampRange()
        excluded_untimed = 0
        skipped_wires: List[Path] = []

        for agent_name, wire_path in agent_wires:
            if matched_files is not None and wire_path not in matched_files:
                skipped_wires.append(wire_path)
                continue
            try:
                for record in iter_jsonl(wire_path, line_keywords=line_keywords):
                    record_timestamp = parse_timestamp(record.get("time"))
                    if record_timestamp is None and record.get("type") == "metadata":
                        record_timestamp = parse_timestamp(record.get("created_at"))
                    if record_timestamp is not None:
                        session_range.observe(record_timestamp)
                    if from_timestamp is not None or to_timestamp is not None:
                        if record_timestamp is None:
                            excluded_untimed += 1
                            continue
                        if not timestamp_in_window(
                            record_timestamp, from_timestamp, to_timestamp
                        ):
                            continue
                    record_counts: Dict[str, int] = defaultdict(int)
                    record_sources: set = set()
                    for segment in kimi_searchable_segments(record):
                        search_text = (
                            segment.text
                            if case_sensitive
                            else segment.text.casefold()
                        )
                        for keyword, search_keyword in search_keywords:
                            count = search_text.count(search_keyword)
                            if count > 0:
                                record_counts[keyword] += count
                                record_sources.add(f"{agent_name}:{segment.source}")
                    record_mentions = sum(record_counts.values())
                    if not record_mentions:
                        continue
                    for keyword, count in record_counts.items():
                        keyword_counts[keyword] += count
                    total_mentions += record_mentions
                    match_sources.update(record_sources)
                    if record_timestamp is not None:
                        match_range.observe(record_timestamp)
            except (OSError, UnicodeError) as e:
                print(f"Warning: Error processing {wire_path}: {e}", file=sys.stderr)
                continue

        if total_mentions > 0:
            # A prefiltered-out wire can still hold earlier/later records of
            # this matched session; fold its time range in so the displayed
            # internal range stays exact.
            for wire_path in skipped_wires:
                try:
                    extra = kimi_wire_time_range(wire_path)
                except OSError:
                    continue
                for value in (extra.earliest, extra.latest):
                    if value is not None:
                        session_range.observe(value)
            matches.append(
                {
                    "session_id": session_id,
                    "path": session_dir,
                    "title": scan_kimi_session(session_dir).title,
                    "cwd": cwd or "",
                    "total_mentions": total_mentions,
                    "keyword_counts": dict(keyword_counts),
                    "match_sources": sorted(match_sources),
                    "created_at": session_range.earliest,
                    "updated_at": session_range.latest,
                    "match_created_at": match_range.earliest,
                    "match_updated_at": match_range.latest,
                    "excluded_untimed_records": excluded_untimed,
                }
            )

    matches.sort(
        key=lambda match: (
            match["total_mentions"],
            match["match_updated_at"]
            if match["match_updated_at"] is not None
            else float("-inf"),
        ),
        reverse=True,
    )
    return matches


class SessionAnalyzer:
    """Analyze Claude Code session history files across all config homes."""

    def __init__(
        self,
        homes: Optional[List[Path]] = None,
        sources: Optional[List[HistorySource]] = None,
        warnings: Optional[List[str]] = None,
    ):
        """
        Initialize analyzer.

        Args:
            homes: Exact list of Claude config home directories to search (each
                must
                contain a ``projects/`` subdir). Pass ``None`` (the default) to
                auto-discover active homes and load registered archives. Pass an
                explicit list to restrict the search — an EMPTY list means
                "search nothing", it must NOT silently fall back to full
                discovery (that would turn a scope-narrowing flag into the
                widest possible scope).
        """
        if homes is not None and sources is not None:
            raise ValueError("Pass homes or sources, not both")
        if sources is not None:
            self.sources = list(sources)
        elif homes is not None:
            self.sources = [
                HistorySource(
                    provider="claude",
                    kind="active",
                    label=home_label(home),
                    home=Path(home),
                )
                for home in homes
            ]
        else:
            self.sources, discovered_warnings = discover_claude_sources()
            warnings = (warnings or []) + discovered_warnings
        self.homes = [source.home for source in self.sources]
        self.warnings = list(warnings or [])

    def find_project_sessions(self, project_path: str) -> List[Dict[str, Any]]:
        """
        Find all session files for a project ACROSS every discovered home.

        Sessions are de-duplicated by session id (the ``.jsonl`` filename), so a
        conversation shared across profiles is reported once. Every physical
        copy remains attached to that session reference: search must union the
        records from active and archive copies because one copy can retain
        content that another copy no longer has. Agent side-files
        (``agent-*.jsonl``) are excluded.

        Args:
            project_path: The project's working directory. An absolute path, a
                ``~`` path, or a relative path are all expanded and resolved to
                an absolute path before encoding. A bare directory name
                (basename) is also accepted and matched via reverse lookup.

        Returns:
            Session-reference dictionaries sorted by the unioned maximum
            internal timestamp (newest first). Each includes a representative
            ``path``, every physical ``copy``, the unioned
            ``created_at``/``updated_at`` range, and all ``sources``/``homes``
            where that session ID was observed. Empty if the project has no
            history in the configured source set.
        """
        return self._merge_sessions_from_dirs(
            self._resolve_project_dirs(project_path)
        )

    def _merge_sessions_from_dirs(
        self,
        pairs: List[tuple],
        candidate_paths: Optional[set[str]] = None,
        candidate_summaries: Optional[Dict[str, Any]] = None,
    ) -> List[Dict[str, Any]]:
        """Collect + de-duplicate sessions from ``(source, project_dir)`` pairs.

        Per-model profile homes (``~/.claude-profiles/<name>``) commonly
        symlink their ``projects/`` straight to the main home's — same real
        directory, different nominal ``source.home``. ``pairs`` is built by
        the caller per source label, so the same physical file can arrive
        here once per profile that happens to alias it. Group by the
        *resolved* real path first so each physical file is opened and
        parsed exactly once no matter how many source labels point at it.

        Every group member keeps its own *nominal* (unresolved) directory
        so per-copy paths stay distinct strings — e.g. a session aliased
        under both ``~/.claude/projects`` and ``~/.claude-profiles/css/
        projects`` still reports two different copy paths, matching what
        the un-grouped code produced. Collapsing them to one shared string
        was tried first and broke ``search``'s "Other matching copies:"
        listing (independent review, 2026-08-04): it distinguishes the
        primary copy from the others by path-string inequality, so giving
        every aliased copy the identical string made all-but-one silently
        vanish from that field even though the ``sources``/``homes``
        label lists stayed correct.
        """
        groups = self._group_project_dirs(pairs)

        by_id: Dict[str, Dict[str, Any]] = {}
        for scan_dir, group_members in groups:
            group_sources = [source for source, _nominal_dir in group_members]
            # any(...) — not group_sources[0].kind — because "at least one
            # active alias in the group" must win regardless of which
            # member happened to be discovered first; index-0 only agrees
            # with that whenever active sources are enumerated before
            # archives (true for both shipped call sites today, but not a
            # contract this function should silently depend on).
            group_has_active = any(source.kind == "active" for source in group_sources)
            group_kind = "active" if group_has_active else group_sources[0].kind
            for file in scan_dir.glob("*.jsonl"):
                if file.name.startswith("agent-"):
                    continue
                try:
                    physical_file = str(file.resolve())
                except (OSError, RuntimeError):
                    physical_file = str(file)
                if candidate_paths is not None:
                    if physical_file not in candidate_paths:
                        continue
                summary = (
                    candidate_summaries.get(physical_file)
                    if candidate_summaries is not None
                    else None
                )
                if summary is None:
                    summary = scan_claude_session(file)
                sid = summary.session_id
                copies = [
                    {
                        "path": nominal_dir / file.name,
                        "source": source,
                        "created_at": summary.created_at,
                        "updated_at": summary.updated_at,
                    }
                    for source, nominal_dir in group_members
                ]
                candidate = {
                    "path": file,
                    "copies": copies,
                    "sources": list(group_sources),
                    "homes": list(
                        dict.fromkeys(source.home for source in group_sources)
                    ),
                    "created_at": summary.created_at,
                    "updated_at": summary.updated_at,
                    "timestamp_source": (
                        "session-record-minmax"
                        if summary.timestamp_count
                        else "unknown"
                    ),
                    "selected_kind": group_kind,
                    "selected_updated_at": summary.updated_at,
                    "session_id": sid,
                }
                entry = by_id.get(sid)
                if entry is None:
                    by_id[sid] = candidate
                    continue
                entry["copies"].extend(copies)
                sources = list(entry["sources"])
                existing_labels = {item.display_label for item in sources}
                for source in group_sources:
                    if source.display_label not in existing_labels:
                        sources.append(source)
                        existing_labels.add(source.display_label)
                homes = list(entry["homes"])
                for source in group_sources:
                    if source.home not in homes:
                        homes.append(source.home)
                existing_selected_time = (
                    entry["selected_updated_at"]
                    if entry["selected_updated_at"] is not None
                    else float("-inf")
                )
                candidate_time = (
                    summary.updated_at
                    if summary.updated_at is not None
                    else float("-inf")
                )
                candidate_wins = candidate_time > existing_selected_time or (
                    candidate_time == existing_selected_time
                    and group_kind == "active"
                    and entry["selected_kind"] != "active"
                )
                if candidate_wins:
                    entry["path"] = file
                    entry["selected_kind"] = group_kind
                    entry["selected_updated_at"] = summary.updated_at
                entry["sources"] = sources
                entry["homes"] = homes
                created_values = [
                    value
                    for value in (entry["created_at"], summary.created_at)
                    if value is not None
                ]
                updated_values = [
                    value
                    for value in (entry["updated_at"], summary.updated_at)
                    if value is not None
                ]
                entry["created_at"] = min(created_values) if created_values else None
                entry["updated_at"] = max(updated_values) if updated_values else None
                if summary.timestamp_count:
                    entry["timestamp_source"] = "session-record-minmax"

        return sorted(
            by_id.values(),
            key=lambda entry: (
                entry["updated_at"]
                if entry["updated_at"] is not None
                else float("-inf")
            ),
            reverse=True,
        )

    @staticmethod
    def _group_project_dirs(pairs: List[tuple]) -> List[tuple]:
        """Group nominal source dirs by physical directory identity."""
        groups: Dict[str, tuple] = {}
        for source, project_dir in pairs:
            try:
                key = str(project_dir.resolve())
            except (OSError, RuntimeError):
                key = str(project_dir)
            group = groups.get(key)
            if group is None:
                groups[key] = (project_dir, [(source, project_dir)])
            else:
                group[1].append((source, project_dir))
        return list(groups.values())

    def project_dir_pairs(self) -> Dict[str, List[tuple]]:
        """Enumerate every project dir across all sources.

        Returns ``{encoded_project_name: [(source, dir), ...]}`` — one entry
        per distinct project, with a pair per source that holds history for
        it. The encoded name (``-Users-<name>-app``) is the project's true
        identity; decoding ``-`` back to ``/`` is lossy (real dir names may
        contain hyphens), so the encoded form is displayed as-is.
        """
        result: Dict[str, List[tuple]] = {}
        for source in self.sources:
            projects_dir = source.home / "projects"
            if not projects_dir.is_dir():
                continue
            for candidate in sorted(projects_dir.iterdir()):
                if candidate.is_dir():
                    result.setdefault(candidate.name, []).append(
                        (source, candidate)
                    )
        return result

    def find_all_projects_sessions(self) -> List[Dict[str, Any]]:
        """Find sessions for EVERY project across all sources (--all-projects).

        Each returned ref carries a ``project`` field with the encoded
        project name. Sorted newest first across all projects.
        """
        sessions: List[Dict[str, Any]] = []
        for project_name, pairs in self.project_dir_pairs().items():
            for ref in self._merge_sessions_from_dirs(pairs):
                ref["project"] = project_name
                sessions.append(ref)
        return sorted(
            sessions,
            key=lambda entry: (
                entry["updated_at"]
                if entry["updated_at"] is not None
                else float("-inf")
            ),
            reverse=True,
        )

    def find_search_sessions(
        self,
        *,
        project_path: Optional[str],
        all_projects: bool,
        keywords: List[str],
        case_sensitive: bool,
        use_prefilter: bool,
        exclude_sessions: set[str],
        on_prefilter_progress: Optional[Callable[[], None]] = None,
    ) -> tuple[List[Dict[str, Any]], int, int, bool]:
        """Collect search refs after a safe file-level candidate pass.

        Metadata discovery used to call ``scan_claude_session`` for every
        session before keyword or date filtering, then ``search_sessions``
        parsed the same corpus again.  On the real 5,293-session inventory
        that made a one-day query spend minutes parsing files that could not
        contain the keyword.

        This method de-duplicates aliased physical directories, peeks only far
        enough from EOF to read each file's final internal ``sessionId``, and
        lets the native byte scanner identify candidate physical files. Once
        any copy is a candidate, every active/archive filename carrying that
        same session id is retained and fully parsed; that preserves renamed
        copy unions and the exact untimed-record count under date windows.
        ``None`` from the scanner means no trustworthy filtering information,
        so it falls back to the original full metadata scan.

        Returns ``(candidate_refs, total_sessions, project_count, applied)``.
        The total is derived from ``(project, sessionId)`` identities so the
        user-visible searched-session count stays stable even when copies were
        renamed. Only the short identity peek is paid by every file; complete
        metadata and conversation parsing remain candidate-only. Conversation
        chronology still comes exclusively from internal records.
        """
        if all_projects:
            project_pairs = self.project_dir_pairs()
        else:
            pairs = self._resolve_project_dirs(project_path or "")
            project_pairs = {pairs[0][1].name: pairs} if pairs else {}

        if not use_prefilter:
            sessions: List[Dict[str, Any]] = []
            for project_name, pairs in project_pairs.items():
                for ref in self._merge_sessions_from_dirs(pairs):
                    if all_projects:
                        ref["project"] = project_name
                    sessions.append(ref)
            sessions = [
                ref
                for ref in sessions
                if ref["session_id"] not in exclude_sessions
            ]
            project_count = len({ref.get("project") for ref in sessions}) if all_projects else int(bool(sessions))
            return sessions, len(sessions), project_count, False

        unique_paths: Dict[str, Path] = {}
        keys_by_path: Dict[str, set[tuple[str, str]]] = defaultdict(set)
        for project_name, pairs in project_pairs.items():
            for scan_dir, _group_members in self._group_project_dirs(pairs):
                for file in scan_dir.glob("*.jsonl"):
                    if file.name.startswith("agent-"):
                        continue
                    key = (project_name, file.name)
                    try:
                        physical = str(file.resolve())
                    except (OSError, RuntimeError):
                        physical = str(file)
                    unique_paths.setdefault(physical, file)
                    keys_by_path[physical].add(key)

        identity_by_path: Dict[str, Optional[str]] = {}
        uncertain_paths: set[str] = set()
        for physical, file in unique_paths.items():
            session_id = peek_final_claude_session_id(file)
            if session_id is None or session_id in exclude_sessions:
                identity_by_path[physical] = None
                uncertain_paths.add(physical)
            else:
                identity_by_path[physical] = session_id

        matched_files = files_possibly_matching(
            unique_paths.values(),
            keywords,
            case_sensitive=case_sensitive,
            on_progress=on_prefilter_progress,
        )
        if matched_files is None:
            return self.find_search_sessions(
                project_path=project_path,
                all_projects=all_projects,
                keywords=keywords,
                case_sensitive=case_sensitive,
                use_prefilter=False,
                exclude_sessions=exclude_sessions,
            )

        matched_physical: set[str] = set()
        for file in matched_files:
            try:
                matched_physical.add(str(file.resolve()))
            except (OSError, RuntimeError):
                matched_physical.add(str(file))

        # A bounded identity miss is uncertainty, not evidence that filename
        # identity is authoritative. Promote it to the candidate set, then pay
        # the full metadata scan only for that small subset. This runs after
        # the native keyword pass, so old/no-ID files cannot silently turn the
        # identity index back into a full-corpus JSON parser.
        candidate_physical = matched_physical | uncertain_paths
        candidate_summaries: Dict[str, Any] = {}
        for physical in candidate_physical:
            summary = scan_claude_session(unique_paths[physical])
            candidate_summaries[physical] = summary
            identity_by_path[physical] = summary.session_id

        total_identities: set[tuple[str, str]] = set()
        projects_with_sessions: set[str] = set()
        for physical in unique_paths:
            session_id = identity_by_path[physical]
            if session_id is None or session_id in exclude_sessions:
                continue
            for project_name, _filename in keys_by_path[physical]:
                identity = (project_name, session_id)
                total_identities.add(identity)
                projects_with_sessions.add(project_name)

        candidate_identities: set[tuple[str, str]] = set()
        for physical in candidate_physical:
            session_id = identity_by_path.get(physical)
            if session_id is None or session_id in exclude_sessions:
                continue
            for project_name, _filename in keys_by_path[physical]:
                candidate_identities.add((project_name, session_id))

        candidate_paths_by_project: Dict[str, set[str]] = defaultdict(set)
        for physical in unique_paths:
            session_id = identity_by_path[physical]
            if session_id is None or session_id in exclude_sessions:
                continue
            for project_name, _filename in keys_by_path[physical]:
                if (project_name, session_id) in candidate_identities:
                    candidate_paths_by_project[project_name].add(physical)

        sessions = []
        for project_name, pairs in project_pairs.items():
            candidate_paths = candidate_paths_by_project.get(project_name, set())
            if not candidate_paths:
                continue
            for ref in self._merge_sessions_from_dirs(
                pairs,
                candidate_paths=candidate_paths,
                candidate_summaries=candidate_summaries,
            ):
                if ref["session_id"] in exclude_sessions:
                    continue
                if all_projects:
                    ref["project"] = project_name
                sessions.append(ref)
        sessions.sort(
            key=lambda entry: (
                entry["updated_at"]
                if entry["updated_at"] is not None
                else float("-inf")
            ),
            reverse=True,
        )
        return (
            sessions,
            len(total_identities),
            len(projects_with_sessions),
            True,
        )

    def _resolve_project_dirs(self, project_path: str) -> List[tuple]:
        """
        Resolve a project path to its encoded dir under EACH home's projects/.

        Claude Code encodes the project's ABSOLUTE working-directory path by
        replacing every ``/`` with ``-`` (e.g. ``/Users/<name>/app`` ->
        ``-Users-<name>-app``). The directory name is NOT the basename, so a
        bare name or an unexpanded ``~`` path never matches directly — the #1
        reason a real history is mistaken for "no sessions". The same project
        has a same-named encoded dir under every home that holds history for it.

        Strategy (the exact-vs-fallback decision is GLOBAL, not per-home):
        1. Try the exact encoded dir in every home. If it matches in ANY home,
           the project identity is known precisely — return those exact matches
           only. A home lacking the exact dir contributes nothing; it is NOT
           fuzzy-matched, so a different project that merely shares the basename
           in another profile home can never be conflated in.
        2. Only if NO home has the exact dir, treat the input as a bare basename
           and reverse-look-up ``-<basename>`` across all homes. Require a SINGLE
           distinct encoded name; if the basename maps to two different projects
           (within OR across homes), that is ambiguous — warn and return nothing
           rather than guess.

        Returns:
            List of ``(source, encoded_dir)`` pairs — one per source where the
            resolved project dir exists.
        """
        # Encode the resolved absolute path once (for exact matching).
        exact_name: Optional[str] = None
        try:
            abs_path = Path(project_path).expanduser().resolve()
            exact_name = str(abs_path).replace("/", "-")
        except (OSError, RuntimeError):
            exact_name = None
        base = Path(project_path).name

        # Pass 1 — exact encoded-dir match across ALL homes. A single exact hit
        # anywhere fixes the project identity, so we never fuzzy-fall-back.
        if exact_name is not None:
            exact_hits = [
                (source, source.home / "projects" / exact_name)
                for source in self.sources
                if (source.home / "projects" / exact_name).is_dir()
            ]
            if exact_hits:
                return exact_hits

        # Pass 2 — no exact match anywhere: reverse-look-up the bare basename,
        # but bind only ONE distinct project so same-basename projects living in
        # different homes are never conflated together.
        if not base:
            return []
        by_name: Dict[str, List[tuple]] = {}
        for source in self.sources:
            projects_dir = source.home / "projects"
            if not projects_dir.is_dir():
                continue
            for d in projects_dir.iterdir():
                if d.is_dir() and d.name.endswith("-" + base):
                    by_name.setdefault(d.name, []).append((source, d))

        if not by_name:
            return []
        if len(by_name) > 1:
            print(
                f"Ambiguous project name '{base}' — {len(by_name)} distinct "
                "projects match across homes; re-run with the full absolute path:",
                file=sys.stderr,
            )
            for name in sorted(by_name):
                sources_str = ", ".join(
                    source.display_label for source, _ in by_name[name]
                )
                print(f"  {name}  [{sources_str}]", file=sys.stderr)
            return []
        # Exactly one distinct project — use it wherever it exists.
        return next(iter(by_name.values()))

    def search_sessions(
        self,
        session_refs: List[Dict[str, Any]],
        keywords: List[str],
        case_sensitive: bool = False,
        from_timestamp: Optional[float] = None,
        to_timestamp: Optional[float] = None,
        use_prefilter: bool = True,
        include_agent_prompts: bool = False,
    ) -> List[Dict[str, Any]]:
        """
        Search sessions for keywords.

        Args:
            session_refs: Session refs from ``find_project_sessions`` (each has
                every active/archive copy plus a representative path).
            keywords: Keywords to search for.
            case_sensitive: Whether to perform case-sensitive search.
            from_timestamp: Inclusive lower internal record timestamp.
            to_timestamp: Inclusive upper internal record timestamp.
            use_prefilter: Skip the expensive per-line json.loads() + text
                extraction pass on copies a raw byte scan already proves
                cannot contain any keyword (see ``files_possibly_matching``).
                Forced off automatically when a date window is active — see
                below for why.
            include_agent_prompts: Include Claude ``type=user`` records marked
                ``isSidechain``. Those records are prompts written by the main
                agent to a subagent, not human-authored conversation. They are
                excluded by default; assistant-side sidechain records remain
                searchable because they contain the subagent's real output.

        Returns:
            List of match dicts (session ref + match counts), most mentions
            first.
        """
        # The pre-filter is only safe to apply when no date window is active.
        # `excluded_untimed_count` below counts records that lack a timestamp
        # ACROSS EVERY COPY of a session, independent of whether those records
        # contain a keyword — it exists purely to report "N records could not
        # be checked against your --from-date/--to-date window". Skipping a
        # copy's full parse because it has no keyword match would silently
        # drop its contribution to that count. Outside a date window this
        # counter is never computed, so skipping is unconditionally safe —
        # the file-level pre-filter is provably an over-approximation (see
        # files_possibly_matching's docstring), so a copy it rules out cannot
        # contain a matchable record.
        use_prefilter = use_prefilter and from_timestamp is None and to_timestamp is None
        matched_files: Optional[set[Path]] = None
        # Always case-folded regardless of `case_sensitive`: casefold-matching
        # is a strict superset of exact-case matching (anything an exact-case
        # check would find, casefold-matching finds too), so it is always a
        # safe over-approximation to hand to the line-level pre-check in
        # iter_jsonl — it costs a little filtering precision in
        # case-sensitive mode, never a missed match.
        # Gate on the ORIGINAL keywords, not the folded ones: "ß".casefold()
        # is "ss" (ASCII), so checking after folding would answer "safe" for
        # precisely the input the check exists to reject.
        line_keywords = (
            [kw.casefold() for kw in keywords]
            if use_prefilter and keywords_are_raw_byte_safe(keywords)
            else None
        )
        if use_prefilter:
            all_copy_paths = [
                copy["path"]
                for ref in session_refs
                for copy in (
                    ref.get("copies")
                    or [{"path": ref["path"], "source": ref["sources"][0]}]
                )
            ]
            matched_files = files_possibly_matching(
                all_copy_paths, keywords, case_sensitive=case_sensitive
            )

        matches: List[Dict[str, Any]] = []
        search_keywords = [
            (keyword, keyword if case_sensitive else keyword.casefold())
            for keyword in keywords
        ]

        for ref in session_refs:
            keyword_counts = defaultdict(int)
            total_mentions = 0
            match_sources: set[str] = set()
            matching_source_labels: set[str] = set()
            match_timestamps = TimestampRange()
            excluded_untimed_count = 0
            excluded_untimed_from_prior_copies: set[str] = set()
            matched_records_from_prior_copies: set[str] = set()
            matched_file_history_paths: set[str] = set()
            matching_copies: List[Dict[str, Any]] = []

            copies = ref.get("copies") or [
                {
                    "path": ref["path"],
                    "source": ref["sources"][0],
                    "created_at": ref["created_at"],
                    "updated_at": ref["updated_at"],
                }
            ]
            # Archive copies of one session are routinely byte-identical: a
            # registered long-term backup snapshots the same .jsonl every run,
            # so one session commonly carries a dozen copies with a single
            # distinct content (measured on a real project: 46 of 60 sampled
            # multi-copy sessions held 13 copies and exactly 1 content; that
            # redundancy was 85% of the corpus's bytes). Parsing all of them
            # costs Nx for one session's worth of records — _record_identity
            # dedupes them straight back down afterwards.
            #
            # A byte-identical copy therefore has a fully predictable outcome:
            # every record identity is already registered by the twin that ran
            # first, so it can add no new mentions, no new untimed records and
            # no new timestamps. Only its own source label is new. Reuse that
            # conclusion instead of re-parsing.
            #
            # Hash only sizes that actually repeat — a unique size cannot have
            # an identical twin, so single-copy sessions pay nothing.
            sizes_worth_hashing = repeated_copy_sizes(copies)
            digest_had_match: Dict[str, bool] = {}

            for copy in copies:
                session_file = copy["path"]
                source = copy["source"]
                if matched_files is not None and session_file not in matched_files:
                    # A raw byte scan already proved this exact copy cannot
                    # contain any keyword — see the use_prefilter comment
                    # above for why skipping it here (no date window active)
                    # cannot under-count anything.
                    continue

                copy_digest: Optional[str] = None
                try:
                    copy_size = session_file.stat().st_size
                except OSError:
                    copy_size = None
                if copy_size in sizes_worth_hashing:
                    copy_digest = file_content_digest(session_file)
                if copy_digest is not None and copy_digest in digest_had_match:
                    if digest_had_match[copy_digest]:
                        # Same bytes as a copy already processed for this
                        # session: it matched, so this one does too, and every
                        # one of its records was already counted.
                        matching_source_labels.add(source.display_label)
                        matching_copies.append(
                            {
                                "path": session_file,
                                "source": source,
                                "new_mentions": 0,
                            }
                        )
                    continue

                copy_had_match = False
                copy_new_mentions = 0
                copy_untimed_records: set[str] = set()
                copy_matched_records: set[str] = set()
                try:
                    records = iter_jsonl(session_file, line_keywords=line_keywords)
                    for data in records:
                        record_identity = _record_identity(data)
                        hide_agent_prompt = (
                            not include_agent_prompts
                            and is_claude_agent_prompt_record(data)
                        )
                        record_timestamp = parse_timestamp(data.get("timestamp"))
                        if (
                            record_timestamp is None
                            and data.get("type") == "file-history-snapshot"
                        ):
                            snapshot = data.get("snapshot")
                            if isinstance(snapshot, dict):
                                record_timestamp = parse_timestamp(
                                    snapshot.get("timestamp")
                                )
                        if from_timestamp is not None or to_timestamp is not None:
                            if record_timestamp is None:
                                if record_identity not in excluded_untimed_from_prior_copies:
                                    excluded_untimed_count += 1
                                copy_untimed_records.add(record_identity)
                                continue
                            if not timestamp_in_window(
                                record_timestamp, from_timestamp, to_timestamp
                            ):
                                continue

                        if data.get("type") == "file-history-snapshot":
                            snapshot = data.get("snapshot")
                            tracked = (
                                snapshot.get("trackedFileBackups")
                                if isinstance(snapshot, dict)
                                else None
                            )
                            if isinstance(tracked, dict):
                                for original_path in tracked:
                                    if not isinstance(original_path, str):
                                        continue
                                    path_text = (
                                        original_path
                                        if case_sensitive
                                        else original_path.casefold()
                                    )
                                    path_counts = {
                                        keyword: 1
                                        for keyword, search_keyword in search_keywords
                                        if search_keyword in path_text
                                    }
                                    if not path_counts:
                                        continue
                                    copy_had_match = True
                                    matching_source_labels.add(source.display_label)
                                    match_sources.add("file_history_path")
                                    if record_timestamp is not None:
                                        match_timestamps.observe(record_timestamp)
                                    if original_path in matched_file_history_paths:
                                        continue
                                    matched_file_history_paths.add(original_path)
                                    path_mentions = sum(path_counts.values())
                                    for keyword, count in path_counts.items():
                                        keyword_counts[keyword] += count
                                    total_mentions += path_mentions
                                    copy_new_mentions += path_mentions

                        record_counts = defaultdict(int)
                        record_sources: set[str] = set()
                        for segment in searchable_segments(data):
                            if hide_agent_prompt and segment.source == "message":
                                continue
                            search_text = (
                                segment.text
                                if case_sensitive
                                else segment.text.casefold()
                            )
                            for keyword, search_keyword in search_keywords:
                                count = search_text.count(search_keyword)
                                if count > 0:
                                    record_counts[keyword] += count
                                    record_sources.add(segment.source)
                        record_mentions = sum(record_counts.values())
                        if not record_mentions:
                            continue

                        copy_had_match = True
                        matching_source_labels.add(source.display_label)
                        copy_matched_records.add(record_identity)
                        if record_identity in matched_records_from_prior_copies:
                            continue
                        for keyword, count in record_counts.items():
                            keyword_counts[keyword] += count
                        total_mentions += record_mentions
                        copy_new_mentions += record_mentions
                        match_sources.update(record_sources)
                        if record_timestamp is not None:
                            match_timestamps.observe(record_timestamp)
                except (OSError, UnicodeError) as e:
                    print(
                        f"Warning: Error processing {session_file}: {e}",
                        file=sys.stderr,
                    )
                    continue

                excluded_untimed_from_prior_copies.update(copy_untimed_records)
                matched_records_from_prior_copies.update(copy_matched_records)
                if copy_digest is not None:
                    digest_had_match[copy_digest] = copy_had_match
                if copy_had_match:
                    matching_copies.append(
                        {
                            "path": session_file,
                            "source": source,
                            "new_mentions": copy_new_mentions,
                        }
                    )

            if total_mentions > 0:
                primary_copy = max(
                    matching_copies,
                    key=lambda item: (
                        item["new_mentions"],
                        item["source"].kind == "active",
                    ),
                )
                matches.append(
                    {
                        "path": primary_copy["path"],
                        "matching_copies": matching_copies,
                        "homes": ref["homes"],
                        "sources": ref["sources"],
                        "matching_source_labels": matching_source_labels,
                        "total_mentions": total_mentions,
                        "keyword_counts": dict(keyword_counts),
                        "match_sources": sorted(match_sources),
                        "created_at": ref["created_at"],
                        "updated_at": ref["updated_at"],
                        "match_created_at": match_timestamps.earliest,
                        "match_updated_at": match_timestamps.latest,
                        "timestamp_source": ref["timestamp_source"],
                        "excluded_untimed_records": excluded_untimed_count,
                        "size": sum(item["path"].stat().st_size for item in copies),
                    }
                )

        matches.sort(
            key=lambda match: (
                match["total_mentions"],
                match["match_updated_at"]
                if match["match_updated_at"] is not None
                else float("-inf"),
            ),
            reverse=True,
        )
        return matches

    def get_session_stats(self, session_file: Path) -> Dict[str, Any]:
        """
        Get detailed statistics for a session file.

        Args:
            session_file: Path to session JSONL file

        Returns:
            Dictionary of session statistics
        """
        stats = {
            "total_lines": 0,
            "user_messages": 0,
            "assistant_messages": 0,
            "tool_uses": defaultdict(int),
            "write_calls": 0,
            "edit_calls": 0,
            "read_calls": 0,
            "bash_calls": 0,
            "file_operations": [],
        }

        try:
            with open(session_file, "r") as f:
                for line in f:
                    stats["total_lines"] += 1

                    try:
                        data = json.loads(line.strip())

                        # Count message types
                        role = data.get("role") or data.get("message", {}).get("role")
                        if role == "user":
                            stats["user_messages"] += 1
                        elif role == "assistant":
                            stats["assistant_messages"] += 1

                        # Analyze tool uses
                        content = data.get("content") or data.get("message", {}).get(
                            "content", []
                        )
                        for item in content:
                            if not isinstance(item, dict):
                                continue

                            if item.get("type") == "tool_use":
                                tool_name = item.get("name", "unknown")
                                stats["tool_uses"][tool_name] += 1

                                # Track file operations
                                if tool_name == "Write":
                                    stats["write_calls"] += 1
                                    file_path = item.get("input", {}).get(
                                        "file_path", ""
                                    )
                                    if file_path:
                                        stats["file_operations"].append(
                                            ("write", file_path)
                                        )
                                elif tool_name == "Edit":
                                    stats["edit_calls"] += 1
                                    file_path = item.get("input", {}).get(
                                        "file_path", ""
                                    )
                                    if file_path:
                                        stats["file_operations"].append(
                                            ("edit", file_path)
                                        )
                                elif tool_name == "Read":
                                    stats["read_calls"] += 1
                                elif tool_name == "Bash":
                                    stats["bash_calls"] += 1

                    except json.JSONDecodeError:
                        continue

        except Exception as e:
            print(f"Error analyzing {session_file}: {e}", file=sys.stderr)

        # Convert defaultdict to regular dict
        stats["tool_uses"] = dict(stats["tool_uses"])

        return stats

    def _extract_text_content(self, data: Dict[str, Any]) -> str:
        """Compatibility wrapper around the structured event extractor."""
        return " ".join(segment.text for segment in searchable_segments(data))


def _add_home_flags(subparser) -> None:
    """Attach the shared home-scoping flags to a subparser (list / search)."""
    subparser.add_argument(
        "--home",
        action="append",
        metavar="DIR",
        help="Restrict to exact Claude home dir(s) and bypass the archive "
        "registry (repeatable). Default: search every active home plus "
        "registered archives.",
    )
    subparser.add_argument(
        "--main-only",
        action="store_true",
        help="Search only ~/.claude, bypassing profile homes and archives.",
    )
    subparser.add_argument(
        "--history-sources",
        metavar="FILE",
        help=(
            "History source registry (default: ~/.claude/history-sources.json "
            "when present). Incompatible with --home/--main-only."
        ),
    )
    subparser.add_argument(
        "--from-date",
        help="Inclusive start: YYYY-MM-DD (local day) or timezone-qualified ISO datetime",
    )
    subparser.add_argument(
        "--to-date",
        help="Inclusive end: YYYY-MM-DD (local day) or timezone-qualified ISO datetime",
    )


def _sources_for(args) -> tuple:
    """Resolve active/archive sources from CLI flags (used by list / search).

    Returns ``(sources, narrowed, warnings)``. ``narrowed`` is True when the user passed
    ``--home`` / ``--main-only``, so the caller can treat an empty result as a
    real "your selection matched no home with history" error instead of
    silently widening back to searching every home.
    """
    main_only = getattr(args, "main_only", False)
    explicit = getattr(args, "home", None)
    registry = getattr(args, "history_sources", None)
    if main_only and explicit:
        raise HistorySourceConfigError("--main-only cannot be combined with --home")
    if registry and (main_only or explicit):
        raise HistorySourceConfigError(
            "--history-sources cannot be combined with --home/--main-only"
        )
    if main_only:
        sources, warnings = discover_claude_sources(
            explicit_homes=[Path.home() / ".claude"]
        )
        return sources, True, warnings
    if explicit:
        sources, warnings = discover_claude_sources(explicit_homes=explicit)
        return sources, True, warnings
    sources, warnings = discover_claude_sources(manifest_path=registry)
    return sources, False, warnings


def _analyzer_or_exit(args) -> "SessionAnalyzer":
    """Build a SessionAnalyzer, erroring out if a narrowing flag matched no home.

    Without this, an explicit ``--home``/``--main-only`` that resolves to no
    home-with-history would (via an empty list) either search nothing or, worse,
    reintroduce full discovery — turning a scope-narrowing flag into the widest
    possible scope. We fail loudly instead.
    """
    try:
        sources, narrowed, warnings = _sources_for(args)
    except HistorySourceConfigError as error:
        print(f"History source configuration error: {error}", file=sys.stderr)
        sys.exit(2)
    if narrowed and not sources:
        print(
            "No Claude home with a projects/ dir matched your --home/--main-only "
            "selection (a --home value must be a config home such as ~/.claude "
            "or ~/.claude-profiles/<name>, not its projects/ subdir).",
            file=sys.stderr,
        )
        sys.exit(1)
    return SessionAnalyzer(sources=sources, warnings=warnings)


def _parse_date_window(args, parser) -> tuple[Optional[float], Optional[float]]:
    try:
        from_timestamp = (
            parse_date_boundary(args.from_date) if args.from_date else None
        )
        to_timestamp = (
            parse_date_boundary(args.to_date, end=True) if args.to_date else None
        )
    except ValueError as error:
        parser.error(str(error))
    if (
        from_timestamp is not None
        and to_timestamp is not None
        and from_timestamp > to_timestamp
    ):
        parser.error("--from-date must not be later than --to-date")
    return from_timestamp, to_timestamp


def _format_range(earliest: Optional[float], latest: Optional[float]) -> str:
    if earliest is None or latest is None:
        return "unknown (no internal timestamp)"
    return f"{format_timestamp(earliest)} .. {format_timestamp(latest)}"


def _source_summary(sources: List[HistorySource]) -> str:
    return ", ".join(source.display_label for source in sources)


def _validate_project_scope(args, parser) -> None:
    """Exactly one of project_path / --all-projects must be given."""
    if args.all_projects and args.project_path:
        parser.error("pass either a project path or --all-projects, not both")
    if not args.all_projects and not args.project_path:
        parser.error(
            "a project path is required unless --all-projects is given "
            "(use --all-projects when you do not know which project it was)"
        )


def _normalize_search_scope(args, parser) -> None:
    """Resolve the search positional grammar without argparse ambiguity.

    ``project_path?`` followed by ``keywords+`` is ambiguous when
    ``--all-projects`` is active: argparse consumes the first of two keywords as
    the optional project. Parse one term list instead, then apply the scope the
    caller selected.
    """
    terms = list(args.search_terms)
    if args.all_projects:
        args.project_path = None
        args.keywords = terms
        return
    if len(terms) < 2:
        parser.error(
            "search requires a project path followed by at least one keyword; "
            "use --all-projects when the project is unknown"
        )
    args.project_path = terms[0]
    args.keywords = terms[1:]


def _maybe_hint_all_projects(project_path: str) -> None:
    """Warn when an unresolved ``project_path`` looks like it was meant as a
    keyword, not a path — the exact trap of running ``search`` with keywords
    only and forgetting ``--all-projects``: argparse's positional grammar
    (``project_path?`` then ``keywords+``) silently consumes the first
    keyword as the project, the lookup finds nothing, and the message alone
    ("No sessions found for project: embedding") reads as "your keyword
    doesn't exist" rather than "you searched for a project by that name".
    A real project path always contains a path separator or resolves to an
    existing directory; a bare word that does neither almost certainly was
    not one.
    """
    if "/" in project_path or "\\" in project_path or Path(project_path).exists():
        return
    print(
        "Hint: this looks like it might be a keyword, not a project path — "
        "if you don't know which project the conversation happened in, "
        "re-run with --all-projects (e.g. `search --all-projects "
        f"{project_path} ...`).",
        file=sys.stderr,
    )


def _collect_sessions(analyzer: "SessionAnalyzer", args) -> List[Dict[str, Any]]:
    """Collect session refs for the requested scope, applying exclusions."""
    if args.all_projects:
        sessions = analyzer.find_all_projects_sessions()
    else:
        sessions = analyzer.find_project_sessions(args.project_path)
    exclude = set(getattr(args, "exclude_session", None) or [])
    if exclude:
        sessions = [
            ref
            for ref in sessions
            if ref["session_id"] not in exclude and ref["path"].stem not in exclude
        ]
    return sessions


def _collect_search_sessions(
    analyzer: "SessionAnalyzer", args
) -> tuple[List[Dict[str, Any]], int, int, bool]:
    """Candidate-first collection for keyword search, with visible progress."""
    exclude = set(getattr(args, "exclude_session", None) or [])
    progress_count = 0

    def prefilter_progress() -> None:
        nonlocal progress_count
        progress_count += 1
        print(
            "History keyword prefilter is still scanning physical files "
            f"({progress_count * 15}s elapsed)...",
            file=sys.stderr,
        )

    return analyzer.find_search_sessions(
        project_path=args.project_path,
        all_projects=args.all_projects,
        keywords=args.keywords,
        case_sensitive=args.case_sensitive,
        use_prefilter=not args.no_prefilter,
        exclude_sessions=exclude,
        on_prefilter_progress=prefilter_progress,
    )


def _codex_home_for(args) -> Path:
    explicit = getattr(args, "codex_home", None)
    if explicit:
        return Path(explicit).expanduser()
    env_home = os.environ.get("CODEX_HOME")
    if env_home:
        return Path(env_home).expanduser()
    return Path.home() / ".codex"


def _kimi_home_for(args) -> Path:
    return resolve_kimi_home(getattr(args, "kimi_home", None))


def _print_search_widening_hint(args) -> None:
    """On zero matches, point at the widening the user has NOT applied yet.

    "Not found" is the expensive failure mode of this tool, and each of the
    three widenings covers a distinct reason a real conversation can be
    missed: wrong project guess, wrong tool (Codex), or a remembered quote
    whose wording differs from the real one.
    """
    tips: List[str] = []
    if not getattr(args, "all_projects", False):
        tips.append(
            "--all-projects (the conversation may belong to a different "
            "project than the one searched)"
        )
    if not getattr(args, "codex", False):
        tips.append(
            "--codex (it may have been a Codex conversation — Codex rollouts "
            "are a separate store this search skips by default)"
        )
    if not getattr(args, "kimi", False):
        tips.append(
            "--kimi (it may have been a Kimi CLI conversation — Kimi sessions "
            "are a separate store this search skips by default)"
        )
    tips.append(
        "shorter distinctive substrings (a remembered quote often differs "
        "from the real wording in punctuation or a few words)"
    )
    print("Tip: no matches. Before concluding it is absent, retry with:", file=sys.stderr)
    for tip in tips:
        print(f"  - {tip}", file=sys.stderr)


def main():
    """Main entry point."""
    import argparse

    parser = argparse.ArgumentParser(
        description="Analyze Claude Code session history files"
    )

    subparsers = parser.add_subparsers(dest="command", help="Command to run")

    # List sessions command
    list_parser = subparsers.add_parser("list", help="List all sessions for a project")
    list_parser.add_argument(
        "project_path",
        nargs="?",
        help="Project path (omit when using --all-projects)",
    )
    list_parser.add_argument(
        "--all-projects",
        action="store_true",
        help="Sweep every project across all sources instead of one project.",
    )
    list_parser.add_argument(
        "--exclude-session",
        action="append",
        metavar="ID",
        default=[],
        help="Exclude a session id (repeatable) — e.g. the current session, "
        "which always matches the phrase you just typed.",
    )
    list_parser.add_argument(
        "--limit", type=int, default=10, help="Max sessions to show (default: 10)"
    )
    _add_home_flags(list_parser)

    # Triage command — classify how sessions in scope ended (crash recovery,
    # backlog audit). Distinct from `list`: prints the full last-assistant
    # text so a human/agent can judge whether a reply is still expected,
    # rather than a truncated title.
    triage_parser = subparsers.add_parser(
        "triage",
        help="Classify how sessions in a time window/project ended "
        "(interrupted / net-error / stuck-on-tool / done) with full tail text",
    )
    triage_parser.add_argument(
        "project_path",
        nargs="?",
        help="Project path (omit when using --all-projects)",
    )
    triage_parser.add_argument(
        "--all-projects",
        action="store_true",
        help="Sweep every project across all sources instead of one project.",
    )
    triage_parser.add_argument(
        "--exclude-session",
        action="append",
        metavar="ID",
        default=[],
        help="Exclude a session id (repeatable) — e.g. the current session.",
    )
    triage_parser.add_argument(
        "--include-automated",
        action="store_true",
        help="Include sessions whose opening prompt matches the generic "
        "smoke-test pattern ('reply/respond exactly ...'). Excluded by "
        "default. This does NOT catch project-specific automation (e.g. a "
        "git hook that always opens with the same review prompt) — use "
        "--exclude-title-prefix for that; it is deliberately not hardcoded "
        "here since the wording is per-project, not a Claude Code convention.",
    )
    triage_parser.add_argument(
        "--exclude-title-prefix",
        action="append",
        metavar="TEXT",
        default=[],
        help="Exclude sessions whose opening prompt starts with TEXT "
        "(repeatable, case-sensitive). Use this for a project's own "
        "automation convention, e.g. a code-review hook that always opens "
        "with the same fixed prompt — these otherwise dominate a triage "
        "pass because they end in a routine structured tool call, not an "
        "interruption.",
    )
    triage_parser.add_argument(
        "--kind",
        action="append",
        choices=[
            TAIL_INTERRUPTED_EXPLICIT,
            TAIL_NET_ERROR,
            TAIL_STUCK_NO_RESULT,
            TAIL_DONE,
            TAIL_EMPTY,
        ],
        metavar="KIND",
        help="Restrict to one or more tail kinds (repeatable). Default: all "
        "kinds. Note 'done' includes sessions that gave a real reply and may "
        "still be awaiting a response — read last_assistant_text to judge "
        "that; 'done' is not a claim that nothing is outstanding.",
    )
    triage_parser.add_argument(
        "--tail-chars",
        type=int,
        default=4000,
        help="Max characters of the last assistant message to print "
        "(default: 4000; 0 = full text, no truncation).",
    )
    triage_parser.add_argument(
        "--limit",
        type=int,
        default=200,
        help="Max sessions to print, 0 = no limit (default: 200). This is a "
        "print-time cap, not a scan-time one: every session in scope is "
        "still classified before --limit or --kind trims the output, so "
        "--kind does not reduce cost the way narrowing --from-date/--to-date "
        "does. The 200 default exists specifically to stop an accidentally "
        "unscoped `--all-projects` with no date bound from dumping tens of "
        "thousands of lines; pass --limit 0 to explicitly opt into an "
        "unbounded dump once you know the scope is narrow.",
    )
    _add_home_flags(triage_parser)

    locate_codex_parser = subparsers.add_parser(
        "locate-codex",
        help="Locate an exact Codex session UUID from metadata without a corpus scan",
    )
    locate_codex_parser.add_argument("session_id", help="Exact Codex session UUID")
    locate_codex_parser.add_argument(
        "--codex-home",
        metavar="DIR",
        help="Codex home (default: $CODEX_HOME or ~/.codex).",
    )

    # Search command
    search_parser = subparsers.add_parser("search", help="Search sessions for keywords")
    search_parser.add_argument(
        "search_terms",
        nargs="+",
        metavar="PROJECT_OR_KEYWORD",
        help=(
            "Project path followed by keywords, or only keywords with "
            "--all-projects"
        ),
    )
    search_parser.add_argument(
        "--all-projects",
        action="store_true",
        help="Sweep every project across all sources — the default move when "
        "you do not know which project the conversation happened in.",
    )
    search_parser.add_argument(
        "--exclude-session",
        action="append",
        metavar="ID",
        default=[],
        help="Exclude a session id (repeatable) — e.g. the current session, "
        "which always matches the phrase you just typed.",
    )
    search_parser.add_argument(
        "--codex",
        action="store_true",
        help="Also search Codex rollout history (sessions/** + "
        "archived_sessions/ under the Codex home). Codex uses a different "
        "store and schema that the Claude registry never covers.",
    )
    search_parser.add_argument(
        "--codex-home",
        metavar="DIR",
        help="Codex home for --codex (default: $CODEX_HOME or ~/.codex).",
    )
    search_parser.add_argument(
        "--codex-max-scan-seconds",
        type=float,
        default=DEFAULT_CODEX_SCAN_BUDGET_SECONDS,
        metavar="SECONDS",
        help="Stop a broad Codex rollout scan after SECONDS (default: 300). "
        "Use 0 only to explicitly accept an unbounded scan; partial results "
        "are never reported as complete.",
    )
    search_parser.add_argument(
        "--kimi",
        action="store_true",
        help="Also search Kimi CLI session history (sessions/**/wire.jsonl "
        "under the Kimi home). Kimi CLI uses a different store and schema "
        "that the Claude registry never covers.",
    )
    search_parser.add_argument(
        "--kimi-home",
        metavar="DIR",
        help="Kimi CLI home for --kimi (default: $KIMI_HOME or ~/.kimi-code).",
    )
    search_parser.add_argument(
        "--case-sensitive", action="store_true", help="Case-sensitive search"
    )
    search_parser.add_argument(
        "--include-agent-prompts",
        action="store_true",
        help="Include Claude user-side isSidechain records. These are prompts "
        "the main agent sent to subagents, so they are excluded by default. "
        "Assistant-side subagent output remains searchable either way.",
    )
    search_parser.add_argument(
        "--no-prefilter",
        action="store_true",
        help="Disable native file/line pre-filters and fully parse every "
        "session file. The filters are conservative speedups: ordinary ASCII "
        "uses literal matching, and uncased Unicode such as CJK uses an exact "
        "raw/JSON-escaped file regex when ripgrep is available. Unsafe escape "
        "characters, cased or multi-codepoint Unicode folds, missing tools, "
        "and scanner errors already fall back to full parsing. Use this flag "
        "to verify output neutrality or diagnose a suspected missed match.",
    )
    _add_home_flags(search_parser)

    # Stats command
    stats_parser = subparsers.add_parser("stats", help="Get session statistics")
    stats_parser.add_argument("session_file", type=Path, help="Session file path")
    stats_parser.add_argument(
        "--show-files", action="store_true", help="Show file operations"
    )

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    if args.command == "locate-codex":
        raise SystemExit(_print_codex_locations(_codex_home_for(args), args.session_id))

    if args.command == "search":
        _normalize_search_scope(args, parser)
        if args.codex_max_scan_seconds < 0:
            parser.error("--codex-max-scan-seconds must be 0 or greater")
        if (
            args.codex
            and len(args.keywords) == 1
            and CODEX_EXACT_SESSION_ID_RE.fullmatch(args.keywords[0])
            and args.keywords[0] not in set(args.exclude_session)
        ):
            print(
                "Exact Codex session ID detected; using session_meta locator "
                "instead of searching conversation bytes."
            )
            raise SystemExit(
                _print_codex_locations(_codex_home_for(args), args.keywords[0])
            )

    if args.command == "list":
        _validate_project_scope(args, parser)
        from_timestamp, to_timestamp = _parse_date_window(args, parser)
        analyzer = _analyzer_or_exit(args)
        source_summary = _source_summary(analyzer.sources)
        sessions = _collect_sessions(analyzer, args)
        for warning in analyzer.warnings:
            print(f"Warning: {warning}", file=sys.stderr)
        unknown = 0
        if from_timestamp is not None or to_timestamp is not None:
            filtered = []
            for ref in sessions:
                if ref["created_at"] is None or ref["updated_at"] is None:
                    unknown += 1
                    continue
                if range_overlaps_window(
                    ref["created_at"],
                    ref["updated_at"],
                    from_timestamp,
                    to_timestamp,
                ):
                    filtered.append(ref)
            sessions = filtered
        if not sessions:
            if args.all_projects:
                print("No sessions found across all projects")
            else:
                print(f"No sessions found for project: {args.project_path}")
                _maybe_hint_all_projects(args.project_path)
            print(
                f"(searched {len(analyzer.sources)} source(s): {source_summary})",
                file=sys.stderr,
            )
            sys.exit(1)

        project_names = sorted(
            {ref.get("project") for ref in sessions if ref.get("project")}
        )
        if args.all_projects:
            print(
                f"Found {len(sessions)} session(s) across "
                f"{len(project_names)} project(s)"
            )
        else:
            print(f"Found {len(sessions)} session(s) for {args.project_path}")
        print(
            f"Searched {len(analyzer.sources)} source(s): {source_summary}\n"
        )
        if unknown:
            print(
                f"Warning: excluded {unknown} session(s) without an internal "
                "timestamp; file mtime was not used as a fallback.",
                file=sys.stderr,
            )
        print(f"Showing {min(args.limit, len(sessions))} most recent:\n")

        shown = sessions[: args.limit]
        last_project: Optional[str] = None
        for i, ref in enumerate(shown, 1):
            if args.all_projects:
                project = ref.get("project") or "unknown-project"
                if project != last_project:
                    count = sum(
                        1 for item in sessions if item.get("project") == project
                    )
                    print(f"== {project} ({count} session(s))")
                    last_project = project
            session = ref["path"]
            size_kb = session.stat().st_size / 1024
            labels = _source_summary(ref["sources"])
            print(f"{i}. {session.name}")
            print(
                f"   Internal range: "
                f"{_format_range(ref['created_at'], ref['updated_at'])}"
            )
            print(f"   Size: {size_kb:.1f} KB")
            print(f"   Source: {labels}")
            if len(ref["copies"]) > 1:
                print(f"   Physical copies searched by keyword queries: {len(ref['copies'])}")
            print(f"   Path: {session}")
            print()

    elif args.command == "triage":
        _validate_project_scope(args, parser)
        from_timestamp, to_timestamp = _parse_date_window(args, parser)
        analyzer = _analyzer_or_exit(args)
        source_summary = _source_summary(analyzer.sources)
        sessions = _collect_sessions(analyzer, args)
        for warning in analyzer.warnings:
            print(f"Warning: {warning}", file=sys.stderr)

        unknown = 0
        if from_timestamp is not None or to_timestamp is not None:
            # Deliberately `timestamp_in_window` on updated_at, NOT
            # `range_overlaps_window` on the full [created_at, updated_at]
            # span: triage asks "did this session's momentum stop around
            # this boundary," so a session that merely happened to be
            # running through the window but kept going for hours/days
            # afterward is not a candidate, even though its range overlaps
            # the window. (list's browsing use case wants the overlap
            # semantics; triage does not — verified against a real reboot
            # window, where overlap swept in every still-running session.)
            filtered = []
            for ref in sessions:
                if ref["updated_at"] is None:
                    unknown += 1
                    continue
                if timestamp_in_window(ref["updated_at"], from_timestamp, to_timestamp):
                    filtered.append(ref)
            sessions = filtered

        # One lightweight scan per session gets both the title (for automated
        # exclusion) and cwd (for display) — cheaper than two passes, and
        # find_*_sessions()'s ref dict does not carry cwd itself.
        exclude_prefixes = tuple(args.exclude_title_prefix)
        excluded_generic = 0
        excluded_prefix = 0
        scanned = []
        for ref in sessions:
            summary = scan_claude_session(ref["path"])
            if not args.include_automated and is_automated_title(summary.title):
                excluded_generic += 1
                continue
            if exclude_prefixes and summary.title.startswith(exclude_prefixes):
                excluded_prefix += 1
                continue
            scanned.append((ref, summary))
        excluded_automated = excluded_generic + excluded_prefix

        allowed_kinds = set(args.kind) if args.kind else None
        results = []
        for ref, summary in scanned:
            tail = classify_session_tail(ref["path"])
            if allowed_kinds is not None and tail.kind not in allowed_kinds:
                continue
            results.append((ref, summary, tail))
        results.sort(
            key=lambda triple: triple[0]["updated_at"]
            if triple[0]["updated_at"] is not None
            else float("-inf"),
            reverse=True,
        )

        # Deliberately one empty-result check, run AFTER --kind filtering,
        # not two separate ones (scanned-empty vs. results-empty) that used
        # to give different exit codes/messages for the same "nothing to
        # show" outcome depending on which filter zeroed it (independent
        # review, 2026-08).
        if not results:
            print("No sessions matched this triage scope.")
            print(
                f"(searched {len(analyzer.sources)} source(s): {source_summary}; "
                f"scanned {len(scanned)}, excluded {excluded_generic} generic-automated "
                f"+ {excluded_prefix} --exclude-title-prefix)",
                file=sys.stderr,
            )
            sys.exit(1)

        total_matched = len(results)
        if args.limit:
            results = results[: args.limit]

        print(
            f"Triaged {len(results)} session(s)"
            + (f" of {total_matched} matched" if len(results) < total_matched else "")
            + f" (excluded {excluded_generic} generic-automated + "
            f"{excluded_prefix} --exclude-title-prefix); "
            f"searched {len(analyzer.sources)} source(s): {source_summary}"
        )
        if unknown:
            print(
                f"Warning: excluded {unknown} session(s) without an internal "
                "timestamp; file mtime was not used as a fallback.",
                file=sys.stderr,
            )
        print()

        for ref, summary, tail in results:
            updated_at = ref["updated_at"]
            when = format_timestamp(updated_at) if updated_at is not None else "(unknown)"
            print("=" * 88)
            print(ref["session_id"])
            print(f"  last update: {when}")
            print(f"  cwd:         {summary.cwd or '(unknown)'}")
            print(f"  kind:        {tail.kind}")
            last_user_preview = " ".join(tail.last_user_text.split())[:200]
            print(f"  last user:   {last_user_preview}")
            print(f"  last assistant ({tail.last_assistant_kind}):")
            text = tail.last_assistant_text
            if args.tail_chars and len(text) > args.tail_chars:
                text = text[: args.tail_chars] + f"... [+{len(text) - args.tail_chars} chars]"
            for line in text.splitlines() or [""]:
                print(f"    {line}")
        print("=" * 88)

    elif args.command == "search":
        _validate_project_scope(args, parser)
        from_timestamp, to_timestamp = _parse_date_window(args, parser)
        analyzer = _analyzer_or_exit(args)
        source_summary = _source_summary(analyzer.sources)
        (
            sessions,
            discovered_session_count,
            discovered_project_count,
            candidate_prefilter_applied,
        ) = _collect_search_sessions(analyzer, args)
        for warning in analyzer.warnings:
            print(f"Warning: {warning}", file=sys.stderr)
        if discovered_session_count == 0 and not args.codex and not args.kimi:
            if args.all_projects:
                print("No sessions found across all projects")
            else:
                print(f"No sessions found for project: {args.project_path}")
                _maybe_hint_all_projects(args.project_path)
            print(
                f"(searched {len(analyzer.sources)} source(s): {source_summary})",
                file=sys.stderr,
            )
            sys.exit(1)

        scope_desc = (
            f" in {discovered_project_count} project(s)"
            if args.all_projects
            else ""
        )
        print(
            f"Searching {discovered_session_count} session(s) across {len(analyzer.sources)} "
            f"source(s) [{source_summary}]{scope_desc} for: {', '.join(args.keywords)}\n"
        )
        if candidate_prefilter_applied:
            print(
                "Candidate prefilter: full-session parsing required for "
                f"{len(sessions)}/{discovered_session_count} session(s).",
                file=sys.stderr,
            )
        matches = (
            analyzer.search_sessions(
                sessions,
                args.keywords,
                args.case_sensitive,
                from_timestamp,
                to_timestamp,
                not args.no_prefilter and not candidate_prefilter_applied,
                args.include_agent_prompts,
            )
            if sessions
            else []
        )

        codex_matches: List[Dict[str, Any]] = []
        codex_home: Optional[Path] = None
        if args.codex:
            codex_home = _codex_home_for(args)
            scan_started = time.monotonic()
            next_discovery_progress = scan_started + CODEX_PROGRESS_INTERVAL_SECONDS

            def discovery_progress(discovered: int) -> None:
                nonlocal next_discovery_progress
                now = time.monotonic()
                elapsed = now - scan_started
                if (
                    args.codex_max_scan_seconds > 0
                    and elapsed > args.codex_max_scan_seconds
                ):
                    raise CodexScanBudgetExceeded(
                        discovered, None, elapsed, "rollout-discovery"
                    )
                if now >= next_discovery_progress:
                    print(
                        "Codex scan progress: "
                        f"stage=rollout-discovery, discovered={discovered}, "
                        f"elapsed={elapsed:.1f}s",
                        file=sys.stderr,
                        flush=True,
                    )
                    while next_discovery_progress <= now:
                        next_discovery_progress += CODEX_PROGRESS_INTERVAL_SECONDS

            try:
                rollouts = discover_codex_rollouts(
                    codex_home, on_progress=discovery_progress
                )
                print(
                    f"Also searching {len(rollouts)} Codex rollout(s) under "
                    f"{codex_home} "
                    f"(--codex; {args.codex_max_scan_seconds:g}s stop-loss)\n",
                    flush=True,
                )
                codex_matches = search_codex_rollouts(
                    rollouts,
                    args.keywords,
                    args.case_sensitive,
                    from_timestamp,
                    to_timestamp,
                    None if args.all_projects else args.project_path,
                    set(args.exclude_session),
                    not args.no_prefilter,
                    args.codex_max_scan_seconds,
                    started_at=scan_started,
                )
            except CodexScanBudgetExceeded as error:
                total_text = "?" if error.total is None else str(error.total)
                print(
                    f"Stopped Codex scan after {error.elapsed:.1f}s at "
                    f"{error.scanned}/{total_text} rollouts ({error.stage}). "
                    "No partial result is being presented as complete. Narrow by "
                    "project/date, use locate-codex for a session UUID, or pass "
                    "--codex-max-scan-seconds 0 only after explicitly accepting an "
                    "unbounded scan.",
                    file=sys.stderr,
                )
                sys.exit(2)
            except CodexScanIncomplete as error:
                sample = ", ".join(str(path) for path in error.paths[:3])
                extra = "" if len(error.paths) <= 3 else f" (+{len(error.paths) - 3} more)"
                print(
                    "Codex scan stopped because rollout input was incomplete; "
                    "no partial result is being presented as complete. "
                    f"Unreadable/malformed: {sample}{extra}",
                    file=sys.stderr,
                )
                sys.exit(2)

        kimi_matches: List[Dict[str, Any]] = []
        kimi_home: Optional[Path] = None
        if args.kimi:
            kimi_home = _kimi_home_for(args)
            kimi_wires = discover_kimi_wires(kimi_home)
            print(
                f"Also searching {len(kimi_wires)} Kimi CLI wire(s) under "
                f"{kimi_home} (--kimi)\n"
            )
            kimi_matches = search_kimi_wires(
                kimi_wires,
                args.keywords,
                args.case_sensitive,
                from_timestamp,
                to_timestamp,
                None if args.all_projects else args.project_path,
                set(args.exclude_session),
                not args.no_prefilter,
            )

        if matches:
            print(f"Found {len(matches)} session(s) with matches:\n")
            project_by_path = {
                ref["path"]: ref.get("project") for ref in sessions
            }
            for info in matches:
                session = info["path"]
                labels = _source_summary(info["sources"])
                matching_labels = ", ".join(sorted(info["matching_source_labels"]))
                print(f"📄 {session.name}")
                project = project_by_path.get(session)
                if project:
                    print(f"   Project: {project}")
                print(
                    "   Internal range: "
                    f"{_format_range(info['created_at'], info['updated_at'])}"
                )
                print(
                    "   Match range: "
                    f"{_format_range(info['match_created_at'], info['match_updated_at'])}"
                )
                print(f"   Session sources: {labels}")
                print(f"   Match sources: {matching_labels}")
                print(f"   Total mentions: {info['total_mentions']}")
                print(
                    f"   Keywords: {', '.join(f'{k}({v})' for k, v in info['keyword_counts'].items())}"
                )
                print(f"   Match fields: {', '.join(info['match_sources'])}")
                if info["excluded_untimed_records"]:
                    print(
                        "   Date-filter note: excluded "
                        f"{info['excluded_untimed_records']} record(s) without an "
                        "internal timestamp; file mtime was not used."
                    )
                print(f"   Path: {session}")
                if len(info["matching_copies"]) > 1:
                    print("   Other matching copies:")
                    for copy in info["matching_copies"]:
                        if copy["path"] == session:
                            continue
                        print(
                            f"     - [{copy['source'].display_label}] {copy['path']}"
                        )
                print()

        if args.codex:
            if codex_matches:
                print(
                    f"Codex rollout matches (home: {codex_home}):\n"
                )
                for info in codex_matches:
                    print(f"📦 {info['path'].name}")
                    print(f"   Session: {info['session_id']}")
                    if info.get("cwd"):
                        print(f"   cwd: {info['cwd']}")
                    print(
                        "   Internal range: "
                        f"{_format_range(info['created_at'], info['updated_at'])}"
                    )
                    print(
                        "   Match range: "
                        f"{_format_range(info['match_created_at'], info['match_updated_at'])}"
                    )
                    print(f"   Total mentions: {info['total_mentions']}")
                    print(
                        f"   Keywords: {', '.join(f'{k}({v})' for k, v in info['keyword_counts'].items())}"
                    )
                    print(f"   Match fields: {', '.join(info['match_sources'])}")
                    if info["excluded_untimed_records"]:
                        print(
                            "   Date-filter note: excluded "
                            f"{info['excluded_untimed_records']} record(s) without an "
                            "internal timestamp; file mtime was not used."
                        )
                    print(f"   Path: {info['path']}")
                    print()
            else:
                print(f"No Codex rollout matches (home: {codex_home}).")

        if args.kimi:
            if kimi_matches:
                print(f"Kimi CLI session matches (home: {kimi_home}):\n")
                for info in kimi_matches:
                    print(f"🌙 {info['path'].name}")
                    print(f"   Session: {info['session_id']}")
                    if info.get("title"):
                        print(f"   Title: {info['title']}")
                    if info.get("cwd"):
                        print(f"   cwd: {info['cwd']}")
                    print(
                        "   Internal range: "
                        f"{_format_range(info['created_at'], info['updated_at'])}"
                    )
                    print(
                        "   Match range: "
                        f"{_format_range(info['match_created_at'], info['match_updated_at'])}"
                    )
                    print(f"   Total mentions: {info['total_mentions']}")
                    print(
                        f"   Keywords: {', '.join(f'{k}({v})' for k, v in info['keyword_counts'].items())}"
                    )
                    print(f"   Match fields: {', '.join(info['match_sources'])}")
                    if info["excluded_untimed_records"]:
                        print(
                            "   Date-filter note: excluded "
                            f"{info['excluded_untimed_records']} record(s) without an "
                            "internal timestamp; file mtime was not used."
                        )
                    print(f"   Path: {info['path']}")
                    print()
            else:
                print(f"No Kimi CLI session matches (home: {kimi_home}).")

        if not matches and not codex_matches and not kimi_matches:
            print("No matches found.")
            _print_search_widening_hint(args)
            sys.exit(0)

    elif args.command == "stats":
        if not args.session_file.exists():
            print(f"Error: Session file not found: {args.session_file}")
            sys.exit(1)

        print(f"Analyzing session: {args.session_file}\n")

        analyzer = SessionAnalyzer(homes=[])
        stats = analyzer.get_session_stats(args.session_file)

        print("=" * 60)
        print("Session Statistics")
        print("=" * 60)
        print("\nMessages:")
        print(f"  Total lines: {stats['total_lines']:,}")
        print(f"  User messages: {stats['user_messages']}")
        print(f"  Assistant messages: {stats['assistant_messages']}")

        print("\nTool Usage:")
        print(f"  Write calls: {stats['write_calls']}")
        print(f"  Edit calls: {stats['edit_calls']}")
        print(f"  Read calls: {stats['read_calls']}")
        print(f"  Bash calls: {stats['bash_calls']}")

        if stats["tool_uses"]:
            print("\n  All tools:")
            for tool, count in sorted(
                stats["tool_uses"].items(), key=lambda x: x[1], reverse=True
            ):
                print(f"    {tool}: {count}")

        if args.show_files and stats["file_operations"]:
            print(f"\nFile Operations ({len(stats['file_operations'])}):")
            # Group by file
            files = defaultdict(list)
            for op, path in stats["file_operations"]:
                files[path].append(op)

            # Limit to 20 files to prevent terminal flooding on large sessions
            for file_path, ops in list(files.items())[:20]:
                filename = Path(file_path).name
                op_summary = ", ".join(
                    f"{op}({ops.count(op)})" for op in set(ops)
                )
                print(f"  {filename}")
                print(f"    Operations: {op_summary}")
                print(f"    Path: {file_path}")

        print()


if __name__ == "__main__":
    main()