"""Knowtation note metrics — Phase 3.3 helpers. This module supplies the vault-aware counterparts to six ``muse code`` commands that the CLI routes to when the active domain is ``knowtation``: ============= ====================================================== CLI command Helper exported here ============= ====================================================== hotspots :func:`note_hotspots` — note churn by commit count gravity :func:`note_gravity` — fraction of vault linking in dead :func:`note_dead` — notes with zero inbound links clones :func:`note_clones` — duplicate / near-duplicate notes entangle :func:`note_entangle` — co-change note pairs velocity :func:`note_velocity` — note additions per commit window ============= ====================================================== Each helper returns a plain Python dict suitable for ``json.dumps``. Design ------ - Helpers that need a commit graph accept ``commits`` (a sequence of :class:`~muse.core.store.CommitRecord`) and use ``snapshot_manifest`` lookups per commit; we never re-walk the DAG inside helpers. - Helpers that need the live link graph accept a pre-built :class:`~muse.plugins.knowtation.link_index.LinkIndex`. The CLI builds it once and passes it through. - All helpers exclude templates / meta / hidden directories implicitly, because the link index already does this and the manifest-walk helpers delegate to :func:`is_note`. Performance budgets (per Plan §3.3): - < 2s on a 5k-note vault for each command. """ from __future__ import annotations import logging import pathlib from collections import Counter, defaultdict from typing import Any, Iterable from muse.plugins.knowtation._query import is_note from muse.plugins.knowtation.link_index import LinkIndex logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Manifest helpers # --------------------------------------------------------------------------- def _notes_in_manifest(manifest: dict[str, str]) -> dict[str, str]: return {p: h for p, h in manifest.items() if is_note(p)} def _changed_notes_between( older: dict[str, str], newer: dict[str, str] ) -> set[str]: """Set of note paths whose content_hash differs (added, removed, or modified). Args: older: Older snapshot manifest. newer: Newer snapshot manifest. Returns: Set of note paths that changed. """ changed: set[str] = set() older_notes = _notes_in_manifest(older) newer_notes = _notes_in_manifest(newer) for path, new_hash in newer_notes.items(): if older_notes.get(path) != new_hash: changed.add(path) for path in older_notes: if path not in newer_notes: changed.add(path) return changed # --------------------------------------------------------------------------- # hotspots — note churn leaderboard # --------------------------------------------------------------------------- def note_hotspots( commits: list, snapshot_loader: Any, *, top: int = 20, min_changes: int = 1, project_filter: str | None = None, ) -> dict[str, Any]: """Rank notes by the number of commits in *commits* that touched them. Args: commits: List of CommitRecord, ordered newest-first. snapshot_loader: Callable ``(commit_id) → manifest`` for fetching each commit's snapshot manifest. Pass a closure that delegates to ``get_commit_snapshot_manifest``. top: Maximum entries in the returned ``ranking`` list. min_changes: Minimum change count required for inclusion. project_filter: Optional path-prefix filter (e.g. ``"projects/"``). Returns: Dict with keys: - ``commits_analysed``: int — number of commits considered. - ``total_notes_touched``: int — distinct notes that changed at least once. - ``ranking``: list of ``{path, changes}`` dicts, sorted by ``changes`` (descending) then path (ascending). """ counts: Counter[str] = Counter() prev_manifest: dict[str, str] | None = None # Walk newest → oldest; compare each commit with its predecessor (older). for commit in commits: try: current = snapshot_loader(commit.commit_id) or {} except Exception as exc: logger.debug("hotspots: snapshot load failed for %s: %s", commit.commit_id, exc) continue if prev_manifest is None: prev_manifest = current continue # commit is older than prev_manifest commit changed = _changed_notes_between(current, prev_manifest) for path in changed: if project_filter and not path.startswith(project_filter): continue counts[path] += 1 prev_manifest = current ranking_pairs = [(path, n) for path, n in counts.items() if n >= min_changes] ranking_pairs.sort(key=lambda x: (-x[1], x[0])) top_n = ranking_pairs[:top] return { "commits_analysed": len(commits), "total_notes_touched": len(counts), "ranking": [{"path": path, "changes": n} for path, n in top_n], } # --------------------------------------------------------------------------- # gravity — structural weight (fraction of vault linking to a note) # --------------------------------------------------------------------------- def note_gravity( index: LinkIndex, *, top: int = 20, min_inbound: int = 1, ) -> dict[str, Any]: """Rank notes by structural weight (inbound link count) in the live graph. Gravity = ``incoming_count(note) / total_notes`` — the fraction of the vault that depends on this note. Args: index: Pre-built :class:`LinkIndex`. top: Maximum entries in ``ranking``. min_inbound: Minimum inbound link count for inclusion. Returns: Dict with keys ``total_notes``, ``ranking`` (list of ``{path, inbound, gravity}`` dicts sorted by ``inbound`` descending). """ total_notes = max(1, index.notes_indexed) counts: dict[str, int] = {} for target, links in index.backward.items(): # Count distinct source notes (a single source may link multiple times). counts[target] = len({link.source for link in links}) pairs = [(p, n) for p, n in counts.items() if n >= min_inbound] pairs.sort(key=lambda x: (-x[1], x[0])) top_n = pairs[:top] return { "total_notes": total_notes, "ranking": [ { "path": path, "inbound": n, "gravity": round(n / total_notes, 6), } for path, n in top_n ], } # --------------------------------------------------------------------------- # dead — notes with zero inbound links and old mtime # --------------------------------------------------------------------------- def note_dead( index: LinkIndex, *, root: pathlib.Path, use_mtime: bool = True, mtime_days_threshold: int = 180, ) -> dict[str, Any]: """Find notes with no inbound links (and optionally, old mtime). Per Plan §3.3: until Phase 4 lands the agent-memory-event signal, dead detection uses inbound links + mtime, gated behind ``use_mtime`` to keep the strict definition (no inbound only) available too. Args: index: Pre-built :class:`LinkIndex`. root: Vault root (used for mtime lookups when ``use_mtime=True``). use_mtime: When ``True``, also require mtime older than ``mtime_days_threshold`` days. mtime_days_threshold: Mtime cutoff in days. Returns: Dict with keys: - ``total_notes_indexed``: int - ``dead_count``: int - ``dead_notes``: sorted list of ``{path, mtime?}`` dicts. """ import time cutoff = time.time() - (mtime_days_threshold * 86400) dead_notes: list[dict[str, Any]] = [] for path in sorted(index.forward.keys()): # No inbound resolved links means dead-by-link definition. if path in index.backward and index.backward[path]: continue entry: dict[str, Any] = {"path": path} if use_mtime: try: mtime = (root / path).stat().st_mtime except OSError: continue if mtime > cutoff: continue entry["mtime"] = mtime dead_notes.append(entry) return { "total_notes_indexed": index.notes_indexed, "dead_count": len(dead_notes), "dead_notes": dead_notes, } # --------------------------------------------------------------------------- # clones — duplicate notes by content hash # --------------------------------------------------------------------------- def note_clones( manifest: dict[str, str], ) -> dict[str, Any]: """Find groups of notes sharing the same content hash (exact duplicates). Args: manifest: Snapshot manifest (path → content_hash). Returns: Dict with keys: - ``total_notes``: int - ``clone_groups``: list of ``{hash, paths}`` dicts, where each group has 2+ paths sharing the same hash. - ``total_clones``: int — total number of clone instances (sum of len(paths) across all groups). """ notes = _notes_in_manifest(manifest) by_hash: dict[str, list[str]] = defaultdict(list) for path, content_hash in notes.items(): by_hash[content_hash].append(path) groups = [ {"hash": h, "paths": sorted(paths)} for h, paths in by_hash.items() if len(paths) > 1 ] groups.sort(key=lambda g: (-len(g["paths"]), g["paths"][0])) return { "total_notes": len(notes), "clone_groups": groups, "total_clones": sum(len(g["paths"]) for g in groups), } # --------------------------------------------------------------------------- # entangle — co-change note pairs # --------------------------------------------------------------------------- def note_entangle( commits: list, snapshot_loader: Any, *, min_co_changes: int = 2, top: int = 20, link_index: LinkIndex | None = None, attachment_weight: int = 1, ) -> dict[str, Any]: """Find note pairs that change together in the same commits. Phase 3.5 extension: when *link_index* is provided, mist-attachment co-references are added to the entanglement score. Two notes that reference the same mist ID gain ``attachment_weight`` co-change credits. Args: commits: List of CommitRecord, ordered newest-first. snapshot_loader: Callable ``(commit_id) → manifest``. min_co_changes: Minimum number of co-occurring commits for inclusion. top: Maximum entries in ``ranking``. link_index: Optional :class:`LinkIndex` for mist-attachment join. attachment_weight: Credits awarded per shared mist attachment. Returns: Dict with keys: - ``commits_analysed``: int - ``attachment_pairs``: int — number of pairs sharing mist IDs (0 if ``link_index`` is None). - ``ranking``: list of ``{pair: [path_a, path_b], co_changes: int, shared_attachments: int}`` dicts, sorted by co_changes desc then pair lex asc. """ pair_counts: Counter[tuple[str, str]] = Counter() attachment_pair_counts: Counter[tuple[str, str]] = Counter() prev_manifest: dict[str, str] | None = None for commit in commits: try: current = snapshot_loader(commit.commit_id) or {} except Exception: continue if prev_manifest is None: prev_manifest = current continue changed_notes = sorted(_changed_notes_between(current, prev_manifest)) for i, a in enumerate(changed_notes): for b in changed_notes[i + 1:]: pair_counts[(a, b)] += 1 prev_manifest = current # Phase 3.5: add mist-attachment co-references to the same counter. if link_index is not None: for _mist_id, members in link_index.attachment_index.items(): members_sorted = sorted(members) for i, a in enumerate(members_sorted): for b in members_sorted[i + 1:]: attachment_pair_counts[(a, b)] += 1 pair_counts[(a, b)] += attachment_weight pairs = [(pair, n) for pair, n in pair_counts.items() if n >= min_co_changes] pairs.sort(key=lambda x: (-x[1], x[0])) return { "commits_analysed": len(commits), "attachment_pairs": len(attachment_pair_counts), "ranking": [ { "pair": list(pair), "co_changes": n, "shared_attachments": attachment_pair_counts.get(pair, 0), } for pair, n in pairs[:top] ], } # --------------------------------------------------------------------------- # velocity — note additions per commit window # --------------------------------------------------------------------------- def note_velocity( commits: list, snapshot_loader: Any, *, window_size: int = 10, ) -> dict[str, Any]: """Compute note additions across two consecutive commit windows. Args: commits: List of CommitRecord, ordered newest-first. snapshot_loader: Callable ``(commit_id) → manifest``. window_size: Number of commits per window. Two windows are compared. Returns: Dict with keys: - ``window_size``: int - ``recent_window``: ``{added, removed, modified, net}`` - ``older_window``: ``{added, removed, modified, net}`` - ``acceleration``: int — recent.net - older.net """ def _window_stats(window_commits: list) -> dict[str, int]: """Compute add/remove/modify counts for a window.""" if not window_commits: return {"added": 0, "removed": 0, "modified": 0, "net": 0} try: newest = snapshot_loader(window_commits[0].commit_id) or {} oldest = snapshot_loader(window_commits[-1].commit_id) or {} except Exception: return {"added": 0, "removed": 0, "modified": 0, "net": 0} newest_notes = _notes_in_manifest(newest) oldest_notes = _notes_in_manifest(oldest) added = len(set(newest_notes) - set(oldest_notes)) removed = len(set(oldest_notes) - set(newest_notes)) modified = sum( 1 for p in (set(newest_notes) & set(oldest_notes)) if newest_notes[p] != oldest_notes[p] ) return { "added": added, "removed": removed, "modified": modified, "net": added - removed, } if window_size <= 0: raise ValueError(f"velocity: window_size must be positive, got {window_size}") recent = commits[:window_size] older = commits[window_size:window_size * 2] recent_stats = _window_stats(recent) older_stats = _window_stats(older) return { "window_size": window_size, "recent_window": recent_stats, "older_window": older_stats, "acceleration": recent_stats["net"] - older_stats["net"], } # --------------------------------------------------------------------------- # Convenience: walk commits + snapshot loader factory # --------------------------------------------------------------------------- def make_snapshot_loader(root: pathlib.Path): """Return a closure ``(commit_id) → manifest`` for use with these helpers. Args: root: Repository root. Returns: Callable that returns the snapshot manifest for *commit_id*, or an empty dict on failure. """ from muse.core.store import get_commit_snapshot_manifest def _loader(commit_id: str) -> dict[str, str]: try: return get_commit_snapshot_manifest(root, commit_id) or {} except Exception: return {} return _loader