"""Commit-graph-derived bootstrap context for ``knowtation://prime``. Phase 4.5 — ``knowtation://prime`` regenerated from commit graph. This module provides :func:`build_prime_context`, which walks the local Muse commit graph and distils two key signals that the Knowtation MCP ``knowtation://prime`` resource needs to give agents a useful session bootstrap: ``last_consolidation`` The most recent commit that carried ``metadata["event_type"] == "consolidation"`` (or ``"consolidation_pass"``). Agents use this to know when the vault was last automatically cleaned up. ``hot_notes`` Notes that appear most frequently in structured deltas across the *N* most recent commits. An agent reading ``knowtation://prime`` can pre-fetch those notes because they are the most actively edited and therefore most relevant. The module is intentionally stateless and pure: it takes a repo root and a max-commit cap, reads the commit graph, and returns a :class:`PrimeContext` dataclass. No network I/O, no mutations. The Knowtation MCP ``knowtation://prime`` resource calls this when Muse is available (guarded by the ``KNOWTATION_MUSE_ENABLED`` environment variable on the JS side) and falls back to its existing static payload when this function is unavailable. When Phase 5 MCP tooling ships, the same data will be exposed via the ``knowtation/prime-context`` MCP tool so any MCP client (including the Knowtation server running remotely) can fetch it over JSON-RPC. Security -------- * All strings read from commit records pass through :func:`_safe_str` before being stored in the returned dataclass, preventing control-character injection when the context is later serialised to JSON and displayed. * No user input enters this module; *root* must be a :class:`pathlib.Path` pointing to a Muse repo root (validated by :func:`muse.core.repo.require_repo` before calling). * The commit walk is bounded by *max_commits* (default 100) to prevent DoS on repos with very long histories. """ from __future__ import annotations import logging import re from dataclasses import dataclass, field from pathlib import Path from typing import Any logger = logging.getLogger(__name__) # Module-level imports so they are patchable in tests and have a single # import cost per process rather than per call. ImportError is raised at # import time when this plugin module is loaded in a Muse environment; the # caller (build_prime_context) will catch it inside the broader try/except. from muse.core.store import ( # noqa: E402 get_head_commit_id, read_commit, read_current_branch, ) from muse.plugins.code._query import walk_commits_bfs # noqa: E402 #: Schema version for the PrimeContext payload. Bump when the shape changes #: in a breaking way; consumers should check this to detect schema drift. PRIME_CONTEXT_SCHEMA_VERSION: str = "1.0.0" #: How many of the most-edited notes to include in ``hot_notes``. _DEFAULT_HOT_NOTE_COUNT: int = 10 #: Maximum number of commits to walk when building the context. _DEFAULT_MAX_COMMITS: int = 100 #: Event kinds that count as a "consolidation". _CONSOLIDATION_KINDS: frozenset[str] = frozenset({"consolidation", "consolidation_pass"}) #: Regex for stripping C0/DEL/C1 control characters from strings read from #: commit records (same policy as :func:`muse.core.validation.sanitize_provenance`). _CONTROL_CHARS_RE: re.Pattern[str] = re.compile(r"[\x00-\x1f\x7f\x80-\x9f]") def _safe_str(value: Any) -> str: """Return a control-char-stripped string representation of *value*. Args: value: Any value (typically a str from a commit record field). Returns: String with all C0/DEL/C1 control characters removed. Returns an empty string for ``None`` inputs. """ if value is None: return "" return _CONTROL_CHARS_RE.sub("", str(value)) @dataclass(frozen=True) class ConsolidationRecord: """Summary of the most recent consolidation commit. Attributes: commit_id: Full commit ID (sha256 hex). committed_at: ISO-8601 timestamp string. message: Commit message (control-char stripped). agent_id: Agent that ran the consolidation (empty when absent). model_id: Model used for the consolidation (empty when absent). """ commit_id: str committed_at: str message: str agent_id: str model_id: str def to_dict(self) -> dict[str, str]: """Serialise to a plain dict for JSON embedding. Returns: Dict with all fields as strings. """ return { "commit_id": self.commit_id, "committed_at": self.committed_at, "message": self.message, "agent_id": self.agent_id, "model_id": self.model_id, } @dataclass(frozen=True) class HotNote: """A frequently-edited note from the recent commit graph. Attributes: path: Vault-relative note path (e.g. ``"notes/session.md"``). edits: Number of commits in the walk that touched this note. """ path: str edits: int def to_dict(self) -> dict[str, Any]: """Serialise to a plain dict for JSON embedding. Returns: Dict with ``path`` (str) and ``edits`` (int). """ return {"path": self.path, "edits": self.edits} @dataclass(frozen=True) class PrimeContext: """Commit-graph-derived bootstrap context for ``knowtation://prime``. Attributes: schema_version: Always :data:`PRIME_CONTEXT_SCHEMA_VERSION`. commits_scanned: Number of commits examined during the walk. last_consolidation: Most recent consolidation commit summary, or ``None`` if no consolidation was found in the walk window. hot_notes: Top-N most-edited notes from the walk window, sorted by edit count descending. source: ``"muse-commit-graph"`` — identifies this as commit-graph-derived data so consumers can distinguish it from the static fallback. """ schema_version: str commits_scanned: int last_consolidation: ConsolidationRecord | None hot_notes: list[HotNote] = field(default_factory=list) source: str = "muse-commit-graph" def to_dict(self) -> dict[str, Any]: """Serialise to a plain dict suitable for JSON embedding in the prime payload. Returns: Dict with all fields. ``last_consolidation`` is ``null`` (``None``) when no consolidation was found in the walk window. """ return { "schema_version": self.schema_version, "source": self.source, "commits_scanned": self.commits_scanned, "last_consolidation": ( self.last_consolidation.to_dict() if self.last_consolidation is not None else None ), "hot_notes": [n.to_dict() for n in self.hot_notes], } def _extract_note_paths_from_delta(structured_delta: Any) -> list[str]: """Extract unique note file paths from a structured delta's op list. Iterates the ops, looking for addresses that look like knowtation note paths (end in ``.md`` or contain ``::section:`` — both indicate the op belongs to a note file). Args: structured_delta: A :class:`~muse.domain.StructuredDelta` dict or ``None``. Returns: Deduplicated list of note file paths (vault-relative) mentioned in the delta. Empty list when *structured_delta* is ``None`` or has no note-path ops. """ if structured_delta is None: return [] ops = structured_delta.get("ops", []) paths: set[str] = set() for op in ops: if not isinstance(op, dict): continue addr = op.get("address", "") if not isinstance(addr, str): continue # Section-level addresses have the form "notes/file.md::section:…" if "::" in addr: path_part = addr.split("::", 1)[0] if path_part.endswith(".md"): paths.add(_safe_str(path_part)) elif addr.endswith(".md"): paths.add(_safe_str(addr)) return list(paths) def build_prime_context( root: Path, *, max_commits: int = _DEFAULT_MAX_COMMITS, hot_note_count: int = _DEFAULT_HOT_NOTE_COUNT, ) -> PrimeContext: """Build commit-graph-derived bootstrap context for ``knowtation://prime``. Walks the most recent commits on the current branch, extracts: * The most recent consolidation commit (``event_type`` in ``{"consolidation", "consolidation_pass"}``). * A frequency map of note paths from structured deltas to surface hot notes. The walk is bounded by *max_commits* to prevent DoS on long histories. On any I/O or deserialization error the function returns a :class:`PrimeContext` with the data collected up to the failure point (partial results are better than no results for the bootstrap use case). Args: root: Path to the Muse repo root (contains ``.muse/``). max_commits: Maximum commits to inspect (default 100). hot_note_count: Number of hot notes to include in the result (default 10). Returns: A :class:`PrimeContext` with schema_version, commits_scanned, last_consolidation, and hot_notes populated. """ try: branch = read_current_branch(root) head_id = get_head_commit_id(root, branch) if head_id is None: return PrimeContext( schema_version=PRIME_CONTEXT_SCHEMA_VERSION, commits_scanned=0, last_consolidation=None, hot_notes=[], ) commits, _truncated = walk_commits_bfs(root, head_id, max_commits) except Exception as exc: logger.warning("build_prime_context: error reading commit graph: %s", exc) return PrimeContext( schema_version=PRIME_CONTEXT_SCHEMA_VERSION, commits_scanned=0, last_consolidation=None, hot_notes=[], ) note_edit_counts: dict[str, int] = {} last_consolidation: ConsolidationRecord | None = None for commit in commits: # Check for consolidation event type. event_type = commit.metadata.get("event_type", "") if last_consolidation is None and event_type in _CONSOLIDATION_KINDS: last_consolidation = ConsolidationRecord( commit_id=_safe_str(commit.commit_id), committed_at=_safe_str(commit.committed_at.isoformat()), message=_safe_str(commit.message), agent_id=_safe_str(commit.agent_id), model_id=_safe_str(commit.model_id), ) # Accumulate note-path edit counts from the structured delta. for path in _extract_note_paths_from_delta(commit.structured_delta): note_edit_counts[path] = note_edit_counts.get(path, 0) + 1 hot = sorted(note_edit_counts.items(), key=lambda kv: (-kv[1], kv[0])) hot_notes = [HotNote(path=p, edits=c) for p, c in hot[:hot_note_count]] return PrimeContext( schema_version=PRIME_CONTEXT_SCHEMA_VERSION, commits_scanned=len(commits), last_consolidation=last_consolidation, hot_notes=hot_notes, ) __all__ = [ "PRIME_CONTEXT_SCHEMA_VERSION", "ConsolidationRecord", "HotNote", "PrimeContext", "build_prime_context", ]