"""Symlog — live per-symbol journal written at commit time.

Every commit that changes a symbol's content ID appends an entry to that
symbol's journal.  Journals are stored in a directory tree that mirrors the
source tree::

    .muse/symlogs/
        src/
            billing.py/
                compute_total          ← one file per symbol
                validate_invoice
        tests/
            test_billing.py/
                test_compute_total

This layout gives O(1) per-symbol reads (no commit-history scan), O(1) per-file
symbol enumeration (one ``os.listdir``), and independent per-symbol GC.

Entry format
------------
Each journal file is an append-only plain-text sequence of lines::

    <old_content_id> <new_content_id> <commit_id> <author> <ts_unix> <tz_offset> TAB <operation>

Fields:

old_content_id  sha256:<64-hex> — symbol content ID before the change, or
                NULL_CONTENT_ID for created symbols.
new_content_id  sha256:<64-hex> — symbol content ID after the change, or
                NULL_CONTENT_ID for deleted/renamed-away symbols.
commit_id       sha256:<64-hex> — commit that caused this entry.
author          sanitized author or agent_id string.
ts_unix         Unix seconds as a decimal integer (UTC).
tz_offset       UTC offset in +HHMM / -HHMM form (always +0000 today).
operation       Free-form description of the lifecycle event.

Lifecycle sentinels
-------------------
symbol-created: <name>
    old_content_id == NULL_CONTENT_ID.

symbol-modified: <commit message first line>
    Both content IDs are populated.

symbol-deleted: <name>
    new_content_id == NULL_CONTENT_ID.

symbol-renamed-to: <new_addr>
    Terminal entry written to the OLD symbol path.  new_content_id == NULL_CONTENT_ID.

symbol-born-from: <old_addr>
    Born-from entry written to the NEW symbol path.  old_content_id == NULL_CONTENT_ID.
    The ``born_from`` field of the parsed entry is set to <old_addr>.

Path encoding
-------------
The symbol name (part after ``::``) is percent-encoded for any character
outside ``[a-zA-Z0-9._-]``.  The file path portion is used as-is (it must
already be a valid relative path).  ``..`` components in either portion raise
``ValueError`` before any filesystem access.

Security model
--------------
``author`` and ``operation`` are sanitized before being written to disk:
newlines are stripped to prevent line injection; tabs are stripped from
``author`` to prevent corruption of the tab-delimited parse boundary.

``list_symlog_symbols`` and ``list_symlog_symbols_for_file`` skip symlinks —
a crafted symlink inside ``.muse/symlogs/`` cannot be followed to enumerate
arbitrary filesystem paths.

``read_symlog`` enforces a ``_MAX_SYMLOG_BYTES`` file-size cap before reading
to prevent OOM from excessively large or maliciously crafted journal files.
"""

from __future__ import annotations

import datetime
import hashlib
import json
import logging
import os
import pathlib
import re
import tempfile
import urllib.parse
from dataclasses import dataclass, field as _field

from muse.core.paths import symlogs_dir as _symlogs_dir
from muse.core.types import DEFAULT_HASH_ALGO

logger = logging.getLogger(__name__)

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

#: Null content ID — used as old_content_id for created symbols and
#: new_content_id for deleted / renamed-away symbols.
NULL_CONTENT_ID: str = f"{DEFAULT_HASH_ALGO}:" + "0" * 64

#: Maximum symlog file size that will be read.  Files larger than this cap
#: are warned about rather than loaded entirely into memory.
_MAX_SYMLOG_BYTES: int = 10 * 1024 * 1024  # 10 MiB

#: Characters in the symbol name that do NOT need percent-encoding.
_SAFE_CHARS: str = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-"

#: Leaf filenames at or under this many UTF-8 bytes are always safe on every
#: filesystem this tool supports (POSIX's 255-byte NAME_MAX, with headroom
#: for the ".lock" suffix atomic writes append). Percent-encoded names longer
#: than this fall back to a hashed leaf name instead of raising OSError.
_MAX_LEAF_BYTES: int = 200

#: Marker prefix for hashed fallback leaf names. Percent-encoding (see
#: ``_encode_symbol_name``) only ever emits characters from ``_SAFE_CHARS``
#: or a literal ``%``, so ``!`` can never appear in a normally-encoded leaf —
#: making it an unambiguous signal that the leaf name must be resolved via
#: the sidecar index rather than percent-decoded directly.
_HASHED_LEAF_PREFIX: str = "!"

#: Sidecar index file, one per symlog directory, mapping hashed leaf names
#: back to their original (pre-hash) symbol name. Only written/consulted for
#: leaves that were too long to encode directly; normal leaves never touch it.
_LEAF_INDEX_FILENAME: str = ".leaf_index.json"

#: Regex that identifies a ``symbol-born-from:`` operation and captures the prior address.
_BORN_FROM_RE = re.compile(r"^symbol-born-from:\s*(.+)$")


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

def is_null_content_id(content_id: str) -> bool:
    """Return ``True`` iff *content_id* is the canonical null content ID."""
    return content_id == NULL_CONTENT_ID


def _encode_symbol_name(name: str) -> str:
    """Percent-encode characters outside ``[a-zA-Z0-9._-]`` in *name*."""
    return urllib.parse.quote(name, safe=_SAFE_CHARS)


def _decode_symbol_name(encoded: str) -> str:
    """Reverse percent-encoding applied by ``_encode_symbol_name``."""
    return urllib.parse.unquote(encoded)


def _hashed_leaf_name(symbol_name: str) -> str:
    """Return a short, collision-resistant leaf name for a too-long symbol name."""
    digest = hashlib.sha256(symbol_name.encode("utf-8")).hexdigest()
    return f"{_HASHED_LEAF_PREFIX}{digest[:48]}"


def _leaf_index_path(leaf: pathlib.Path) -> pathlib.Path:
    """Return the sidecar index path for the symlog directory containing *leaf*."""
    return leaf.parent / _LEAF_INDEX_FILENAME


def _record_hashed_leaf(leaf: pathlib.Path, symbol_name: str) -> None:
    """Record *symbol_name* under *leaf*'s hashed name in the sidecar index.

    Idempotent and atomic (write-to-temp + ``os.replace``) — safe to call on
    every append, not just the first.
    """
    index_path = _leaf_index_path(leaf)
    try:
        data = (
            json.loads(index_path.read_text(encoding="utf-8"))
            if index_path.exists()
            else {}
        )
    except (OSError, ValueError):
        data = {}
    if data.get(leaf.name) == symbol_name:
        return
    data[leaf.name] = symbol_name
    tmp_path = index_path.with_name(index_path.name + ".tmp")
    tmp_path.write_text(json.dumps(data, sort_keys=True), encoding="utf-8")
    os.replace(tmp_path, index_path)


def _resolve_hashed_leaf(leaf: pathlib.Path) -> str | None:
    """Look up the original symbol name for a hashed *leaf* via the sidecar index."""
    index_path = _leaf_index_path(leaf)
    if not index_path.exists():
        return None
    try:
        data = json.loads(index_path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    value = data.get(leaf.name)
    return value if isinstance(value, str) else None


def _reject_traversal(part: str, label: str) -> None:
    """Raise ``ValueError`` when *part* contains a ``..`` path component."""
    segments = pathlib.PurePosixPath(part).parts
    if ".." in segments:
        raise ValueError(
            f"Path traversal detected in {label!r}: '..' components are not allowed."
        )


def _sanitize_author(s: str) -> str:
    """Strip characters that would corrupt the symlog line structure."""
    return s.replace("\n", "").replace("\r", "").replace("\t", "")


def _sanitize_operation(s: str) -> str:
    """Strip newlines to prevent line injection in the operation field."""
    return s.replace("\n", "").replace("\r", "")


# ---------------------------------------------------------------------------
# Types
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class SymlogEntry:
    """One line of a symbol journal.

    The ``born_from`` field is non-None only when the operation string starts
    with ``symbol-born-from:``; it is always derived from ``operation`` at
    construction time and is never passed by callers.  This ensures the two
    fields are always consistent.
    """

    old_content_id: str          # sha256:… or NULL_CONTENT_ID
    new_content_id: str          # sha256:… or NULL_CONTENT_ID
    commit_id:      str          # sha256:… of the commit that caused this entry
    author:         str          # sanitized author / agent_id
    timestamp:      datetime.datetime
    operation:      str          # "symbol-created: …", "symbol-modified: …", etc.
    born_from:      str | None = _field(default=None, init=False)

    def __post_init__(self) -> None:
        m = _BORN_FROM_RE.match(self.operation)
        object.__setattr__(self, "born_from", m.group(1).strip() if m else None)


def _make_entry(
    old_content_id: str,
    new_content_id: str,
    commit_id: str,
    author: str,
    timestamp: datetime.datetime,
    operation: str,
) -> SymlogEntry:
    """Construct a ``SymlogEntry`` from parsed line components."""
    return SymlogEntry(
        old_content_id=old_content_id,
        new_content_id=new_content_id,
        commit_id=commit_id,
        author=author,
        timestamp=timestamp,
        operation=operation,
    )


# ---------------------------------------------------------------------------
# Path helpers
# ---------------------------------------------------------------------------

def symlog_path(repo_root: pathlib.Path, symbol_addr: str) -> pathlib.Path:
    """Return the journal file path for *symbol_addr*.

    *symbol_addr* must contain exactly one ``::`` separator, e.g.
    ``src/billing.py::compute_total``.  The file path portion maps to a
    directory under ``.muse/symlogs/``; the symbol name portion becomes the
    leaf filename (percent-encoded for non-ASCII-safe characters).

    Raises:
        ValueError: ``::`` separator is missing, symbol name is empty,
                    or a ``..`` path component is detected.
    """
    if "::" not in symbol_addr:
        raise ValueError(
            f"Missing '::' separator in symbol address {symbol_addr!r}. "
            "Expected form: 'path/to/file.py::symbol_name'."
        )
    file_path, _, symbol_name = symbol_addr.partition("::")
    if not symbol_name:
        raise ValueError(
            f"Empty symbol name in symbol address {symbol_addr!r}."
        )
    # Reject path traversal in both parts.
    _reject_traversal(file_path, "file_path")
    _reject_traversal(symbol_name, "symbol_name")
    encoded_name = _encode_symbol_name(symbol_name)
    if len(encoded_name.encode("utf-8")) > _MAX_LEAF_BYTES:
        encoded_name = _hashed_leaf_name(symbol_name)
    return _symlogs_dir(repo_root) / file_path / encoded_name


def _addr_from_path(symlogs_root: pathlib.Path, leaf: pathlib.Path) -> str:
    """Reconstruct a symbol address from a leaf path under *symlogs_root*.

    Raises:
        ValueError: *leaf* is a hashed fallback name with no sidecar index
                    entry (index missing or corrupt).
    """
    rel = leaf.relative_to(symlogs_root)
    parts = rel.parts
    # All parts up to the last are the file path; the last is the encoded symbol name.
    file_path = str(pathlib.PurePosixPath(*parts[:-1]))
    leaf_name = parts[-1]
    if leaf_name.startswith(_HASHED_LEAF_PREFIX):
        symbol_name = _resolve_hashed_leaf(leaf)
        if symbol_name is None:
            raise ValueError(
                f"No sidecar index entry for hashed symlog leaf {leaf!r}."
            )
    else:
        symbol_name = _decode_symbol_name(leaf_name)
    return f"{file_path}::{symbol_name}"


# ---------------------------------------------------------------------------
# Write
# ---------------------------------------------------------------------------

def append_symlog(
    repo_root: pathlib.Path,
    symbol_addr: str,
    old_content_id: str,
    new_content_id: str,
    commit_id: str,
    author: str,
    operation: str,
) -> None:
    """Append one entry to the symbol journal for *symbol_addr*.

    Parent directories are created automatically on first write.  Author and
    operation are sanitized to prevent line injection.

    Args:
        repo_root:       Root of the Muse repository.
        symbol_addr:     ``"path/to/file.py::symbol_name"``.
        old_content_id:  Content ID before the change (NULL_CONTENT_ID for created).
        new_content_id:  Content ID after the change (NULL_CONTENT_ID for deleted).
        commit_id:       SHA-256 commit ID that caused this entry.
        author:          Agent ID or author string; sanitized before write.
        operation:       Lifecycle description; sanitized before write.
    """
    safe_author = _sanitize_author(author)
    safe_operation = _sanitize_operation(operation)
    now = datetime.datetime.now(tz=datetime.timezone.utc)
    ts = int(now.timestamp())
    tz_offset = "+0000"

    line = (
        f"{old_content_id} {new_content_id} {commit_id} "
        f"{safe_author} {ts} {tz_offset}\t{safe_operation}\n"
    )

    p = symlog_path(repo_root, symbol_addr)
    p.parent.mkdir(parents=True, exist_ok=True)
    if p.name.startswith(_HASHED_LEAF_PREFIX):
        _, _, symbol_name = symbol_addr.partition("::")
        _record_hashed_leaf(p, symbol_name)
    with p.open("a", encoding="utf-8") as fh:
        fh.write(line)


# ---------------------------------------------------------------------------
# Parse
# ---------------------------------------------------------------------------

def _parse_line(line: str) -> SymlogEntry | None:
    """Parse one symlog line; return None on malformed input.

    Format::

        <old_cid> <new_cid> <commit_id> <author…> <ts_unix> <tz_offset> TAB <operation>
    """
    line = line.rstrip("\n")
    parts = line.split("\t", 1)
    if len(parts) != 2:
        return None
    meta, operation = parts
    tokens = meta.split()
    # Minimum: old_cid new_cid commit_id author ts tz  (6 tokens)
    if len(tokens) < 6:
        return None
    old_content_id = tokens[0]
    new_content_id = tokens[1]
    commit_id = tokens[2]
    ts_str = tokens[-2]
    # author may contain spaces — everything between tokens[3] and the last two
    author = " ".join(tokens[3:-2])
    try:
        ts = datetime.datetime.fromtimestamp(int(ts_str), tz=datetime.timezone.utc)
    except (ValueError, OSError):
        ts = datetime.datetime.now(tz=datetime.timezone.utc)
    return _make_entry(
        old_content_id=old_content_id,
        new_content_id=new_content_id,
        commit_id=commit_id,
        author=author,
        timestamp=ts,
        operation=operation,
    )


# ---------------------------------------------------------------------------
# Read
# ---------------------------------------------------------------------------

def read_symlog(
    repo_root: pathlib.Path,
    symbol_addr: str,
    limit: int = 100,
    *,
    follow: bool = False,
) -> list[SymlogEntry]:
    """Return symlog entries for *symbol_addr*, newest-first, up to *limit*.

    When *follow* is ``True`` and the oldest entry in this symbol's log has a
    ``born_from`` pointer (i.e. the operation starts with ``symbol-born-from:``),
    the function recursively reads the prior symbol's log and appends those
    entries so the full rename chain is visible in one list.

    Files larger than ``_MAX_SYMLOG_BYTES`` trigger a warning and are still
    read (the OS read may truncate at a line boundary, but at least a partial
    result is returned rather than an empty list for very large files).
    """
    p = symlog_path(repo_root, symbol_addr)
    if not p.exists():
        return []

    try:
        size = p.stat().st_size
        if size > _MAX_SYMLOG_BYTES:
            logger.warning(
                "⚠️ Symlog %s is %.1f MiB — exceeds cap of %d MiB; "
                "results may be incomplete.  Consider pruning old entries.",
                p,
                size / (1024 * 1024),
                _MAX_SYMLOG_BYTES // (1024 * 1024),
            )
        lines = p.read_text(encoding="utf-8", errors="replace").splitlines()
    except OSError as exc:
        logger.warning("⚠️ Could not read symlog %s: %s", p, exc)
        return []

    entries: list[SymlogEntry] = []
    for line in reversed(lines):
        if not line.strip():
            continue
        entry = _parse_line(line)
        if entry is not None:
            entries.append(entry)
        if len(entries) >= limit:
            break

    if follow and entries:
        # Check the last (oldest) entry for a born-from pointer.
        oldest = entries[-1]
        if oldest.born_from:
            prior_entries = read_symlog(
                repo_root,
                oldest.born_from,
                limit=max(limit, 10_000),
                follow=True,
            )
            entries.extend(prior_entries)

    return entries


# ---------------------------------------------------------------------------
# Enumerate
# ---------------------------------------------------------------------------

def list_symlog_symbols(repo_root: pathlib.Path) -> list[str]:
    """Return all symbol addresses that have a journal, sorted.

    Symlinks are excluded — only regular files are yielded.
    """
    root = _symlogs_dir(repo_root)
    if not root.exists():
        return []
    results: list[str] = []
    for leaf in root.rglob("*"):
        if leaf.name == _LEAF_INDEX_FILENAME:
            continue
        if leaf.is_file() and not leaf.is_symlink():
            try:
                results.append(_addr_from_path(root, leaf))
            except ValueError:
                continue
    return sorted(results)


def list_symlog_symbols_for_file(
    repo_root: pathlib.Path,
    file_path: str,
) -> list[str]:
    """Return symbol addresses for all symbols in *file_path* that have a journal.

    Uses a single ``os.listdir`` — O(1) regardless of total journal count.
    Symlinks are excluded.

    Args:
        repo_root:  Root of the Muse repository.
        file_path:  Repo-relative path to the source file, e.g. ``"src/billing.py"``.
    """
    symlogs_root = _symlogs_dir(repo_root)
    file_dir = symlogs_root / file_path
    if not file_dir.exists():
        return []
    results: list[str] = []
    for p in file_dir.iterdir():
        if p.name == _LEAF_INDEX_FILENAME:
            continue
        if p.is_file() and not p.is_symlink():
            try:
                results.append(_addr_from_path(symlogs_root, p))
            except ValueError:
                continue
    return sorted(results)


# ---------------------------------------------------------------------------
# Expire
# ---------------------------------------------------------------------------

def expire_symlog(
    repo_root: pathlib.Path,
    symbol_addr: str,
    expire_days: int,
    *,
    dry_run: bool = False,
) -> tuple[int, int]:
    """Remove entries older than *expire_days* from one symbol journal.

    The write is atomic: content is first written to a sibling ``.lock``
    temp file then renamed into place with ``os.replace``.  An empty journal
    after expiry is deleted rather than left as a zero-byte stub.

    Args:
        repo_root:    Root of the Muse repository.
        symbol_addr:  ``"path/to/file.py::symbol_name"``.
        expire_days:  Entries whose timestamp is older than this many days
                      are removed.
        dry_run:      When ``True``, compute counts but write nothing.

    Returns:
        ``(expired_count, kept_count)``.
    """
    p = symlog_path(repo_root, symbol_addr)
    if not p.exists():
        return 0, 0

    try:
        raw = p.read_text(encoding="utf-8", errors="replace")
    except OSError as exc:
        logger.warning("⚠️ symlog expire: could not read %s: %s", p, exc)
        return 0, 0

    cutoff_ts = int(datetime.datetime.now(tz=datetime.timezone.utc).timestamp()) - expire_days * 86400

    kept_lines: list[str] = []
    expired = 0
    kept = 0
    for line in raw.splitlines(keepends=True):
        stripped = line.strip()
        if not stripped:
            continue
        entry = _parse_line(stripped)
        if entry is None:
            kept_lines.append(line)
            kept += 1
            continue
        entry_ts = int(entry.timestamp.timestamp())
        if entry_ts < cutoff_ts:
            expired += 1
        else:
            kept_lines.append(line)
            kept += 1

    if dry_run:
        return expired, kept

    if not kept_lines:
        try:
            p.unlink()
        except OSError as exc:
            logger.warning("⚠️ symlog expire: could not remove %s: %s", p, exc)
        return expired, kept

    tmp_path = p.with_suffix(".lock")
    try:
        tmp_path.write_text("".join(kept_lines), encoding="utf-8")
        os.replace(tmp_path, p)
    except OSError as exc:
        logger.warning("⚠️ symlog expire: atomic write failed for %s: %s", p, exc)
        try:
            tmp_path.unlink(missing_ok=True)
        except OSError:
            pass
        raise

    return expired, kept


# ---------------------------------------------------------------------------
# Delete
# ---------------------------------------------------------------------------

def delete_symlog_entry(
    repo_root: pathlib.Path,
    symbol_addr: str,
    index: int | None,
) -> tuple[int, int]:
    """Remove one entry (by index) or all entries from a symbol journal.

    Args:
        repo_root:    Root of the Muse repository.
        symbol_addr:  ``"path/to/file.py::symbol_name"``.
        index:        0-based index from newest (``@{0}`` = newest).
                      Pass ``None`` to delete **all** entries.

    Returns:
        ``(deleted_count, remaining_count)``.

    Raises:
        FileNotFoundError:  Journal file does not exist (only when index is not None).
        IndexError:         *index* is out of range; args are ``(index, total)``.

    The write is atomic when entries remain.  An empty journal is deleted
    rather than left as a zero-byte stub.
    """
    p = symlog_path(repo_root, symbol_addr)

    if not p.exists():
        if index is None:
            return 0, 0  # --all on missing log is graceful
        raise FileNotFoundError(p)

    raw = p.read_text(encoding="utf-8", errors="replace")
    all_lines = [ln for ln in raw.splitlines(keepends=True) if ln.strip()]
    total = len(all_lines)

    if index is None:
        try:
            p.unlink()
        except OSError as exc:
            logger.warning("⚠️ symlog delete --all: could not remove %s: %s", p, exc)
        return total, 0

    if index < 0 or index >= total:
        raise IndexError(index, total)

    # index is 0-based from newest; lines are stored oldest-first.
    line_pos = total - 1 - index
    kept_lines = all_lines[:line_pos] + all_lines[line_pos + 1:]

    if not kept_lines:
        try:
            p.unlink()
        except OSError as exc:
            logger.warning("⚠️ symlog delete: could not remove %s: %s", p, exc)
        return 1, 0

    tmp_path = p.with_suffix(".lock")
    try:
        tmp_path.write_text("".join(kept_lines), encoding="utf-8")
        os.replace(tmp_path, p)
    except OSError as exc:
        logger.warning("⚠️ symlog delete: atomic write failed for %s: %s", p, exc)
        try:
            tmp_path.unlink(missing_ok=True)
        except OSError:
            pass
        raise

    return 1, len(kept_lines)


# ---------------------------------------------------------------------------
# Phase 2 — Symbol diff and commit integration
# ---------------------------------------------------------------------------

@dataclass
class SymbolDiff:
    """Result of comparing two symbol maps (old vs. new content IDs).

    ``renames`` holds ``(old_name, new_name)`` pairs where the same content ID
    appears in both the deleted and created sets — same object, different name.
    Renamed symbols are removed from ``created`` and ``deleted`` so callers
    never double-count them.
    """

    created:  set[str]
    deleted:  set[str]
    modified: set[str]
    renames:  list[tuple[str, str]]


def compute_symbol_diff(
    old_symbols: dict[str, str],
    new_symbols: dict[str, str],
    old_rename_keys: dict[str, str] | None = None,
    new_rename_keys: dict[str, str] | None = None,
) -> SymbolDiff:
    """Diff two ``{symbol_name: content_id}`` maps.

    Rename detection: a name that disappeared in *old* whose rename key matches
    that of a new name that didn't exist before is classified as a rename rather
    than a delete + create.

    *old_rename_keys* / *new_rename_keys* are optional ``{name: key}`` maps
    used exclusively for rename matching — typically ``body_hash`` values from
    ``parse_symbols``, which are stable across pure renames.  When not supplied,
    the ``content_id`` from *old_symbols* / *new_symbols* is used as the key
    instead (correct for unit tests with artificial IDs; rarely matches in
    practice for real code because ``content_id`` encodes the function name).

    A content-ID change with a name change is always treated as delete + create.
    """
    _old_rk = old_rename_keys if old_rename_keys is not None else old_symbols
    _new_rk = new_rename_keys if new_rename_keys is not None else new_symbols

    old_names = set(old_symbols)
    new_names = set(new_symbols)

    raw_deleted  = old_names - new_names
    raw_created  = new_names - old_names
    modified     = {n for n in old_names & new_names if old_symbols[n] != new_symbols[n]}

    # Build an inverted map of rename_key → name for the *created* side.
    created_by_rk: dict[str, str] = {}
    for name in raw_created:
        rk = _new_rk.get(name, "")
        if not rk:
            continue
        # Only map when a single created name has this key — ambiguous keys
        # cannot be deterministically paired with a deleted name.
        if rk not in created_by_rk:
            created_by_rk[rk] = name
        else:
            created_by_rk[rk] = ""  # sentinel: ambiguous

    renames: list[tuple[str, str]] = []
    confirmed_deleted  = set(raw_deleted)
    confirmed_created  = set(raw_created)

    for old_name in sorted(raw_deleted):
        rk = _old_rk.get(old_name, "")
        new_name = created_by_rk.get(rk) if rk else None
        if new_name:  # non-empty sentinel → unambiguous match
            renames.append((old_name, new_name))
            confirmed_deleted.discard(old_name)
            confirmed_created.discard(new_name)

    return SymbolDiff(
        created=confirmed_created,
        deleted=confirmed_deleted,
        modified=modified,
        renames=renames,
    )


def extract_symbols(
    repo_root: pathlib.Path,
    file_object_id: str,
    file_path: str = "",
) -> dict[str, str]:
    """Read a file object from the store and return ``{symbol_name: content_id}``.

    Uses the code AST parser to extract top-level symbols.  Returns an empty
    dict for objects that are not present in the store, cannot be parsed, or
    are non-code files.  Never raises — parse failures are silently swallowed
    because symlog writes are non-fatal.

    Args:
        repo_root:      Root of the Muse repository.
        file_object_id: Content-addressed ID of the file object (``sha256:…``).
        file_path:      Repo-relative file path passed to ``parse_symbols`` so
                        symbol keys are properly namespaced (``"src/f.py::name"``).
    """
    try:
        from muse.core.object_store import read_object
        source = read_object(repo_root, file_object_id)
        if source is None:
            return {}
        from muse.plugins.code.ast_parser import parse_symbols
        # parse_symbols returns {full_addr: {content_id: …, …}}
        # Keys are "file.py::name" — we want just the name part.
        raw = parse_symbols(source, file_path)
        result: dict[str, str] = {}
        for addr, sym in raw.items():
            cid = sym.get("content_id", "")
            if not cid:
                continue
            name = addr.split("::")[-1] if "::" in addr else addr
            result[name] = cid
        return result
    except Exception:
        return {}


def _extract_body_hashes(
    repo_root: pathlib.Path,
    file_object_id: str,
    file_path: str = "",
) -> dict[str, str]:
    """Return ``{symbol_name: body_hash}`` for rename detection.

    ``body_hash`` hashes only the function body, not the name — two functions
    with the same body but different names share the same ``body_hash``, making
    it the correct key for rename matching.  Falls back to ``content_id`` if
    ``body_hash`` is absent from the symbol record.
    """
    try:
        from muse.core.object_store import read_object
        source = read_object(repo_root, file_object_id)
        if source is None:
            return {}
        from muse.plugins.code.ast_parser import parse_symbols
        raw = parse_symbols(source, file_path)
        result: dict[str, str] = {}
        for addr, sym in raw.items():
            key = sym.get("body_hash", "") or sym.get("content_id", "")
            if not key:
                continue
            name = addr.split("::")[-1] if "::" in addr else addr
            result[name] = key
        return result
    except Exception:
        return {}


def _write_symlogs(
    repo_root: pathlib.Path,
    parent_commit_id: str | None,
    new_snapshot_id: str,
    new_commit_id: str,
    author: str,
    commit_message: str,
) -> None:
    """Write symlog entries for every symbol that changed in this commit.

    Called by ``muse commit`` after the commit record is finalised.  Wrapped
    in a broad ``try/except`` at the call site — a failure here must never
    abort a commit.

    Algorithm:

    1. Build ``old_manifest`` from the parent commit (empty for initial commit).
    2. For each file whose object ID changed, extract old and new symbol maps.
    3. Diff the two maps to classify changes as created/modified/deleted/renamed.
    4. Append the appropriate symlog entry for each changed symbol.
    """
    from muse.core.object_store import read_object
    from muse.core.commits import read_commit
    from muse.core.snapshots import read_snapshot

    # Load parent manifest.
    old_manifest: dict[str, str] = {}
    if parent_commit_id is not None:
        parent_rec = read_commit(repo_root, parent_commit_id)
        if parent_rec is not None:
            snap_rec = read_snapshot(repo_root, parent_rec.snapshot_id)
            if snap_rec is not None:
                old_manifest = dict(snap_rec.manifest)

    # Load new manifest.
    new_snap_rec = read_snapshot(repo_root, new_snapshot_id)
    if new_snap_rec is None:
        return
    new_manifest: dict[str, str] = dict(new_snap_rec.manifest)

    first_line = (commit_message.splitlines()[0] if commit_message else "").strip()

    # Find files whose object IDs differ.
    all_paths = set(old_manifest) | set(new_manifest)
    for file_path in sorted(all_paths):
        old_obj = old_manifest.get(file_path, "")
        new_obj = new_manifest.get(file_path, "")
        if old_obj == new_obj:
            continue  # file unchanged at object level

        old_syms = extract_symbols(repo_root, old_obj, file_path) if old_obj else {}
        new_syms = extract_symbols(repo_root, new_obj, file_path) if new_obj else {}

        if not old_syms and not new_syms:
            continue  # non-code file

        old_bodies = _extract_body_hashes(repo_root, old_obj, file_path) if old_obj else {}
        new_bodies = _extract_body_hashes(repo_root, new_obj, file_path) if new_obj else {}
        diff = compute_symbol_diff(old_syms, new_syms, old_bodies, new_bodies)

        # Created
        for name in sorted(diff.created):
            append_symlog(
                repo_root,
                f"{file_path}::{name}",
                old_content_id=NULL_CONTENT_ID,
                new_content_id=new_syms[name],
                commit_id=new_commit_id,
                author=author,
                operation=f"symbol-created: {name}",
            )

        # Modified
        for name in sorted(diff.modified):
            append_symlog(
                repo_root,
                f"{file_path}::{name}",
                old_content_id=old_syms[name],
                new_content_id=new_syms[name],
                commit_id=new_commit_id,
                author=author,
                operation=f"symbol-modified: {first_line}",
            )

        # Deleted
        for name in sorted(diff.deleted):
            append_symlog(
                repo_root,
                f"{file_path}::{name}",
                old_content_id=old_syms[name],
                new_content_id=NULL_CONTENT_ID,
                commit_id=new_commit_id,
                author=author,
                operation=f"symbol-deleted: {name}",
            )

        # Renamed — two entries per rename
        for old_name, new_name in diff.renames:
            new_addr = f"{file_path}::{new_name}"
            old_addr = f"{file_path}::{old_name}"
            # Terminal entry on the old path
            append_symlog(
                repo_root,
                old_addr,
                old_content_id=old_syms[old_name],
                new_content_id=NULL_CONTENT_ID,
                commit_id=new_commit_id,
                author=author,
                operation=f"symbol-renamed-to: {new_addr}",
            )
            # Born-from entry on the new path
            append_symlog(
                repo_root,
                new_addr,
                old_content_id=NULL_CONTENT_ID,
                new_content_id=new_syms[new_name],
                commit_id=new_commit_id,
                author=author,
                operation=f"symbol-born-from: {old_addr}",
            )


# ---------------------------------------------------------------------------
# Phase 5 — @{N} ref resolution
# ---------------------------------------------------------------------------

#: Regex matching ``<addr>@{N}`` — anchored at both ends.
_SYMLOG_REF_RE = re.compile(r"^(.+)@\{(\d+)\}$")


@dataclass(frozen=True)
class SymlogResolution:
    """Result of resolving a ``<addr>@{N}`` symlog reference.

    ``followed_from`` is non-None only when ``--follow`` crossed a rename
    boundary — it holds the prior symbol address that contributed this entry.
    ``content_id`` equals ``new_content_id`` of the resolved ``SymlogEntry``.
    """

    symbol:       str                   # base symbol address (without @{N})
    index:        int                   # 0-based index (0 = newest)
    content_id:   str                   # new_content_id at this index
    commit_id:    str                   # commit that caused this entry
    operation:    str                   # lifecycle description
    timestamp:    datetime.datetime     # UTC-aware entry timestamp
    followed_from: str | None = None    # prior addr when follow crossed a boundary


def resolve_symlog_addr(
    spec: str,
    repo_root: pathlib.Path,
    *,
    follow: bool = False,
) -> "SymlogResolution | None":
    """Resolve a ``<addr>@{N}`` symlog reference to a :class:`SymlogResolution`.

    Args:
        spec:       Full spec string, e.g. ``"billing.py::compute_total@{2}"``.
        repo_root:  Root of the Muse repository.
        follow:     When ``True``, traverse rename chains so indices beyond the
                    primary symbol's log are resolved from prior symbol addresses.

    Returns:
        A :class:`SymlogResolution` on success, or ``None`` if *spec* does not
        match the ``@{N}`` pattern (so callers can use ``if result is None:``
        to fall through to normal processing).

    Raises:
        FileNotFoundError:  No journal exists for the resolved symbol address.
        IndexError:         ``(index, total)`` when *index* is out of range.
        ValueError:         *addr* is not a valid symbol address (missing ``::``).
    """
    m = _SYMLOG_REF_RE.match(spec)
    if m is None:
        return None

    addr = m.group(1)
    index = int(m.group(2))

    # Validate that the addr portion is a proper symbol address.
    # This will raise ValueError if :: is missing (propagated to caller).
    p = symlog_path(repo_root, addr)
    if not p.exists():
        raise FileNotFoundError(p)

    if follow:
        primary = read_symlog(repo_root, addr, limit=100_000, follow=False)
        all_entries = read_symlog(repo_root, addr, limit=100_000, follow=True)
        total = len(all_entries)
        if index >= total:
            raise IndexError(index, total)
        entry = all_entries[index]
        followed_from: str | None = None
        if index >= len(primary) and primary:
            followed_from = primary[-1].born_from
        return SymlogResolution(
            symbol=addr,
            index=index,
            content_id=entry.new_content_id,
            commit_id=entry.commit_id,
            operation=entry.operation,
            timestamp=entry.timestamp,
            followed_from=followed_from,
        )

    entries = read_symlog(repo_root, addr, limit=100_000, follow=False)
    total = len(entries)
    if index >= total:
        raise IndexError(index, total)
    entry = entries[index]
    return SymlogResolution(
        symbol=addr,
        index=index,
        content_id=entry.new_content_id,
        commit_id=entry.commit_id,
        operation=entry.operation,
        timestamp=entry.timestamp,
        followed_from=None,
    )


def resolve_symbol_body(
    repo_root: pathlib.Path,
    symbol_addr: str,
    commit_id: str,
) -> "dict | None":
    """Return symbol body info for *symbol_addr* at *commit_id*.

    Reconstructs the symbol by reading: commit record → snapshot manifest →
    file object → AST parse → slice by ``lineno``/``end_lineno``.  Symbol
    ``content_id`` values are AST-computed hashes not stored as raw objects,
    so this walk is the only correct reconstruction path.

    Returns a dict with keys ``source``, ``kind``, ``lineno``, ``end_lineno``,
    ``qualified_name``, ``file_path`` on success, or ``None`` on any failure.
    """
    if "::" not in symbol_addr:
        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 adapter_for_path

        commit_rec = read_commit(repo_root, commit_id)
        if commit_rec is None:
            return None
        snap_rec = read_snapshot(repo_root, commit_rec.snapshot_id)
        if snap_rec is None:
            return None
        file_obj_id = snap_rec.manifest.get(file_path)
        if file_obj_id is None:
            return None
        raw = read_object(repo_root, file_obj_id)
        if raw is None:
            return None

        adapter = adapter_for_path(file_path)
        tree = adapter.parse_symbols(raw, file_path)

        # Prefer exact qualified_name match, fall back to bare name.
        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()
        # lineno is 1-based inclusive; end_lineno is 1-based inclusive.
        source = "\n".join(lines[lineno - 1:end_lineno])
        return {
            "source": source,
            "kind": found["kind"],
            "lineno": lineno,
            "end_lineno": end_lineno,
            "qualified_name": found["qualified_name"],
            "file_path": file_path,
        }
    except Exception:
        return None