"""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:: TAB 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: old_content_id == NULL_CONTENT_ID. symbol-modified: Both content IDs are populated. symbol-deleted: new_content_id == NULL_CONTENT_ID. symbol-renamed-to: Terminal entry written to the OLD symbol path. new_content_id == NULL_CONTENT_ID. symbol-born-from: 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 . 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:: TAB """ 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 def read_symlog_with_follow( repo_root: pathlib.Path, symbol_addr: str, *, follow: bool, limit: int, ) -> tuple[list[SymlogEntry], dict[int, str]]: """Read symlog entries with optional follow, returning entries + from_symbol map. ``read_symlog(follow=True)`` already appends a renamed-from symbol's entries transparently, but the flat result loses which address each entry actually belongs to -- callers that need to annotate a rename boundary (``muse symlog`` read output, ``muse code blame``'s symlog-based fast path) need that per-entry origin. When ``follow=True``, entries that came from a prior symbol address (via a rename chain) are annotated in the returned ``from_symbol`` map (index → prior_addr); entries belonging to *symbol_addr* itself are not present in the map at all. """ 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_symlog_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, {} # --------------------------------------------------------------------------- # 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) # --------------------------------------------------------------------------- # Orphan pruning (muse#53 item 12) # --------------------------------------------------------------------------- def _symbol_exists_in_head(repo_root: pathlib.Path, symbol_addr: str) -> bool: """Return ``True`` if *symbol_addr*'s file and symbol both exist at HEAD. Never raises — any failure along the way (no HEAD commit, file removed, parse failure) is treated as "doesn't exist", matching :func:`extract_symbols`'s own fail-silent contract. """ if "::" not in symbol_addr: return False file_path, _, symbol_name = symbol_addr.partition("::") if not symbol_name: return False try: from muse.core.refs import get_head_commit_id, read_current_branch from muse.core.commits import read_commit from muse.core.snapshots import read_snapshot branch = read_current_branch(repo_root) commit_id = get_head_commit_id(repo_root, branch) if commit_id is None: return False commit_rec = read_commit(repo_root, commit_id) if commit_rec is None: return False snap_rec = read_snapshot(repo_root, commit_rec.snapshot_id) if snap_rec is None: return False file_obj_id = snap_rec.manifest.get(file_path) if file_obj_id is None: return False symbols = extract_symbols(repo_root, file_obj_id, file_path) return symbol_name in symbols except Exception: return False def find_orphan_symlogs(repo_root: pathlib.Path) -> list[str]: """Return symbol addresses with a journal whose symbol no longer exists at HEAD. A symbol is orphaned when either its file was deleted/renamed away, or the symbol itself was removed from that file — in both cases the journal under ``.muse/symlogs/`` survives (journals are append-only and never auto-deleted on symbol removal) with no live symbol to show it for. """ return [ addr for addr in list_symlog_symbols(repo_root) if not _symbol_exists_in_head(repo_root, addr) ] def prune_orphan_symlogs( repo_root: pathlib.Path, *, dry_run: bool = False, ) -> list[str]: """Delete (or report, with *dry_run*) every orphaned symlog journal. Reuses :func:`delete_symlog_entry` with ``index=None`` for the actual removal — the same "delete the whole journal" path ``muse symlog delete --all`` already uses and already has tested atomic-delete behavior for. Returns: The list of symbol addresses pruned (or that would be pruned, under ``dry_run``). """ orphans = find_orphan_symlogs(repo_root) if dry_run: return orphans for addr in orphans: try: delete_symlog_entry(repo_root, addr, None) except FileNotFoundError: continue return orphans # --------------------------------------------------------------------------- # Analytics: hotspots and coupling (muse#53 item 1) # --------------------------------------------------------------------------- def compute_symlog_hotspots( repo_root: pathlib.Path, *, file_path: str | None = None, top: int = 20, ) -> list[tuple[str, int]]: """Rank symbols by churn — number of ``symbol-modified`` journal entries. Reads the already-written symlog journal in O(1) per symbol (no commit history scan) — always up to date, unlike ``muse code hotspots`` which walks the commit DAG. Only ``symbol-modified`` entries count toward churn; creation/deletion/rename events don't reflect ongoing edit activity the way repeated modification does. Args: repo_root: Root of the Muse repository. file_path: Restrict to symbols in this file, or ``None`` for every symbol with a journal. top: Maximum number of ranked symbols to return. Returns: ``[(symbol_addr, modification_count), ...]`` sorted by count descending, ties broken by symbol address ascending for determinism. """ addrs = ( list_symlog_symbols_for_file(repo_root, file_path) if file_path is not None else list_symlog_symbols(repo_root) ) counts: list[tuple[str, int]] = [] for addr in addrs: entries = read_symlog(repo_root, addr, limit=100_000, follow=False) modified = sum( 1 for e in entries if e.operation.split(":", 1)[0].strip() == "symbol-modified" ) if modified > 0: counts.append((addr, modified)) counts.sort(key=lambda kv: (-kv[1], kv[0])) return counts[:top] def compute_symlog_coupling( repo_root: pathlib.Path, *, file_path: str | None = None, min_count: int = 2, top: int = 20, ) -> list[tuple[str, str, int]]: """Find symbol pairs that change together most often. Two symbols are "coupled" when their journals share commit IDs — a commit touched both. Surfaced purely from already-written symlog data, no AST traversal of the commit history needed. Args: repo_root: Root of the Muse repository. file_path: Restrict to symbols in this file, or ``None`` for every symbol with a journal. min_count: Minimum number of shared commits for a pair to be included (default 2 — one shared commit is too common to be a meaningful signal). top: Maximum number of ranked pairs to return. Returns: ``[(symbol_a, symbol_b, shared_commit_count), ...]`` sorted by count descending, ties broken by ``(symbol_a, symbol_b)`` ascending. ``symbol_a < symbol_b`` lexicographically within each tuple, so a pair never appears twice in swapped order. """ addrs = ( list_symlog_symbols_for_file(repo_root, file_path) if file_path is not None else list_symlog_symbols(repo_root) ) commit_sets: dict[str, set[str]] = {} for addr in addrs: entries = read_symlog(repo_root, addr, limit=100_000, follow=False) commit_sets[addr] = {e.commit_id for e in entries} pairs: list[tuple[str, str, int]] = [] sorted_addrs = sorted(commit_sets) for i, a in enumerate(sorted_addrs): for b in sorted_addrs[i + 1:]: shared = len(commit_sets[a] & commit_sets[b]) if shared >= min_count: pairs.append((a, b, shared)) pairs.sort(key=lambda t: (-t[2], t[0], t[1])) return pairs[:top] # --------------------------------------------------------------------------- # 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 ``@{N}`` or ``@{}`` — anchored at both #: ends. Group 2 is disambiguated (digit index vs relative-time spec) inside #: ``resolve_symlog_addr`` / ``_resolve_symlog_index`` (muse#53 item 6). _SYMLOG_REF_RE = re.compile(r"^(.+)@\{([^}]+)\}$") #: Matches "N.unit.ago" relative-time specs, e.g. "1.hour.ago", "2.days.ago". _SYMLOG_TIME_AGO_RE = re.compile( r"^(?P\d+)\.(?Psecond|minute|hour|day|week)s?\.ago$" ) _SYMLOG_TIME_UNIT_SECONDS = { "second": 1, "minute": 60, "hour": 3600, "day": 86_400, "week": 604_800, } def _parse_symlog_time_cutoff(inner: str) -> datetime.datetime | None: """Parse a relative-time ``@{...}`` spec into an absolute UTC cutoff. Supported forms (muse#53 item 6, mirrors the reflog equivalent): - ``N.unit.ago`` — unit is second/minute/hour/day/week, singular or plural (e.g. ``"1.hour.ago"``, ``"2.days.ago"``). - ``yesterday`` — start of today (UTC midnight). Returns ``None`` for anything that isn't a recognized time spec. """ now = datetime.datetime.now(datetime.timezone.utc) if inner == "yesterday": return now.replace(hour=0, minute=0, second=0, microsecond=0) m = _SYMLOG_TIME_AGO_RE.match(inner) if m is None: return None num = int(m.group("num")) unit_seconds = _SYMLOG_TIME_UNIT_SECONDS[m.group("unit")] return now - datetime.timedelta(seconds=num * unit_seconds) def _resolve_symlog_index(inner: str, entries: "list[SymlogEntry]") -> int: """Resolve the inner content of ``@{...}`` to a 0-based index into *entries*. *entries* must already be newest-first (the ``read_symlog`` contract). Digit-only *inner* resolves by position; anything else is tried as a relative-time spec, resolving to the first (newest) entry at or before the computed cutoff. Raises: IndexError: ``(index, total)`` for an out-of-range digit index, or ``(-1, total)`` when a valid time spec has no qualifying entry. ValueError: *inner* is neither a digit index nor a recognized relative-time spec. """ total = len(entries) if inner.isdigit(): index = int(inner) if index >= total: raise IndexError(index, total) return index cutoff = _parse_symlog_time_cutoff(inner) if cutoff is None: raise ValueError( f"{inner!r} inside @{{...}} is neither an index nor a recognized " "relative-time spec (e.g. '1.hour.ago', 'yesterday')" ) for idx, entry in enumerate(entries): if entry.timestamp <= cutoff: return idx raise IndexError(-1, total) @dataclass(frozen=True) class SymlogResolution: """Result of resolving a ``@{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 author: str = "" # author/agent_id recorded on the entry (muse#53 item 9) def resolve_symlog_addr( spec: str, repo_root: pathlib.Path, *, follow: bool = False, ) -> "SymlogResolution | None": """Resolve a ``@{N}`` symlog reference to a :class:`SymlogResolution`. Args: spec: Full spec string, e.g. ``"billing.py::compute_total@{2}"``, or a relative-time spec (muse#53 item 6), e.g. ``"billing.py::compute_total@{1.hour.ago}"``. 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 ``@{...}`` 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)`` for an out-of-range digit index, or ``(-1, total)`` when a time spec has no qualifying entry. ValueError: *addr* is not a valid symbol address (missing ``::``), or the ``@{...}`` content is neither a digit index nor a recognized relative-time spec. """ m = _SYMLOG_REF_RE.match(spec) if m is None: return None addr = m.group(1) inner = 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) index = _resolve_symlog_index(inner, all_entries) 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, author=entry.author, ) entries = read_symlog(repo_root, addr, limit=100_000, follow=False) index = _resolve_symlog_index(inner, entries) 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, author=entry.author, ) 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