"""``muse symlog`` — live per-symbol journal.

Every commit that changes a symbol writes an entry to its journal.  The
journal lives in ``.muse/symlogs/`` and is readable in O(1) per symbol,
independent of total commit count.

Usage::

    muse symlog "src/billing.py::compute_total"            # human text, newest-first
    muse symlog "src/billing.py::compute_total" --json     # machine-readable
    muse symlog "src/billing.py::compute_total" --limit 50
    muse symlog "src/billing.py::compute_total" --follow   # traverse rename chain
    muse symlog "src/billing.py::compute_total" --diff     # include body diff per entry
    muse symlog "src/billing.py::compute_total" --operation symbol-modified
    muse symlog "src/billing.py::compute_total" --author claude-code
    muse symlog "src/billing.py::compute_total" --since 2026-06-01
    muse symlog "src/billing.py::compute_total" --until 2026-07-01
    muse symlog --file src/billing.py                      # all symbols in file
    muse symlog --all                                      # list all symbols with logs
    muse symlog exists "src/billing.py::compute_total"     # exit 0/1
    muse symlog stat "src/billing.py::compute_total"       # summary stats
    muse symlog stat --file src/billing.py                 # stats aggregated over a file
    muse symlog show "src/billing.py::compute_total@{2}"   # header + full body at @{2}
    muse symlog prune-orphans --dry-run                     # report orphaned journals
    muse symlog prune-orphans                                # remove orphaned journals

Each text row shows::

    @{N}  <new_sha12>  (<old_sha12>)  <when>  <author>  <operation>

Null content IDs are rendered as ``"initial"`` (created) or ``"deleted"``
(terminal) so the column is always readable.

JSON schema (single-symbol query)::

    {
      "exit_code":   0,
      "duration_ms": 1.2,
      "symbol":      "src/billing.py::compute_total",
      "total":       12,
      "limit":       20,
      "followed":    false,
      "entries": [
        {
          "index":          0,
          "old_content_id": "sha256:<64-hex>",
          "new_content_id": "sha256:<64-hex>",
          "commit_id":      "sha256:<64-hex>",
          "author":         "claude-code",
          "timestamp":      "2026-06-28T14:22:11+00:00",
          "operation":      "symbol-modified: fix off-by-one in total",
          "born_from":      null,
          "from_symbol":    null,
          "diff":           null
        }
      ]
    }

``from_symbol`` is non-null on entries that came from a prior symbol
address via ``--follow``.  ``diff`` is non-null only when ``--diff`` is
passed.

Exit codes::

    0  Success (including empty log for ``exists``).
    1  Invalid arguments / user error / not found.
    2  Not inside a Muse repository.
"""

from __future__ import annotations

import argparse
import datetime
import difflib
import json
import logging
import re
import sys
from typing import TypedDict

from muse.core.envelope import EnvelopeJson, make_envelope
from muse.core.errors import ExitCode
from muse.core.repo import require_repo
from muse.core.symlog import (
    NULL_CONTENT_ID,
    SymlogEntry,
    SymlogResolution,
    delete_symlog_entry,
    expire_symlog,
    is_null_content_id,
    list_symlog_symbols,
    list_symlog_symbols_for_file,
    prune_orphan_symlogs,
    read_symlog,
    resolve_symlog_addr,
    resolve_symbol_body,
    symlog_path,
)
from muse.core.timing import start_timer
from muse.core.types import long_id, short_id
from muse.core.validation import clamp_int, sanitize_display

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# JSON TypedDicts
# ---------------------------------------------------------------------------


class _SymlogEntryJson(TypedDict):
    index: int
    old_content_id: str
    new_content_id: str
    commit_id: str
    author: str
    timestamp: str
    operation: str
    born_from: str | None
    from_symbol: str | None
    diff: str | None


class _SymlogResultJson(EnvelopeJson):
    symbol: str
    total: int
    limit: int
    followed: bool
    entries: list[_SymlogEntryJson]


class _SymlogFileJson(EnvelopeJson):
    file: str
    symbols: list[_SymlogResultJson]
    count: int


class _SymlogAllJson(EnvelopeJson):
    symbols: list[str]
    count: int


class _ExistsJson(EnvelopeJson):
    exists: bool
    count: int
    symbol: str


class _SymlogStatJson(EnvelopeJson):
    """JSON object returned by ``muse symlog stat`` (muse#53 item 10)."""

    symbol: str | None
    file: str | None
    total_entries: int
    operation_breakdown: dict[str, int]
    most_active_author: str | None
    first_commit_id: str | None
    last_commit_id: str | None
    first_timestamp: str | None
    last_timestamp: str | None


class _PruneOrphansJson(EnvelopeJson):
    """JSON object returned by ``muse symlog prune-orphans`` (muse#53 item 12)."""

    pruned: int
    dry_run: bool
    symbols: list[str]


class _ExpireJson(EnvelopeJson):
    expired: int
    kept: int
    dry_run: bool
    symbols_processed: list[str]


class _DeleteJson(EnvelopeJson):
    deleted: int
    remaining: int
    symbol: str


class _ResolveJson(EnvelopeJson):
    symbol: str
    index: int
    content_id: str
    commit_id: str
    operation: str
    followed_from: str | None


class _ShowJson(EnvelopeJson):
    """JSON object returned by ``muse symlog show`` (muse#53 item 9)."""

    symbol: str
    index: int
    content_id: str
    commit_id: str
    author: str
    operation: str
    followed_from: str | None
    source: str | None


class _DiffJson(EnvelopeJson):
    left: str
    right: str
    symbol: str
    diff_lines: list[str]
    added: int
    removed: int
    context: int


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

_NULL_NEW_LABEL = "deleted"
_NULL_OLD_LABEL = "initial"


def _fmt_cid(cid: str, is_old: bool) -> str:
    """Short-form content ID, or a human sentinel for null IDs."""
    if is_null_content_id(cid):
        return _NULL_OLD_LABEL if is_old else _NULL_NEW_LABEL
    return short_id(cid)


def _fmt_entry_text(idx: int, entry: SymlogEntry, from_symbol: str | None = None) -> str:
    """Format one symlog entry for terminal output."""
    new_short = sanitize_display(_fmt_cid(entry.new_content_id, is_old=False))
    old_short = sanitize_display(_fmt_cid(entry.old_content_id, is_old=True))
    when = entry.timestamp.strftime("%Y-%m-%d %H:%M:%S UTC")
    safe_op = sanitize_display(entry.operation)
    safe_author = sanitize_display(entry.author or "")
    author_col = f"  {safe_author}" if safe_author else ""
    line = f"@{{{idx}}}  {new_short}  ({old_short})  {when}{author_col}  {safe_op}"
    if from_symbol:
        line += f"  [from {sanitize_display(from_symbol)}]"
    return line


def _parse_date(value: str, flag: str) -> datetime.datetime:
    """Parse ``YYYY-MM-DD`` into a UTC-aware datetime; exits USER_ERROR on bad format."""
    try:
        d = datetime.date.fromisoformat(value)
    except ValueError:
        print(
            f"❌ Invalid date for {flag}: {sanitize_display(value)!r} — "
            "expected YYYY-MM-DD.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    return datetime.datetime(d.year, d.month, d.day, tzinfo=datetime.timezone.utc)


def _validate_symbol_addr(addr: str) -> None:
    """Validate symbol address format; exits USER_ERROR if invalid."""
    if not addr:
        print(
            "❌ Symbol address is empty — expected form: 'path/to/file.py::symbol_name'.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    if "::" not in addr:
        print(
            f"❌ Missing '::' separator in symbol address {sanitize_display(addr)!r}. "
            "Expected form: 'path/to/file.py::symbol_name'.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    _file, _, sym_name = addr.partition("::")
    if not sym_name:
        print(
            f"❌ Empty symbol name in {sanitize_display(addr)!r}. "
            "Expected form: 'path/to/file.py::symbol_name'.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    if ".." in addr.split("::")[0].split("/"):
        print(
            f"❌ Path traversal detected in {sanitize_display(addr)!r}: "
            "'..' components are not allowed.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)


def _compute_diff(
    old_cid: str,
    new_cid: str,
    repo_root,
    symbol_addr: str | None = None,
    entry: "SymlogEntry | None" = None,
) -> str | None:
    """Produce a unified diff of the symbol body between the old and new commit.

    Symbol content IDs are hashes computed by the AST parser and are not stored
    as raw objects in the Muse object store, so we reconstruct each body by
    reading the file from the commit's snapshot and extracting the symbol.
    Returns ``None`` when either side is null (created/deleted) or on any error.
    """
    if is_null_content_id(old_cid) or is_null_content_id(new_cid):
        return None
    if symbol_addr is None or entry is None:
        return None

    file_path, _, symbol_name = symbol_addr.partition("::")
    if not symbol_name:
        return None

    try:
        from muse.core.commits import read_commit
        from muse.core.object_store import read_object
        from muse.core.snapshots import read_snapshot
        from muse.plugins.code.ast_parser import parse_symbols

        # Read the new commit and its snapshot.
        new_record = read_commit(repo_root, entry.commit_id)
        if new_record is None or not new_record.parent_commit_id:
            return None
        new_snap = read_snapshot(repo_root, new_record.snapshot_id)
        if new_snap is None:
            return None
        new_file_oid = new_snap.manifest.get(file_path)
        if new_file_oid is None:
            return None
        new_file_bytes = read_object(repo_root, new_file_oid)
        if new_file_bytes is None:
            return None

        # Read the parent commit and its snapshot (the "old" state).
        old_record = read_commit(repo_root, new_record.parent_commit_id)
        if old_record is None:
            return None
        old_snap = read_snapshot(repo_root, old_record.snapshot_id)
        if old_snap is None:
            return None
        old_file_oid = old_snap.manifest.get(file_path)
        if old_file_oid is None:
            return None
        old_file_bytes = read_object(repo_root, old_file_oid)
        if old_file_bytes is None:
            return None

        # Extract the symbol body from each file version using line numbers.
        full_addr = f"{file_path}::{symbol_name}"
        new_syms = parse_symbols(new_file_bytes, file_path)
        old_syms = parse_symbols(old_file_bytes, file_path)

        def _body_text(syms, raw_bytes: bytes) -> str:
            rec = syms.get(full_addr)
            if rec is None:
                return ""
            lines = raw_bytes.decode("utf-8", errors="replace").splitlines()
            start = rec["lineno"] - 1   # lineno is 1-based, inclusive
            end = rec["end_lineno"]     # end_lineno is 1-based, inclusive → exclusive slice
            return "\n".join(lines[start:end])

        new_body = _body_text(new_syms, new_file_bytes)
        old_body = _body_text(old_syms, old_file_bytes)

        if old_body == new_body:
            return ""
        old_lines = old_body.splitlines(keepends=True)
        new_lines = new_body.splitlines(keepends=True)
        diff_lines = list(difflib.unified_diff(
            old_lines, new_lines,
            fromfile=f"a/{symbol_name}",
            tofile=f"b/{symbol_name}",
            lineterm="",
        ))
        return "\n".join(diff_lines) if diff_lines else ""
    except Exception:
        return None


def _apply_filters(
    entries: list[SymlogEntry],
    operation_filter: str | None,
    author_filter: str | None,
    since_dt: datetime.datetime | None,
    until_dt: datetime.datetime | None,
) -> list[SymlogEntry]:
    """Apply all active filters to the entry list."""
    filtered = entries

    if operation_filter is not None:
        needle = operation_filter.lower()
        filtered = [e for e in filtered if needle in e.operation.lower()]

    if author_filter is not None:
        needle_a = author_filter.lower()
        filtered = [e for e in filtered if needle_a in (e.author or "").lower()]

    if since_dt is not None:
        filtered = [e for e in filtered if e.timestamp >= since_dt]

    if until_dt is not None:
        until_end = until_dt + datetime.timedelta(days=1)
        filtered = [e for e in filtered if e.timestamp < until_end]

    return filtered


def _entries_to_json(
    entries: list[SymlogEntry],
    from_symbol_map: dict[int, str],
    include_diff: bool,
    repo_root,
    symbol_addr: str | None = None,
) -> list[_SymlogEntryJson]:
    """Convert SymlogEntry list to the JSON schema."""
    out: list[_SymlogEntryJson] = []
    for idx, entry in enumerate(entries):
        diff_text: str | None = None
        if include_diff:
            diff_text = _compute_diff(
                entry.old_content_id, entry.new_content_id, repo_root,
                symbol_addr=symbol_addr, entry=entry,
            )
        out.append(_SymlogEntryJson(
            index=idx,
            old_content_id=long_id(entry.old_content_id),
            new_content_id=long_id(entry.new_content_id),
            commit_id=long_id(entry.commit_id),
            author=entry.author,
            timestamp=entry.timestamp.isoformat(),
            operation=entry.operation,
            born_from=entry.born_from,
            from_symbol=from_symbol_map.get(idx),
            diff=diff_text,
        ))
    return out


def _read_with_follow(
    repo_root,
    symbol_addr: str,
    follow: bool,
    limit: int,
) -> tuple[list[SymlogEntry], dict[int, str]]:
    """Read symlog entries with optional follow, returning entries + from_symbol map.

    When ``follow=True``, entries that came from a prior symbol address
    (via a rename chain) are annotated in the ``from_symbol`` map
    (index → prior_addr).
    """
    if not follow:
        entries = read_symlog(repo_root, symbol_addr, limit=limit, follow=False)
        return entries, {}

    # Read with follow from the core — it appends prior entries.
    # We need to identify which entries came from the prior symbol.
    # Strategy: read without follow first, then read the prior symbol separately
    # and mark those entries with their origin address.
    own_entries = read_symlog(repo_root, symbol_addr, limit=limit * 10, follow=False)
    from_symbol_map: dict[int, str] = {}

    if own_entries:
        oldest = own_entries[-1]
        if oldest.born_from:
            prior_addr = oldest.born_from
            # Recursively read with follow to get the full chain.
            prior_entries, prior_map = _read_with_follow(
                repo_root, prior_addr, follow=True, limit=limit * 10
            )
            # Merge: own entries first (newer), then prior entries (older).
            combined = own_entries + prior_entries
            # Build from_symbol map: prior entries get from_symbol = prior_addr
            for i in range(len(own_entries), len(combined)):
                # The prior entry itself may come from an even earlier symbol;
                # use the prior_map to get its annotation, falling back to prior_addr.
                prior_idx = i - len(own_entries)
                from_symbol_map[i] = prior_map.get(prior_idx, prior_addr)
            return combined, from_symbol_map

    return own_entries, {}


# ---------------------------------------------------------------------------
# Argument parser registration
# ---------------------------------------------------------------------------


def register(
    subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
) -> None:
    """Register the ``symlog`` subcommand tree."""
    parser = subparsers.add_parser(
        "symlog",
        help="Show the live per-symbol journal written at commit time.",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "positional", nargs="*", metavar="FILE::SYMBOL",
        help=(
            "Symbol address, e.g. 'src/billing.py::compute_total'. "
            "Or 'exists FILE::SYMBOL' to run the existence check."
        ),
    )
    parser.add_argument(
        "--limit", type=int, default=20,
        help="Maximum number of entries to show (after filters, default: 20).",
    )
    parser.add_argument(
        "--follow", action="store_true", dest="follow",
        help="Traverse rename chain — follow born-from pointers to prior symbol addresses.",
    )
    parser.add_argument(
        "--diff", action="store_true", dest="include_diff",
        help="Include unified body diff per entry (--json only for full content).",
    )
    parser.add_argument(
        "--file", default=None, metavar="FILE_PATH", dest="file_path",
        help="Show symlogs for all symbols in FILE_PATH.",
    )
    parser.add_argument(
        "--all", action="store_true", dest="all_symbols",
        help="List all symbols that have a symlog.",
    )
    parser.add_argument(
        "--operation", default=None, metavar="PATTERN", dest="operation_filter",
        help="Filter to entries whose operation contains PATTERN (case-insensitive).",
    )
    parser.add_argument(
        "--author", default=None, metavar="PATTERN", dest="author_filter",
        help="Filter to entries whose author contains PATTERN (case-insensitive).",
    )
    parser.add_argument(
        "--since", default=None, metavar="YYYY-MM-DD", dest="since",
        help="Show only entries on or after this date.",
    )
    parser.add_argument(
        "--until", default=None, metavar="YYYY-MM-DD", dest="until",
        help="Show only entries on or before this date.",
    )
    parser.add_argument(
        "--expire-days", type=int, default=None, metavar="N", dest="expire_days",
        help=(
            "Prune entries older than N days.  Defaults to the symlog.expire-days "
            "config key, then 90.  Used by ``expire`` only."
        ),
    )
    parser.add_argument(
        "--dry-run", "-n", action="store_true", dest="dry_run",
        help="Report counts without writing any files.  Used by ``expire`` only.",
    )
    parser.add_argument(
        "--json", "-j", action="store_true", dest="json_out",
        help="Emit machine-readable JSON instead of human text.",
    )

    parser.set_defaults(func=run)


# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

_AT_INDEX_RE = re.compile(r"^@\{(\d+)\}$")
#: Mirrors muse.core.symlog._SYMLOG_REF_RE -- accepts a digit index OR a
#: relative-time spec inside the braces (muse#53 item 6); disambiguation
#: happens inside resolve_symlog_addr()/_resolve_symlog_index().
_SYMLOG_REF_RE = re.compile(r"^(.+)@\{([^}]+)\}$")
_DEFAULT_SYMLOG_EXPIRE_DAYS: int = 90

# ---------------------------------------------------------------------------
# exists subcommand
# ---------------------------------------------------------------------------


def run_exists(symbol: str, json_out: bool) -> None:
    """Fast existence check — does this symbol have any symlog entries?

    Exits 0 when entries are present, 1 when none.

    JSON schema::

        {
          "exit_code":   0 or 1,
          "duration_ms": 0.3,
          "symbol":      "src/billing.py::compute_total",
          "exists":      true,
          "count":       7
        }
    """
    elapsed = start_timer()

    _validate_symbol_addr(symbol)

    repo_root = require_repo()

    entries = read_symlog(repo_root, symbol, limit=100_000, follow=False)
    count = len(entries)
    exists = count > 0

    if json_out:
        payload = _ExistsJson(
            **make_envelope(elapsed, exit_code=0 if exists else 1),
            exists=exists,
            count=count,
            symbol=symbol,
        )
        print(json.dumps(payload))
    else:
        safe_sym = sanitize_display(symbol)
        if exists:
            print(f"{safe_sym} has {count} symlog entr{'y' if count == 1 else 'ies'}.")
        else:
            print(f"No symlog entries for {safe_sym}.")

    if not exists:
        raise SystemExit(1)


# ---------------------------------------------------------------------------
# stat subcommand (muse#53 item 10)
# ---------------------------------------------------------------------------


def _categorize_symlog_operation(operation: str) -> str:
    """Bucket a symlog operation string by its leading ``symbol-<kind>:`` prefix."""
    return operation.split(":", 1)[0].strip().lower() if ":" in operation else operation.strip().lower()


def run_stat(args: argparse.Namespace) -> None:
    """Summary statistics for a symbol's (or a file's) symlog (muse#53 item 10).

    Either a single symbol address (``muse symlog stat "file.py::Fn"``) or
    ``--file FILE_PATH`` (aggregating every symbol's journal in that file)
    must be given.

    JSON schema::

        {
          "exit_code":           0,
          "duration_ms":         1.1,
          "symbol":              "src/billing.py::compute_total",
          "file":                null,
          "total_entries":       12,
          "operation_breakdown": {"symbol-created": 1, "symbol-modified": 10, "symbol-deleted": 1},
          "most_active_author":  "claude-code",
          "first_commit_id":     "sha256:<64-hex>",
          "last_commit_id":      "sha256:<64-hex>",
          "first_timestamp":     "2026-01-01T00:00:00+00:00",
          "last_timestamp":      "2026-03-16T12:00:00+00:00"
        }

    An empty symlog reports ``total_entries: 0``, an empty
    ``operation_breakdown``, and every other field ``null``.

    Exit codes::

        0  Success (including an empty symlog).
        1  Invalid arguments (missing symbol/--file, bad symbol address).
        2  Not inside a Muse repository.
    """
    elapsed = start_timer()
    positional: list[str] = getattr(args, "positional", [])
    json_out: bool = getattr(args, "json_out", False)
    file_path: str | None = getattr(args, "file_path", None)
    symbol: str | None = positional[1] if len(positional) > 1 else None

    if symbol is not None and file_path is not None:
        print("❌ Specify either a symbol address or --file, not both.", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    if symbol is None and file_path is None:
        print(
            "❌ 'stat' requires a symbol address or --file.\n"
            "   Example: muse symlog stat 'src/billing.py::compute_total'\n"
            "   Example: muse symlog stat --file src/billing.py",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    repo_root = require_repo()

    entries: list[SymlogEntry] = []
    if symbol is not None:
        _validate_symbol_addr(symbol)
        entries = read_symlog(repo_root, symbol, limit=100_000, follow=False)
    else:
        assert file_path is not None
        for addr in list_symlog_symbols_for_file(repo_root, file_path):
            entries.extend(read_symlog(repo_root, addr, limit=100_000, follow=False))
        # Merge-sort by timestamp descending (newest-first), matching
        # read_symlog's own per-symbol ordering contract.
        entries.sort(key=lambda e: e.timestamp, reverse=True)

    operation_breakdown: dict[str, int] = {}
    author_counts: dict[str, int] = {}
    for entry in entries:
        category = _categorize_symlog_operation(entry.operation)
        operation_breakdown[category] = operation_breakdown.get(category, 0) + 1
        if entry.author:
            author_counts[entry.author] = author_counts.get(entry.author, 0) + 1

    most_active_author: str | None = None
    if author_counts:
        most_active_author = max(author_counts.items(), key=lambda kv: kv[1])[0]

    # entries are newest-first (per-symbol read_symlog contract, and the
    # file-mode merge above re-sorts to match).
    first_commit_id = entries[-1].commit_id if entries else None
    last_commit_id = entries[0].commit_id if entries else None
    first_timestamp = entries[-1].timestamp.isoformat() if entries else None
    last_timestamp = entries[0].timestamp.isoformat() if entries else None

    if json_out:
        payload = _SymlogStatJson(
            **make_envelope(elapsed),
            symbol=symbol,
            file=file_path,
            total_entries=len(entries),
            operation_breakdown=operation_breakdown,
            most_active_author=most_active_author,
            first_commit_id=first_commit_id,
            last_commit_id=last_commit_id,
            first_timestamp=first_timestamp,
            last_timestamp=last_timestamp,
        )
        print(json.dumps(payload))
        return

    label = sanitize_display(symbol) if symbol is not None else sanitize_display(file_path or "")
    if not entries:
        print(f"No symlog entries for {label}.")
        return

    print(f"Symlog stats for {label}\n")
    print(f"  Total entries:  {len(entries)}")
    print(f"  Most active:    {sanitize_display(most_active_author) if most_active_author else '(none)'}")
    print(f"  First entry:    {first_timestamp}")
    print(f"  Last entry:     {last_timestamp}")
    print("\n  Operation breakdown:")
    for category, count in sorted(operation_breakdown.items(), key=lambda kv: -kv[1]):
        print(f"    {sanitize_display(category):<20} {count}")


# ---------------------------------------------------------------------------
# prune-orphans subcommand (muse#53 item 12)
# ---------------------------------------------------------------------------


def run_prune_orphans(args: argparse.Namespace) -> None:
    """Find and remove symlog journals whose symbol no longer exists at HEAD.

    A journal becomes orphaned when its file is deleted/renamed away, or the
    symbol itself is removed from that file — the journal (append-only,
    never auto-deleted) survives with no live symbol to show it for.
    Analogous to ``muse gc`` for unreachable objects.

    JSON schema::

        {
          "exit_code":  0,
          "duration_ms": 3.2,
          "pruned":     2,
          "dry_run":    false,
          "symbols":    ["src/old.py::removed_fn", "src/gone.py::deleted_class"]
        }

    Exit codes::

        0  Success (including zero orphans).
        2  Not inside a Muse repository.
    """
    elapsed = start_timer()
    json_out: bool = getattr(args, "json_out", False)
    dry_run: bool = getattr(args, "dry_run", False)

    repo_root = require_repo()

    orphans = prune_orphan_symlogs(repo_root, dry_run=dry_run)

    if json_out:
        payload = _PruneOrphansJson(
            **make_envelope(elapsed),
            pruned=len(orphans),
            dry_run=dry_run,
            symbols=orphans,
        )
        print(json.dumps(payload))
        return

    if not orphans:
        print("No orphaned symlog journals found.")
        return

    verb = "Would prune" if dry_run else "Pruned"
    print(f"{verb} {len(orphans)} orphaned symlog journal(s):")
    for addr in orphans:
        print(f"  {sanitize_display(addr)}")


# ---------------------------------------------------------------------------
# expire subcommand
# ---------------------------------------------------------------------------


def run_expire(args: argparse.Namespace) -> None:
    """Prune entries older than --expire-days from one or more symbol journals.

    Dispatches to single-symbol, --file, or --all mode depending on flags.

    JSON schema::

        {
          "exit_code":         0,
          "duration_ms":       3.1,
          "expired":           34,
          "kept":              18,
          "dry_run":           false,
          "symbols_processed": ["src/billing.py::compute_total"]
        }
    """
    elapsed = start_timer()

    positional: list[str] = getattr(args, "positional", [])
    file_path: str | None = getattr(args, "file_path", None)
    all_symbols: bool = getattr(args, "all_symbols", False)
    dry_run: bool = getattr(args, "dry_run", False)
    json_out: bool = getattr(args, "json_out", False)
    expire_days_arg: int | None = getattr(args, "expire_days", None)

    # Resolve symbol address (everything in positional after "expire")
    expire_positional = positional[1:]
    symbol: str | None = expire_positional[0] if expire_positional else None

    # Resolve TTL: explicit flag → config → default
    if expire_days_arg is not None:
        expire_days = expire_days_arg
    else:
        repo_root_early = require_repo()
        from muse.cli.config import get_config_value as _gcv
        _raw = _gcv("symlog.expire-days", repo_root_early)
        try:
            expire_days = int(_raw) if _raw is not None else _DEFAULT_SYMLOG_EXPIRE_DAYS
        except (ValueError, TypeError):
            expire_days = _DEFAULT_SYMLOG_EXPIRE_DAYS

    repo_root = require_repo()

    total_expired = 0
    total_kept = 0
    symbols_processed: list[str] = []

    if symbol and not file_path and not all_symbols:
        # Single-symbol mode
        _validate_symbol_addr(symbol)
        expired, kept = expire_symlog(repo_root, symbol, expire_days, dry_run=dry_run)
        total_expired += expired
        total_kept += kept
        if expired > 0 or kept > 0:
            symbols_processed.append(symbol)

    elif file_path is not None:
        # --file mode
        addrs = list_symlog_symbols_for_file(repo_root, file_path)
        for addr in addrs:
            expired, kept = expire_symlog(repo_root, addr, expire_days, dry_run=dry_run)
            total_expired += expired
            total_kept += kept
            symbols_processed.append(addr)

    elif all_symbols:
        # --all mode
        addrs = list_symlog_symbols(repo_root)
        for addr in addrs:
            expired, kept = expire_symlog(repo_root, addr, expire_days, dry_run=dry_run)
            total_expired += expired
            total_kept += kept
            symbols_processed.append(addr)

    else:
        print(
            "❌ Specify a symbol address, --file FILE, or --all.\n"
            "   Example: muse symlog expire 'src/billing.py::compute_total' --expire-days 90",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    if json_out:
        payload = _ExpireJson(
            **make_envelope(elapsed),
            expired=total_expired,
            kept=total_kept,
            dry_run=dry_run,
            symbols_processed=symbols_processed,
        )
        print(json.dumps(payload))
    else:
        prefix = "[dry-run] " if dry_run else ""
        action = "Would expire" if dry_run else "Expired"
        print(
            f"{prefix}{action} {total_expired} entr{'y' if total_expired == 1 else 'ies'}, "
            f"kept {total_kept} across {len(symbols_processed)} symbol(s)."
        )


# ---------------------------------------------------------------------------
# delete subcommand
# ---------------------------------------------------------------------------


def run_delete(args: argparse.Namespace) -> None:
    """Remove one entry by index or all entries from a symbol journal.

    JSON schema::

        {
          "exit_code":  0,
          "duration_ms": 0.8,
          "deleted":    1,
          "remaining":  7,
          "symbol":     "src/billing.py::compute_total"
        }
    """
    elapsed = start_timer()

    positional: list[str] = getattr(args, "positional", [])
    all_symbols: bool = getattr(args, "all_symbols", False)
    json_out: bool = getattr(args, "json_out", False)

    # positional: ["delete", SYMBOL, optional @{N}]
    delete_positional = positional[1:]  # everything after "delete"

    if not delete_positional:
        print(
            "❌ 'delete' requires a symbol address.\n"
            "   Example: muse symlog delete 'src/billing.py::compute_total' @{0}\n"
            "   Example: muse symlog delete 'src/billing.py::compute_total' --all",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    symbol = delete_positional[0]
    _validate_symbol_addr(symbol)

    # Parse optional @{N} token
    index: int | None = None
    if len(delete_positional) >= 2:
        token = delete_positional[1]
        m = _AT_INDEX_RE.match(token)
        if m is None:
            print(
                f"❌ Unrecognised index reference {sanitize_display(token)!r} — "
                "expected @{N} where N is a non-negative integer.",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)
        index = int(m.group(1))

    if index is None and not all_symbols:
        print(
            "❌ Specify @{N} to delete one entry or --all to delete all entries.\n"
            "   Example: muse symlog delete 'src/billing.py::compute_total' @{0}\n"
            "   Example: muse symlog delete 'src/billing.py::compute_total' --all",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    repo_root = require_repo()

    try:
        deleted, remaining = delete_symlog_entry(
            repo_root, symbol, index if not all_symbols else None
        )
    except FileNotFoundError:
        print(
            f"❌ No symlog entries for {sanitize_display(symbol)!r}.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    except IndexError as exc:
        bad_index, total = exc.args[0], exc.args[1]
        print(
            f"❌ @{{{bad_index}}} is out of range — "
            f"valid range: @{{0}} to @{{{total - 1}}} ({total} entr{'y' if total == 1 else 'ies'}).",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    if json_out:
        payload = _DeleteJson(
            **make_envelope(elapsed),
            deleted=deleted,
            remaining=remaining,
            symbol=symbol,
        )
        print(json.dumps(payload))
    else:
        safe_sym = sanitize_display(symbol)
        print(f"Deleted {deleted} entr{'y' if deleted == 1 else 'ies'} from {safe_sym} ({remaining} remaining).")


# ---------------------------------------------------------------------------
# resolve subcommand
# ---------------------------------------------------------------------------


def run_resolve(args: argparse.Namespace) -> None:
    """Resolve a ``<addr>@{N}`` symlog reference to its content_id and commit_id.

    JSON schema::

        {
          "exit_code":    0,
          "duration_ms":  0.4,
          "symbol":       "src/billing.py::compute_total",
          "index":        2,
          "content_id":   "sha256:<64-hex>",
          "commit_id":    "sha256:<64-hex>",
          "operation":    "symbol-modified: add rounding",
          "timestamp":    "2026-06-20T09:11:04+00:00",
          "followed_from": null
        }
    """
    elapsed = start_timer()

    positional: list[str] = getattr(args, "positional", [])
    follow: bool = getattr(args, "follow", False)
    json_out: bool = getattr(args, "json_out", False)

    # positional: ["resolve", SPEC]
    resolve_positional = positional[1:]

    if not resolve_positional:
        print(
            "❌ 'resolve' requires a symbol ref: muse symlog resolve 'file.py::Sym@{N}' [--follow]",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    spec = resolve_positional[0]

    # Validate the spec matches addr@{N} form.
    m = _SYMLOG_REF_RE.match(spec)
    if m is None:
        print(
            f"❌ {sanitize_display(spec)!r} is not a valid symlog ref — "
            "expected form: 'file.py::Symbol@{N}'.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    repo_root = require_repo()

    try:
        resolution = resolve_symlog_addr(spec, repo_root, follow=follow)
    except FileNotFoundError:
        addr = m.group(1)
        print(
            f"❌ No symlog entries for {sanitize_display(addr)!r}.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    except IndexError as exc:
        bad_index, total = exc.args[0], exc.args[1]
        print(
            f"❌ @{{{bad_index}}} is out of range — "
            f"valid range: @{{0}} to @{{{max(0, total - 1)}}} "
            f"({total} entr{'y' if total == 1 else 'ies'}).",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    except ValueError as exc:
        print(f"❌ Invalid symbol address: {sanitize_display(str(exc))}", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    # resolution cannot be None here since _SYMLOG_REF_RE matched.
    assert resolution is not None

    if json_out:
        payload = _ResolveJson(
            **make_envelope(elapsed),
            symbol=resolution.symbol,
            index=resolution.index,
            content_id=resolution.content_id,
            commit_id=resolution.commit_id,
            operation=resolution.operation,
            followed_from=resolution.followed_from,
        )
        # Overwrite the envelope's wall-clock timestamp with the symlog
        # entry's own recorded timestamp so callers see when the change
        # happened, not when this command ran.
        payload["timestamp"] = resolution.timestamp.isoformat()
        print(json.dumps(payload))
    else:
        safe_sym = sanitize_display(resolution.symbol)
        safe_op = sanitize_display(resolution.operation)
        when = resolution.timestamp.strftime("%Y-%m-%d %H:%M:%S UTC")
        print(f"{safe_sym}@{{{resolution.index}}}")
        print(f"  content_id:   {sanitize_display(resolution.content_id)}")
        print(f"  commit_id:    {sanitize_display(resolution.commit_id)}")
        print(f"  operation:    {safe_op}")
        print(f"  timestamp:    {when}")
        if resolution.followed_from:
            print(f"  followed_from: {sanitize_display(resolution.followed_from)}")


# ---------------------------------------------------------------------------
# show subcommand (muse#53 item 9)
# ---------------------------------------------------------------------------


def run_show(args: argparse.Namespace) -> None:
    """Human-focused convenience view of a ``<addr>@{N}`` symlog reference.

    Equivalent to ``muse code cat "addr@{N}"`` but with a header (operation,
    author, timestamp) printed before the full symbol body — useful for
    interactive exploration ("show me what this function looked like two
    changes ago") without needing ``--json`` and a second lookup.

    JSON schema::

        {
          "exit_code":     0,
          "duration_ms":   0.6,
          "symbol":        "src/billing.py::compute_total",
          "index":         2,
          "content_id":    "sha256:<64-hex>",
          "commit_id":     "sha256:<64-hex>",
          "author":        "claude-code",
          "operation":     "symbol-modified: add rounding",
          "followed_from": null,
          "source":        "def compute_total(items):\n    return sum(items)"
        }

    ``source`` is ``null`` when the body could not be reconstructed (e.g. the
    entry is a deletion, or the file/commit no longer resolves).

    Exit codes::

        0  Success.
        1  Invalid ref, no symlog, or index out of range.
        2  Not inside a Muse repository.
    """
    elapsed = start_timer()

    positional: list[str] = getattr(args, "positional", [])
    follow: bool = getattr(args, "follow", False)
    json_out: bool = getattr(args, "json_out", False)

    # positional: ["show", SPEC]
    show_positional = positional[1:]

    if not show_positional:
        print(
            "❌ 'show' requires a symbol ref: muse symlog show 'file.py::Sym@{N}' [--follow]",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    spec = show_positional[0]

    m = _SYMLOG_REF_RE.match(spec)
    if m is None:
        print(
            f"❌ {sanitize_display(spec)!r} is not a valid symlog ref — "
            "expected form: 'file.py::Symbol@{N}'.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    repo_root = require_repo()

    try:
        resolution = resolve_symlog_addr(spec, repo_root, follow=follow)
    except FileNotFoundError:
        addr = m.group(1)
        print(f"❌ No symlog entries for {sanitize_display(addr)!r}.", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)
    except IndexError as exc:
        bad_index, total = exc.args[0], exc.args[1]
        print(
            f"❌ @{{{bad_index}}} is out of range — "
            f"valid range: @{{0}} to @{{{max(0, total - 1)}}} "
            f"({total} entr{'y' if total == 1 else 'ies'}).",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    except ValueError as exc:
        print(f"❌ Invalid symbol address: {sanitize_display(str(exc))}", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    assert resolution is not None

    body = resolve_symbol_body(repo_root, resolution.symbol, resolution.commit_id)
    source = body["source"] if body is not None else None

    if json_out:
        payload = _ShowJson(
            **make_envelope(elapsed),
            symbol=resolution.symbol,
            index=resolution.index,
            content_id=resolution.content_id,
            commit_id=resolution.commit_id,
            author=resolution.author,
            operation=resolution.operation,
            followed_from=resolution.followed_from,
            source=source,
        )
        print(json.dumps(payload))
        return

    safe_sym = sanitize_display(resolution.symbol)
    safe_op = sanitize_display(resolution.operation)
    safe_author = sanitize_display(resolution.author)
    when = resolution.timestamp.strftime("%Y-%m-%d %H:%M:%S UTC")
    print(f"{safe_sym}@{{{resolution.index}}}")
    print(f"  commit:    {sanitize_display(resolution.commit_id)}")
    print(f"  author:    {safe_author}")
    print(f"  when:      {when}")
    print(f"  operation: {safe_op}")
    if resolution.followed_from:
        print(f"  followed_from: {sanitize_display(resolution.followed_from)}")
    print()
    if source is None:
        print("  (no body available — symbol was deleted, or source could not be reconstructed)")
    else:
        for line in source.splitlines():
            print(f"  {line}")


# ---------------------------------------------------------------------------
# diff subcommand
# ---------------------------------------------------------------------------


def _get_symbol_body_from_workdir(
    repo_root: "pathlib.Path",
    symbol_addr: str,
) -> "str | None":
    """Return the current working-tree body for *symbol_addr*, or None on failure."""
    if "::" not in symbol_addr:
        return None
    file_path, _, symbol_name = symbol_addr.partition("::")
    if not symbol_name:
        return None

    try:
        import pathlib as _pathlib
        from muse.plugins.code.ast_parser import adapter_for_path

        disk_path = repo_root / file_path
        if not disk_path.is_file():
            return None
        raw = disk_path.read_bytes()
        adapter = adapter_for_path(file_path)
        tree = adapter.parse_symbols(raw, file_path)

        found = next(
            (r for r in tree.values() if r["qualified_name"] == symbol_name),
            None,
        )
        if found is None:
            found = next(
                (r for r in tree.values()
                 if r["name"] == symbol_name and r["kind"] != "import"),
                None,
            )
        if found is None:
            return None

        lineno = found["lineno"]
        end_lineno = found["end_lineno"]
        text = raw.decode("utf-8", errors="replace")
        lines = text.splitlines()
        return "\n".join(lines[lineno - 1:end_lineno])
    except Exception:
        return None


def run_diff(args: argparse.Namespace) -> None:
    """Show the unified diff between two symlog-indexed symbol versions.

    JSON schema::

        {
          "exit_code":  0,
          "duration_ms": 0.8,
          "left":        "billing.py::compute_total@{1}",
          "right":       "billing.py::compute_total@{0}",
          "symbol":      "billing.py::compute_total",
          "diff_lines":  ["-old line", "+new line", " context"],
          "added":       1,
          "removed":     1,
          "context":     5
        }
    """
    elapsed = start_timer()

    positional: list[str] = getattr(args, "positional", [])
    follow: bool = getattr(args, "follow", False)
    json_out: bool = getattr(args, "json_out", False)

    # positional: ["diff", LEFT_SPEC, RIGHT_SPEC]
    diff_positional = positional[1:]

    if len(diff_positional) < 2:
        print(
            "❌ 'diff' requires two specs:\n"
            "   muse symlog diff 'file.py::Sym@{1}' 'file.py::Sym@{0}'\n"
            "   muse symlog diff 'file.py::Sym@{2}' HEAD",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    left_spec = diff_positional[0]
    right_spec = diff_positional[1]

    # Parse left spec — must be addr@{N}.
    left_m = _SYMLOG_REF_RE.match(left_spec)
    if left_m is None:
        print(
            f"❌ Left spec {sanitize_display(left_spec)!r} must be 'addr@{{N}}'.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    left_addr = left_m.group(1)
    right_from_head = right_spec.upper() == "HEAD"

    # Parse right spec — addr@{M} or HEAD.
    if not right_from_head:
        right_m = _SYMLOG_REF_RE.match(right_spec)
        if right_m is None:
            print(
                f"❌ Right spec {sanitize_display(right_spec)!r} must be 'addr@{{N}}' or HEAD.",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)
        right_addr = right_m.group(1)
    else:
        right_addr = left_addr

    repo_root = require_repo()

    # Resolve left body from symlog.
    try:
        left_res = resolve_symlog_addr(left_spec, repo_root, follow=follow)
    except FileNotFoundError:
        print(
            f"❌ No symlog entries for {sanitize_display(left_addr)!r}.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    except IndexError as exc:
        bad_index, total = exc.args[0], exc.args[1]
        print(
            f"❌ Left @{{{bad_index}}} is out of range — "
            f"valid range: @{{0}} to @{{{max(0, total - 1)}}}.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    except ValueError as exc:
        print(f"❌ Invalid left symbol address: {sanitize_display(str(exc))}", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    assert left_res is not None

    left_body_info = resolve_symbol_body(repo_root, left_addr, left_res.commit_id)
    if left_body_info is None:
        print(
            f"❌ Could not reconstruct body for {sanitize_display(left_addr)!r} "
            f"at commit {sanitize_display(left_res.commit_id[:8])}.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)
    left_body = left_body_info["source"]

    # Resolve right body.
    if right_from_head:
        right_body = _get_symbol_body_from_workdir(repo_root, right_addr)
        if right_body is None:
            print(
                f"❌ Could not read working-tree body for {sanitize_display(right_addr)!r}.",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)
    else:
        try:
            right_res = resolve_symlog_addr(right_spec, repo_root, follow=follow)
        except FileNotFoundError:
            print(
                f"❌ No symlog entries for {sanitize_display(right_addr)!r}.",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)
        except IndexError as exc:
            bad_index, total = exc.args[0], exc.args[1]
            print(
                f"❌ Right @{{{bad_index}}} is out of range — "
                f"valid range: @{{0}} to @{{{max(0, total - 1)}}}.",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)
        except ValueError as exc:
            print(f"❌ Invalid right symbol address: {sanitize_display(str(exc))}", file=sys.stderr)
            raise SystemExit(ExitCode.USER_ERROR)

        assert right_res is not None
        right_body_info = resolve_symbol_body(repo_root, right_addr, right_res.commit_id)
        if right_body_info is None:
            print(
                f"❌ Could not reconstruct body for {sanitize_display(right_addr)!r}.",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)
        right_body = right_body_info["source"]

    # Compute unified diff.
    symbol_name = left_addr.split("::")[-1] if "::" in left_addr else left_addr
    left_lines = left_body.splitlines(keepends=True)
    right_lines = right_body.splitlines(keepends=True)
    raw_diff = list(difflib.unified_diff(
        left_lines, right_lines,
        fromfile=f"a/{symbol_name}",
        tofile=f"b/{symbol_name}",
        lineterm="",
    ))

    # Count added/removed/context lines (skip --- and +++ header lines).
    added = sum(1 for ln in raw_diff if ln.startswith("+") and not ln.startswith("+++"))
    removed = sum(1 for ln in raw_diff if ln.startswith("-") and not ln.startswith("---"))
    context = sum(1 for ln in raw_diff if ln.startswith(" "))

    if json_out:
        payload = _DiffJson(
            **make_envelope(elapsed),
            left=left_spec,
            right=right_spec,
            symbol=left_addr,
            diff_lines=raw_diff,
            added=added,
            removed=removed,
            context=context,
        )
        print(json.dumps(payload))
    else:
        if not raw_diff:
            print(f"No differences between {sanitize_display(left_spec)} and {sanitize_display(right_spec)}.")
        else:
            for line in raw_diff:
                print(line)


# ---------------------------------------------------------------------------
# Command entry point
# ---------------------------------------------------------------------------


def run(args: argparse.Namespace) -> None:
    """Show the live per-symbol journal.

    Dispatches to ``exists``, ``--file``, ``--all``, or single-symbol mode
    depending on the flags provided.
    """
    elapsed = start_timer()

    positional: list[str] = getattr(args, "positional", [])
    json_out_top: bool = getattr(args, "json_out", False)

    # ── resolve subcommand dispatch ───────────────────────────────────────────
    if positional and positional[0] == "resolve":
        run_resolve(args)
        return

    # ── show subcommand dispatch (muse#53 item 9) ─────────────────────────────
    if positional and positional[0] == "show":
        run_show(args)
        return

    # ── diff subcommand dispatch ──────────────────────────────────────────────
    if positional and positional[0] == "diff":
        run_diff(args)
        return

    # ── expire subcommand dispatch ────────────────────────────────────────────
    if positional and positional[0] == "expire":
        run_expire(args)
        return

    # ── delete subcommand dispatch ────────────────────────────────────────────
    if positional and positional[0] == "delete":
        run_delete(args)
        return

    # ── exists subcommand dispatch ────────────────────────────────────────────
    if positional and positional[0] == "exists":
        if len(positional) < 2:
            print(
                "❌ 'exists' requires a symbol address.\n"
                "   Example: muse symlog exists 'src/billing.py::compute_total'",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)
        run_exists(positional[1], json_out_top)
        return

    # ── stat subcommand dispatch (muse#53 item 10) ────────────────────────────
    if positional and positional[0] == "stat":
        run_stat(args)
        return

    # ── prune-orphans subcommand dispatch (muse#53 item 12) ───────────────────
    if positional and positional[0] == "prune-orphans":
        run_prune_orphans(args)
        return

    file_path: str | None = getattr(args, "file_path", None)
    all_symbols: bool = getattr(args, "all_symbols", False)
    symbol: str | None = positional[0] if positional else None
    limit: int = clamp_int(getattr(args, "limit", 20), 1, 100_000, "limit")
    follow: bool = getattr(args, "follow", False)
    include_diff: bool = getattr(args, "include_diff", False)
    json_out: bool = getattr(args, "json_out", False)
    operation_filter: str | None = getattr(args, "operation_filter", None)
    author_filter: str | None = getattr(args, "author_filter", None)
    since_str: str | None = getattr(args, "since", None)
    until_str: str | None = getattr(args, "until", None)

    # Parse dates early so bad values fail before repo access.
    since_dt: datetime.datetime | None = (
        _parse_date(since_str, "--since") if since_str else None
    )
    until_dt: datetime.datetime | None = (
        _parse_date(until_str, "--until") if until_str else None
    )
    if since_dt and until_dt and since_dt > until_dt:
        print("❌ --since must not be after --until.", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    repo_root = require_repo()

    # ── --all mode ─────────────────────────────────────────────────────────────

    if all_symbols:
        addrs = list_symlog_symbols(repo_root)
        if json_out:
            payload = _SymlogAllJson(
                **make_envelope(elapsed),
                symbols=addrs,
                count=len(addrs),
            )
            print(json.dumps(payload))
        else:
            if not addrs:
                print("No symlog entries found in this repository.")
                return
            print(f"Symbols with symlog entries ({len(addrs)}):")
            for addr in addrs:
                print(f"  {sanitize_display(addr)}")
        return

    # ── --file mode ────────────────────────────────────────────────────────────

    if file_path is not None:
        addrs = list_symlog_symbols_for_file(repo_root, file_path)
        results: list[_SymlogResultJson] = []
        for addr in addrs:
            entries_raw, fsmap = _read_with_follow(repo_root, addr, follow=follow, limit=limit * 10)
            filtered = _apply_filters(entries_raw, operation_filter, author_filter, since_dt, until_dt)
            displayed = filtered[:limit]
            json_entries = _entries_to_json(displayed, fsmap, include_diff, repo_root, symbol_addr=addr)
            results.append(_SymlogResultJson(
                **make_envelope(elapsed),
                symbol=addr,
                total=len(filtered),
                limit=limit,
                followed=follow,
                entries=json_entries,
            ))
        if json_out:
            payload = _SymlogFileJson(
                **make_envelope(elapsed),
                file=file_path,
                symbols=results,
                count=len(results),
            )
            print(json.dumps(payload))
        else:
            if not results:
                print(f"No symlog entries found for file: {sanitize_display(file_path)}")
                return
            for res in results:
                safe_sym = sanitize_display(res["symbol"])
                print(f"\n── {safe_sym}  ({res['total']} entries) ──")
                for entry_json in res["entries"]:
                    e = SymlogEntry(
                        old_content_id=entry_json["old_content_id"],
                        new_content_id=entry_json["new_content_id"],
                        commit_id=entry_json["commit_id"],
                        author=entry_json["author"],
                        timestamp=datetime.datetime.fromisoformat(entry_json["timestamp"]),
                        operation=entry_json["operation"],
                    )
                    print(_fmt_entry_text(entry_json["index"], e, entry_json.get("from_symbol")))
        return

    # ── Single-symbol mode ────────────────────────────────────────────────────

    if symbol is None:
        print(
            "❌ Specify a symbol address, --file FILE, or --all.\n"
            "   Example: muse symlog 'src/billing.py::compute_total'",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    _validate_symbol_addr(symbol)

    # Read a generous cap so post-filter limit can be satisfied.
    raw_limit = min(limit * 50, 100_000)
    entries_raw, fsmap = _read_with_follow(repo_root, symbol, follow=follow, limit=raw_limit)

    filtered = _apply_filters(entries_raw, operation_filter, author_filter, since_dt, until_dt)
    displayed = filtered[:limit]

    if json_out:
        json_entries = _entries_to_json(displayed, fsmap, include_diff, repo_root, symbol_addr=symbol)
        payload_result = _SymlogResultJson(
            **make_envelope(elapsed),
            symbol=symbol,
            total=len(filtered),
            limit=limit,
            followed=follow,
            entries=json_entries,
        )
        print(json.dumps(payload_result))
        return

    safe_sym = sanitize_display(symbol)
    if not displayed:
        if filtered:
            print(f"No symlog entries for {safe_sym} match the active filters.")
        else:
            print(f"No symlog entries for {safe_sym}.")
        return

    print(f"Symlog for {safe_sym}  (newest first)\n")
    for idx, entry in enumerate(displayed):
        from_sym = fsmap.get(idx)
        print(_fmt_entry_text(idx, entry, from_sym))
        if include_diff:
            diff_text = _compute_diff(
                entry.old_content_id, entry.new_content_id, repo_root,
                symbol_addr=symbol, entry=entry,
            )
            if diff_text is not None:
                for line in diff_text.splitlines():
                    print(f"  {line}")

    if len(filtered) > limit:
        remaining = len(filtered) - limit
        print(f"\n  … {remaining} older entr{'y' if remaining == 1 else 'ies'} — increase --limit to see more.")