"""muse log — display commit history.

Output modes
------------

Default (long form)::

    commit a1b2c3d4 (HEAD -> main)
    Author: gabriel
    Date:   2026-03-16 12:00:00 UTC

        Add verse melody

--oneline::

    a1b2c3d4 (HEAD -> main) Add verse melody
    f9e8d7c6 Initial commit

--graph::

    * a1b2c3d4 (HEAD -> main) Add verse melody
    * f9e8d7c6 Initial commit

--stat::

    commit a1b2c3d4 (HEAD -> main)
    Date:   2026-03-16 12:00:00 UTC

        Add verse melody

     + tracks/drums.mid
     1 added, 0 removed

--json (agent-native, always stable)::

    {
      "truncated": false,
      "commits": [
        {
          "commit_id": "a1b2c3d4…",
          "branch": "main",
          "message": "Add verse melody",
          "author": "gabriel",
          "committed_at": "2026-03-16T12:00:00+00:00",
          "parent_commit_id": "f9e8d7c6…",
          "parent2_commit_id": null,
          "snapshot_id": "…",
          "sem_ver_bump": "minor",
          "breaking_changes": [],
          "metadata": {},
          "files_added": [],
          "files_removed": [],
          "files_modified": []
        }
      ]
    }

Combine ``--stat`` with ``--json`` to populate ``files_added``,
``files_removed``, and ``files_modified`` on each commit object.
Without ``--stat`` all three lists are always present but empty.

SemVer bumps are coloured: PATCH  MINOR  MAJOR

Filters: --since, --until, --author, --section, --track, --emotion

Pathspec (git-compatible)::

    muse log -- path/to/file.py          # commits touching a specific file
    muse log -n 5 -- src/               # last 5 commits touching anything in src/
    muse log dev -- README.md            # commits on branch dev that touched README.md

Paths are matched as prefixes: ``src/`` matches ``src/foo.py`` and ``src/bar/baz.py``.
Exact paths match only that exact file.  The ``--`` separator is optional when the
path cannot be mistaken for a branch name, but is always safe to include.
"""

from __future__ import annotations

import argparse
import heapq
import json
import logging
import pathlib
import re
import sys
import textwrap
from collections import deque
from datetime import datetime, timedelta, timezone
from typing import TypedDict

from muse.cli.config import get_limit
from muse.core.errors import ExitCode
from muse.core.repo import read_repo_id, require_repo
from muse.core.store import (
    CommitRecord,
    Metadata,
    get_commit_snapshot_manifest,
    get_commits_for_branch,
    get_head_commit_id,
    read_commit,
    read_current_branch,
)
from muse.core.validation import clamp_int, sanitize_display

type CommitIndex = dict[str, CommitRecord]
type BranchTips = dict[str, list[str]]
type CounterMap = dict[str, int]
type _DeltaMap = dict[str, timedelta]

_HEX_CHARS = frozenset("0123456789abcdefABCDEF")


def _is_known_ref(root: pathlib.Path, ref: str) -> bool:
    """Return True if *ref* resolves to a local branch, remote tracking branch, or commit ID.

    Used to disambiguate ``muse log <ref>`` (branch/commit) from
    ``muse log -- <path>`` (pathspec that argparse mistakenly assigned to ref).
    """
    # Short or full commit ID (hex string, min 8 chars).
    if len(ref) >= 8 and all(c in _HEX_CHARS for c in ref):
        return True
    # Local branch: .muse/refs/heads/<ref>
    if (root / ".muse" / "refs" / "heads" / ref).exists():
        return True
    # Remote-tracking branch: .muse/remotes/<remote>/<branch>
    # ref may be "origin/dev" → remote="origin", branch="dev"
    if "/" in ref:
        remote, _, branch = ref.partition("/")
        if branch and (root / ".muse" / "remotes" / remote / branch).exists():
            return True
    return False

logger = logging.getLogger(__name__)

_DEFAULT_LIMIT = 1000

# ANSI colour helpers — only emitted when stdout is a TTY.
_RESET  = "\033[0m"
_BOLD   = "\033[1m"
_YELLOW = "\033[33m"
_GREEN  = "\033[32m"
_RED    = "\033[31m"
_CYAN   = "\033[36m"
_DIM    = "\033[2m"

class _CommitJson(TypedDict):
    """Stable JSON wire format for a single commit in ``muse log --json``.

    ``files_added``, ``files_removed``, and ``files_modified`` are always
    present.  They are populated only when ``--stat`` is also passed;
    otherwise they are empty lists.
    """

    commit_id: str
    branch: str
    message: str
    author: str
    committed_at: str
    parent_commit_id: str | None
    parent2_commit_id: str | None
    snapshot_id: str | None
    sem_ver_bump: str | None
    breaking_changes: list[str]
    metadata: Metadata
    files_added: list[str]
    files_removed: list[str]
    files_modified: list[str]


_SEMVER_COLOUR: Metadata = {
    "major": _RED,
    "minor": _YELLOW,
    "patch": _GREEN,
}


def _c(text: str, *codes: str, tty: bool) -> str:
    """Wrap *text* in ANSI *codes* when *tty* is True."""
    if not tty:
        return text
    return "".join(codes) + text + _RESET


def _ref_label(branch: str, is_head: bool, tty: bool) -> str:
    """Format the ``(HEAD -> branch)`` decoration for a commit line."""
    if not is_head:
        return ""
    if not tty:
        return f" (HEAD -> {branch})"
    head = _c("HEAD", _BOLD, _CYAN, tty=tty)
    arrow = _c(" -> ", _RESET, tty=tty)
    br = _c(branch, _BOLD, _GREEN, tty=tty)
    paren_open  = _c("(", _YELLOW, tty=tty)
    paren_close = _c(")", _YELLOW, tty=tty)
    return f" {paren_open}{head}{arrow}{br}{paren_close}"


def _parse_date(text: str) -> datetime:
    """Parse a human-readable date string into an aware UTC datetime.

    Accepted formats:
    - ``"today"`` / ``"yesterday"``
    - ``"<n> days|weeks|months|years ago"``  (months = 30 days, years = 365 days)
    - ``"YYYY-MM-DD"``
    - ``"YYYY-MM-DDTHH:MM:SS"`` or ``"YYYY-MM-DD HH:MM:SS"``

    Raises:
        ValueError: When the input does not match any recognised format.
    """
    text = text.strip().lower()
    now = datetime.now(timezone.utc)
    if text == "today":
        return now.replace(hour=0, minute=0, second=0, microsecond=0)
    if text == "yesterday":
        return (now - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
    m = re.match(r"^(\d+)\s+(day|week|month|year)s?\s+ago$", text)
    if m:
        n = int(m.group(1))
        unit = m.group(2)
        deltas: _DeltaMap = {
            "day": timedelta(days=n),
            "week": timedelta(weeks=n),
            "month": timedelta(days=n * 30),
            "year": timedelta(days=n * 365),
        }
        return now - deltas[unit]
    for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S"):
        try:
            return datetime.strptime(text, fmt).replace(tzinfo=timezone.utc)
        except ValueError:
            continue
    raise ValueError(f"Cannot parse date: {text!r}")


def _commit_touches_path(root: pathlib.Path, commit: CommitRecord, path: str) -> bool:
    """Return True if *commit* changed *path* (added, modified, or removed).

    *path* is matched as a prefix so that ``src/`` matches any file under
    ``src/``, while ``src/foo.py`` matches only that exact file.  Trailing
    slashes on directory prefixes are normalised away before comparison.
    """
    norm = path.rstrip("/")
    current_manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {}
    if commit.parent_commit_id:
        parent_manifest = get_commit_snapshot_manifest(root, commit.parent_commit_id) or {}
    else:
        parent_manifest = {}
    all_paths = set(current_manifest) | set(parent_manifest)
    for p in all_paths:
        if p == norm or p.startswith(norm + "/"):
            if current_manifest.get(p) != parent_manifest.get(p):
                return True
    return False


def _file_diff(
    root: pathlib.Path,
    commit: CommitRecord,
) -> tuple[list[str], list[str], list[str]]:
    """Return ``(added, removed, modified)`` file lists relative to the commit's parent.

    Compares the commit's snapshot manifest against its first parent's manifest.
    For merge commits the second parent is ignored — this is a ``--stat``-style
    summary, not a three-way diff.

    Returns:
        A 3-tuple of sorted file-path lists:
        ``added``    — files present in the commit but not in its parent.
        ``removed``  — files present in the parent but not in the commit.
        ``modified`` — files present in both with a different snapshot hash.
    """
    current_manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {}
    if commit.parent_commit_id:
        parent_manifest = get_commit_snapshot_manifest(root, commit.parent_commit_id) or {}
    else:
        parent_manifest = {}
    added = sorted(set(current_manifest) - set(parent_manifest))
    removed = sorted(set(parent_manifest) - set(current_manifest))
    modified = sorted(
        f for f in set(current_manifest) & set(parent_manifest)
        if current_manifest[f] != parent_manifest[f]
    )
    return added, removed, modified


def _format_date(dt: datetime) -> str:
    """Format a datetime as ``YYYY-MM-DD HH:MM:SS UTC`` for display."""
    return dt.strftime("%Y-%m-%d %H:%M:%S UTC") if dt.tzinfo else str(dt)


# ---------------------------------------------------------------------------
# Filter helpers — single implementation shared by JSON and text paths
# ---------------------------------------------------------------------------


def _apply_filters(
    commits: list[CommitRecord],
    *,
    since_dt: datetime | None,
    until_dt: datetime | None,
    author: str | None,
    section: str | None,
    track: str | None,
    emotion: str | None,
    limit: int,
) -> tuple[list[CommitRecord], bool]:
    """Apply all active filters to *commits* and enforce *limit*.

    Previously the filter loop was copy-pasted between the JSON path and the
    text path.  Any new filter would have required updating both.  This
    function is the single source of truth; both paths call it.

    Args:
        commits:   Pre-fetched commit list from ``get_commits_for_branch``.
        since_dt:  Inclusive lower bound on ``committed_at``.
        until_dt:  Inclusive upper bound on ``committed_at``.
        author:    Substring match on ``author`` (case-insensitive).
        section:   Exact match on ``metadata["section"]``.
        track:     Exact match on ``metadata["track"]``.
        emotion:   Exact match on ``metadata["emotion"]``.
        limit:     Maximum number of commits to return (0 = unlimited).

    Returns:
        ``(filtered_commits, truncated)`` where *truncated* is True when the
        walk was capped before the filter had seen all available commits.
    """
    filtered: list[CommitRecord] = []
    for c in commits:
        if since_dt and c.committed_at < since_dt:
            continue
        if until_dt and c.committed_at > until_dt:
            continue
        if author and author.lower() not in c.author.lower():
            continue
        if section and c.metadata.get("section") != section:
            continue
        if track and c.metadata.get("track") != track:
            continue
        if emotion and c.metadata.get("emotion") != emotion:
            continue
        filtered.append(c)
        if limit > 0 and len(filtered) >= limit:
            # We hit the display limit — the walk may have had more commits.
            truncated = len(commits) > len(filtered)
            return filtered, truncated
    return filtered, False


def _commit_to_json(
    c: CommitRecord,
    root: pathlib.Path | None = None,
    *,
    stat: bool = False,
) -> _CommitJson:
    """Serialise *c* to the stable ``--json`` wire format.

    All keys are always present so agents can read them without ``dict.get``
    guards.  Merge commits include ``parent2_commit_id``; linear commits have
    it set to ``null``.

    When *stat* is True and *root* is provided, ``files_added``,
    ``files_removed``, and ``files_modified`` are populated by diffing the
    commit's snapshot against its parent.  Otherwise all three are empty lists.

    The ``--json`` output path always calls this with ``stat=True`` so that
    agents always receive populated file lists without needing to pass
    ``--stat`` explicitly.  The *stat* parameter is kept for callers that
    intentionally want the lightweight (no-diff) form.

    Args:
        c:    The commit to serialise.
        root: Repository root directory.  Required when *stat* is True.
        stat: Whether to compute and include per-file change lists.

    Returns:
        JSON-serialisable dict matching the schema documented in the module
        docstring.
    """
    files_added: list[str] = []
    files_removed: list[str] = []
    files_modified: list[str] = []
    if stat and root is not None:
        files_added, files_removed, files_modified = _file_diff(root, c)
    return _CommitJson(
        commit_id=c.commit_id,
        branch=c.branch,
        message=c.message,
        author=c.author,
        committed_at=c.committed_at.isoformat(),
        parent_commit_id=c.parent_commit_id,
        parent2_commit_id=c.parent2_commit_id,
        snapshot_id=c.snapshot_id,
        sem_ver_bump=c.sem_ver_bump,
        breaking_changes=list(c.breaking_changes) if c.breaking_changes else [],
        metadata=c.metadata,
        files_added=files_added,
        files_removed=files_removed,
        files_modified=files_modified,
    )


# ---------------------------------------------------------------------------
# DAG graph rendering helpers
# ---------------------------------------------------------------------------


def _branch_tips(root: pathlib.Path) -> BranchTips:
    """Return ``{commit_id: [branch_name, …]}`` for all local branch tips."""
    heads_dir = root / ".muse" / "refs" / "heads"
    if not heads_dir.exists():
        return {}
    tips: BranchTips = {}
    for p in heads_dir.rglob("*"):
        if p.is_file():
            cid = p.read_text().strip()
            name = p.relative_to(heads_dir).as_posix()
            if cid:
                tips.setdefault(cid, []).append(name)
    return tips


def _collect_all_commits(
    root: pathlib.Path,
    start_ids: list[str],
    max_commits: int = 50_000,
) -> tuple[dict[str, CommitRecord], bool]:
    """BFS from *start_ids*, returning every reachable commit up to *max_commits*.

    Uses a :class:`~collections.deque` for O(1) ``popleft`` — a ``list.pop(0)``
    would be O(n) per call, making the full BFS O(n²) at scale.

    Args:
        root:        Repository root.
        start_ids:   BFS seed commit IDs (branch tips).
        max_commits: Safety cap — default 50 000.

    Returns:
        ``({commit_id: CommitRecord}, truncated)`` pair where *truncated* is
        True when *max_commits* was reached before the graph was fully walked.
    """
    seen: CommitIndex = {}
    queue: deque[str] = deque(start_ids)
    while queue:
        if len(seen) >= max_commits:
            return seen, True
        cid = queue.popleft()
        if cid in seen:
            continue
        rec = read_commit(root, cid)
        if rec is None:
            continue
        seen[cid] = rec
        for parent in (rec.parent_commit_id, rec.parent2_commit_id):
            if parent and parent not in seen:
                queue.append(parent)
    return seen, False


def _topo_sort(commits: CommitIndex) -> list[CommitRecord]:
    """Return commits newest-first using Kahn's algorithm.

    In-degree counts the number of *child* commits that reference each commit
    as a parent, so commits with no children (branch tips) are processed first.
    Ties are broken by timestamp (most recent first).
    """
    in_degree: CounterMap = {cid: 0 for cid in commits}
    for rec in commits.values():
        for parent in (rec.parent_commit_id, rec.parent2_commit_id):
            if parent and parent in commits:
                in_degree[parent] += 1

    heap: list[tuple[float, str]] = []
    for cid, deg in in_degree.items():
        if deg == 0:
            ts = -commits[cid].committed_at.timestamp()
            heapq.heappush(heap, (ts, cid))

    result: list[CommitRecord] = []
    while heap:
        _, cid = heapq.heappop(heap)
        result.append(commits[cid])
        rec = commits[cid]
        for parent in (rec.parent_commit_id, rec.parent2_commit_id):
            if parent and parent in commits:
                in_degree[parent] -= 1
                if in_degree[parent] == 0:
                    ts = -commits[parent].committed_at.timestamp()
                    heapq.heappush(heap, (ts, parent))
    return result


def _deco_str(
    cid: str,
    head_cid: str,
    current: str,
    tips: BranchTips,
    tty: bool,
) -> str:
    """Format the ``(HEAD -> branch, other-branch)`` decoration for a commit."""
    branches = tips.get(cid, [])
    if not branches:
        return ""
    labels: list[str] = []
    if cid == head_cid and current in branches:
        head = _c("HEAD", _BOLD, _CYAN, tty=tty)
        br = _c(current, _BOLD, _GREEN, tty=tty)
        labels.append(f"{head}{_c(' -> ', _RESET, tty=tty)}{br}")
        for b in branches:
            if b != current:
                labels.append(_c(b, _BOLD, _GREEN, tty=tty))
    else:
        for b in branches:
            labels.append(_c(b, _BOLD, _GREEN, tty=tty))
    inner = ", ".join(labels)
    return f" {_c('(', _YELLOW, tty=tty)}{inner}{_c(')', _YELLOW, tty=tty)}"


def _render_graph(
    root: pathlib.Path,
    branch: str,
    all_branches: bool,
    tty: bool,
) -> None:
    """Render a lane-based ASCII DAG, git-log-style."""
    current = read_current_branch(root)
    tips = _branch_tips(root)

    if all_branches:
        start_ids = list(tips.keys())
    else:
        head = get_head_commit_id(root, branch)
        start_ids = [head] if head else []

    if not start_ids:
        print("(no commits)")
        return

    graph_cap = get_limit("max_graph_commits", root)
    all_commits, graph_truncated = _collect_all_commits(root, start_ids, max_commits=graph_cap)
    if not all_commits:
        print("(no commits)")
        return
    if graph_truncated:
        print(
            f"⚠️  Graph truncated at {graph_cap:,} commits. "
            "Raise [limits] max_graph_commits in .muse/config.toml to see more."
        )

    head_cid = get_head_commit_id(root, current) or ""
    sorted_commits = _topo_sort(all_commits)

    # lanes: list of commit IDs we're "awaiting" (open lines of descent).
    # None marks a closed / empty column slot.
    lanes: list[str | None] = []

    for idx, commit in enumerate(sorted_commits):
        cid = commit.commit_id
        parents = [
            p for p in (commit.parent_commit_id, commit.parent2_commit_id)
            if p and p in all_commits
        ]

        # Assign this commit to a column.
        col = lanes.index(cid) if cid in lanes else -1
        if col == -1:
            if None in lanes:
                col = lanes.index(None)
                lanes[col] = cid
            else:
                col = len(lanes)
                lanes.append(cid)

        width = len(lanes)
        row: list[str] = []
        for i in range(width):
            if i == col:
                row.append(_c("*", _BOLD, tty=tty))
            elif lanes[i] is not None:
                row.append("|")
            else:
                row.append(" ")

        graph_prefix = " ".join(row).rstrip()
        short_hash = _c(cid[:8], _YELLOW, tty=tty)
        deco = _deco_str(cid, head_cid, current, tips, tty)
        msg = sanitize_display(commit.message.splitlines()[0])
        print(f"{graph_prefix} {short_hash}{deco} {msg}")

        if parents:
            lanes[col] = parents[0]
            for extra in parents[1:]:
                if extra not in lanes:
                    if None in lanes:
                        lanes[lanes.index(None)] = extra
                    else:
                        lanes.append(extra)
        else:
            lanes[col] = None

        if idx < len(sorted_commits) - 1:
            is_merge = len(parents) >= 2
            connector: list[str] = []
            for i in range(len(lanes)):
                if i == col and is_merge:
                    connector.append("|\\")
                elif lanes[i] is not None:
                    connector.append("| ")
                else:
                    connector.append("  ")
            line = "".join(connector).rstrip()
            if line:
                print(line)

        while lanes and lanes[-1] is None:
            lanes.pop()


# ---------------------------------------------------------------------------
# CLI registration
# ---------------------------------------------------------------------------


def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
    """Register the ``muse log`` subcommand and its flags."""
    parser = subparsers.add_parser(
        "log",
        help="Display commit history.",
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "ref", nargs="?", default=None,
        help="Branch or commit to start from (default: current branch).",
    )
    parser.add_argument(
        "--oneline", action="store_true",
        help="One line per commit.",
    )
    parser.add_argument(
        "--graph", "-g", action="store_true",
        help="ASCII DAG graph.",
    )
    parser.add_argument(
        "--all", "-A", action="store_true", dest="all_branches",
        help="Include all local branches in the graph (implies --graph).",
    )
    parser.add_argument(
        "--stat", action="store_true",
        help="Show added/removed file summary for each commit.",
    )
    parser.add_argument(
        "-n", "--max-count", type=int, default=_DEFAULT_LIMIT, dest="limit",
        help="Limit number of commits shown (default: %(default)s).",
    )
    parser.add_argument("--since", default=None, help="Show commits after date.")
    parser.add_argument("--until", default=None, help="Show commits before date.")
    parser.add_argument("--author", default=None, help="Filter by author substring.")
    parser.add_argument("--section", default=None, help="Filter by section metadata.")
    parser.add_argument("--track", default=None, help="Filter by track metadata.")
    parser.add_argument("--emotion", default=None, help="Filter by emotion metadata.")
    parser.add_argument(
        "--format", "-f", default="text", dest="fmt",
        help="Output format: text (default) or json.",
    )
    parser.add_argument(
        "--json", action="store_const", const="json", dest="fmt",
        help="Shorthand for --format json.",
    )
    parser.add_argument(
        "pathspec", nargs="*", metavar="path",
        help=(
            "Limit output to commits that touched these paths. "
            "Prefix-matched: 'src/' matches all files under src/. "
            "Separate from options with '--' (e.g. muse log -- src/foo.py)."
        ),
    )
    parser.set_defaults(func=run)


def run(args: argparse.Namespace) -> None:
    """Display commit history.

    Agents should pass ``--json`` to receive a stable JSON envelope::

        {
          "truncated": false,
          "commits": [ { … }, … ]
        }

    Each commit object always contains: ``commit_id``, ``branch``,
    ``message``, ``author``, ``committed_at``, ``parent_commit_id``,
    ``parent2_commit_id`` (null for linear commits), ``snapshot_id``,
    ``sem_ver_bump``, ``breaking_changes`` (list), ``metadata`` (dict), and
    the per-file change lists ``files_added``, ``files_removed``, and
    ``files_modified``.  File lists are always populated in ``--json`` mode —
    agents do not need to pass ``--stat`` to receive them.

    The ``truncated`` flag is true when the walk was capped before the full
    history was read.  Raise ``[limits] max_walk_commits`` in
    ``.muse/config.toml`` or pass ``-n <N>`` to control the cap.

    Exit codes:
        0 — success.
        2 — usage error (invalid ``--format``, invalid ``--since``/``--until``
            value, ``--max-count`` less than 1).
        3 — internal error (repository not found).
    """
    ref: str | None = args.ref
    oneline: bool = args.oneline
    graph: bool = args.graph
    all_branches: bool = args.all_branches
    stat: bool = args.stat
    limit: int = clamp_int(args.limit, 1, 100000, "limit")
    since: str | None = args.since
    until: str | None = args.until
    author: str | None = args.author
    section: str | None = args.section
    track: str | None = args.track
    emotion: str | None = args.emotion
    fmt: str = args.fmt
    pathspec: list[str] = list(args.pathspec) if args.pathspec else []

    if all_branches:
        graph = True

    # Support git-style -<n> shorthand (e.g. `muse log -5` as alias for `-n 5`).
    # argparse captures "-5" as the positional `ref`; detect and reinterpret it.
    if ref is not None and ref.lstrip("-").isdigit() and ref.startswith("-"):
        limit = int(ref.lstrip("-"))
        ref = None

    if fmt not in ("text", "json"):
        print(
            f"❌ Unknown --format '{sanitize_display(fmt)}'. Choose text or json.",
            file=sys.stderr,
        )
        raise SystemExit(ExitCode.USER_ERROR)

    if limit < 1:
        print("❌ --max-count must be at least 1.", file=sys.stderr)
        raise SystemExit(ExitCode.USER_ERROR)

    # Validate date filters before touching the repository.
    since_dt: datetime | None = None
    until_dt: datetime | None = None
    if since:
        try:
            since_dt = _parse_date(since)
        except ValueError:
            print(
                f"❌ Cannot parse --since value: {sanitize_display(since)!r}. "
                "Use YYYY-MM-DD, 'today', 'yesterday', or '<n> days ago'.",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)
    if until:
        try:
            until_dt = _parse_date(until)
        except ValueError:
            print(
                f"❌ Cannot parse --until value: {sanitize_display(until)!r}. "
                "Use YYYY-MM-DD, 'today', 'yesterday', or '<n> days ago'.",
                file=sys.stderr,
            )
            raise SystemExit(ExitCode.USER_ERROR)

    root = require_repo()
    repo_id = read_repo_id(root)

    # When the -N shorthand consumed `ref` (e.g. `muse log -1 dev`), argparse
    # puts the branch name in pathspec[0].  Reclaim it now that we have a repo
    # root available to check whether it's a real ref.
    if ref is None and pathspec and _is_known_ref(root, pathspec[0]):
        ref = pathspec[0]
        pathspec = pathspec[1:]

    # Disambiguate: argparse assigns the first positional after `--` to `ref`
    # (because ref is nargs="?").  If what landed in `ref` is not a known
    # branch or commit ID, treat it as the first pathspec element.
    # Preserve the original value for the "no commits on '<ref>'" message so
    # that nonexistent-branch errors are still contextual.
    user_ref = ref  # may be a nonexistent branch name — kept for error display
    if ref is not None and not _is_known_ref(root, ref):
        pathspec = [ref] + pathspec
        ref = None

    branch = ref if ref is not None else read_current_branch(root)

    if graph and fmt == "text":
        _render_graph(root, branch=branch, all_branches=all_branches, tty=sys.stdout.isatty())
        return

    has_filters = any([since_dt, until_dt, author, section, track, emotion])
    walk_cap = get_limit("max_walk_commits", root)
    # When pathspec is given we must walk all commits and filter; the limit is
    # applied afterwards, just as git log does.
    walk_limit = 0 if (has_filters or pathspec) else limit
    raw_commits = get_commits_for_branch(root, repo_id, branch, max_count=walk_limit or walk_cap)

    # Apply pathspec filter: keep only commits that touched one of the given paths.
    if pathspec:
        raw_commits = [
            c for c in raw_commits
            if any(_commit_touches_path(root, c, p) for p in pathspec)
        ]

    walk_truncated = (has_filters or bool(pathspec)) and len(raw_commits) >= walk_cap

    filtered, filter_truncated = _apply_filters(
        raw_commits,
        since_dt=since_dt,
        until_dt=until_dt,
        author=author,
        section=section,
        track=track,
        emotion=emotion,
        limit=limit,
    )
    truncated = walk_truncated or filter_truncated

    # ── JSON output ───────────────────────────────────────────────────────────
    if fmt == "json":
        out = sys.stdout
        out.write('{"truncated":')
        out.write(json.dumps(truncated))
        out.write(',"commits":[\n')
        for i, c in enumerate(filtered):
            if i:
                out.write(",\n")
            out.write(json.dumps(_commit_to_json(c, root, stat=True), default=str))
        out.write("\n]}\n")
        return

    # ── Text output ───────────────────────────────────────────────────────────
    if not filtered:
        if user_ref is not None:
            # Distinguish "branch exists but is empty" from "branch not found".
            print(f"(no commits on '{sanitize_display(user_ref)}')")
        else:
            print("(no commits)")
        return

    if truncated:
        print(
            f"⚠️  History truncated at {len(filtered):,} commits — "
            "use -n or raise [limits] max_walk_commits in .muse/config.toml to see more."
        )

    head_commit_id = filtered[0].commit_id if filtered else None
    tty: bool = sys.stdout.isatty()

    for c in filtered:
        is_head = c.commit_id == head_commit_id
        decoration = _ref_label(branch, is_head, tty)
        short_hash = _c(c.commit_id[:8], _YELLOW, tty=tty)
        subject = sanitize_display(c.message.splitlines()[0])
        author_display = sanitize_display(c.author)

        if oneline:
            print(f"{short_hash}{decoration} {subject}")
            continue

        commit_word = _c("commit", _YELLOW, tty=tty)
        print(f"{commit_word} {short_hash}{decoration}")
        if author_display:
            print(f"Author: {author_display}")
        print(f"Date:   {_c(_format_date(c.committed_at), _DIM, tty=tty)}")

        if c.sem_ver_bump and c.sem_ver_bump != "none":
            bump_key = c.sem_ver_bump.lower()
            bump_colour = _SEMVER_COLOUR.get(bump_key, "")
            bump_label = (
                _c(c.sem_ver_bump.upper(), bump_colour, tty=tty)
                if bump_colour else c.sem_ver_bump.upper()
            )
            print(f"SemVer: {bump_label}")
            if c.breaking_changes:
                safe_breaks = [sanitize_display(b) for b in c.breaking_changes[:3]]
                breaking_text = ", ".join(safe_breaks)
                if len(c.breaking_changes) > 3:
                    breaking_text += f" +{len(c.breaking_changes) - 3} more"
                print(f"Breaking: {_c(breaking_text, _RED, tty=tty)}")

        if c.metadata:
            meta_parts = [
                f"{sanitize_display(k)}: {sanitize_display(str(v))}"
                for k, v in sorted(c.metadata.items())
            ]
            print(f"Meta:   {', '.join(meta_parts)}")

        # Indent every line of the message body uniformly.
        body = sanitize_display(c.message)
        indented = textwrap.indent(body, "    ")
        print(f"\n{indented}\n")

        if stat:
            added, removed, modified = _file_diff(root, c)
            for p in added:
                print(_c(f" + {p}", _GREEN, tty=tty))
            for p in modified:
                print(_c(f" ~ {p}", _YELLOW, tty=tty))
            for p in removed:
                print(_c(f" - {p}", _RED, tty=tty))
            if added or removed or modified:
                summary = (
                    f" {len(added)} added, {len(modified)} modified, "
                    f"{len(removed)} removed"
                )
                print(_c(summary, _DIM, tty=tty) + "\n")