prime_context.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Commit-graph-derived bootstrap context for ``knowtation://prime``. |
| 2 | |
| 3 | Phase 4.5 — ``knowtation://prime`` regenerated from commit graph. |
| 4 | |
| 5 | This module provides :func:`build_prime_context`, which walks the local Muse |
| 6 | commit graph and distils two key signals that the Knowtation MCP |
| 7 | ``knowtation://prime`` resource needs to give agents a useful session bootstrap: |
| 8 | |
| 9 | ``last_consolidation`` |
| 10 | The most recent commit that carried ``metadata["event_type"] == "consolidation"`` |
| 11 | (or ``"consolidation_pass"``). Agents use this to know when the vault was |
| 12 | last automatically cleaned up. |
| 13 | |
| 14 | ``hot_notes`` |
| 15 | Notes that appear most frequently in structured deltas across the *N* most |
| 16 | recent commits. An agent reading ``knowtation://prime`` can pre-fetch those |
| 17 | notes because they are the most actively edited and therefore most relevant. |
| 18 | |
| 19 | The module is intentionally stateless and pure: it takes a repo root and a |
| 20 | max-commit cap, reads the commit graph, and returns a :class:`PrimeContext` |
| 21 | dataclass. No network I/O, no mutations. |
| 22 | |
| 23 | The Knowtation MCP ``knowtation://prime`` resource calls this when Muse is |
| 24 | available (guarded by the ``KNOWTATION_MUSE_ENABLED`` environment variable on |
| 25 | the JS side) and falls back to its existing static payload when this function |
| 26 | is unavailable. |
| 27 | |
| 28 | When Phase 5 MCP tooling ships, the same data will be exposed via the |
| 29 | ``knowtation/prime-context`` MCP tool so any MCP client (including the |
| 30 | Knowtation server running remotely) can fetch it over JSON-RPC. |
| 31 | |
| 32 | Security |
| 33 | -------- |
| 34 | * All strings read from commit records pass through :func:`_safe_str` before |
| 35 | being stored in the returned dataclass, preventing control-character |
| 36 | injection when the context is later serialised to JSON and displayed. |
| 37 | * No user input enters this module; *root* must be a :class:`pathlib.Path` |
| 38 | pointing to a Muse repo root (validated by :func:`muse.core.repo.require_repo` |
| 39 | before calling). |
| 40 | * The commit walk is bounded by *max_commits* (default 100) to prevent DoS |
| 41 | on repos with very long histories. |
| 42 | """ |
| 43 | |
| 44 | from __future__ import annotations |
| 45 | |
| 46 | import logging |
| 47 | import re |
| 48 | from dataclasses import dataclass, field |
| 49 | from pathlib import Path |
| 50 | from typing import Any |
| 51 | |
| 52 | logger = logging.getLogger(__name__) |
| 53 | |
| 54 | # Module-level imports so they are patchable in tests and have a single |
| 55 | # import cost per process rather than per call. ImportError is raised at |
| 56 | # import time when this plugin module is loaded in a Muse environment; the |
| 57 | # caller (build_prime_context) will catch it inside the broader try/except. |
| 58 | from muse.core.store import ( # noqa: E402 |
| 59 | get_head_commit_id, |
| 60 | read_commit, |
| 61 | read_current_branch, |
| 62 | ) |
| 63 | from muse.plugins.code._query import walk_commits_bfs # noqa: E402 |
| 64 | |
| 65 | #: Schema version for the PrimeContext payload. Bump when the shape changes |
| 66 | #: in a breaking way; consumers should check this to detect schema drift. |
| 67 | PRIME_CONTEXT_SCHEMA_VERSION: str = "1.0.0" |
| 68 | |
| 69 | #: How many of the most-edited notes to include in ``hot_notes``. |
| 70 | _DEFAULT_HOT_NOTE_COUNT: int = 10 |
| 71 | |
| 72 | #: Maximum number of commits to walk when building the context. |
| 73 | _DEFAULT_MAX_COMMITS: int = 100 |
| 74 | |
| 75 | #: Event kinds that count as a "consolidation". |
| 76 | _CONSOLIDATION_KINDS: frozenset[str] = frozenset({"consolidation", "consolidation_pass"}) |
| 77 | |
| 78 | #: Regex for stripping C0/DEL/C1 control characters from strings read from |
| 79 | #: commit records (same policy as :func:`muse.core.validation.sanitize_provenance`). |
| 80 | _CONTROL_CHARS_RE: re.Pattern[str] = re.compile(r"[\x00-\x1f\x7f\x80-\x9f]") |
| 81 | |
| 82 | |
| 83 | def _safe_str(value: Any) -> str: |
| 84 | """Return a control-char-stripped string representation of *value*. |
| 85 | |
| 86 | Args: |
| 87 | value: Any value (typically a str from a commit record field). |
| 88 | |
| 89 | Returns: |
| 90 | String with all C0/DEL/C1 control characters removed. Returns an |
| 91 | empty string for ``None`` inputs. |
| 92 | """ |
| 93 | if value is None: |
| 94 | return "" |
| 95 | return _CONTROL_CHARS_RE.sub("", str(value)) |
| 96 | |
| 97 | |
| 98 | @dataclass(frozen=True) |
| 99 | class ConsolidationRecord: |
| 100 | """Summary of the most recent consolidation commit. |
| 101 | |
| 102 | Attributes: |
| 103 | commit_id: Full commit ID (sha256 hex). |
| 104 | committed_at: ISO-8601 timestamp string. |
| 105 | message: Commit message (control-char stripped). |
| 106 | agent_id: Agent that ran the consolidation (empty when absent). |
| 107 | model_id: Model used for the consolidation (empty when absent). |
| 108 | """ |
| 109 | |
| 110 | commit_id: str |
| 111 | committed_at: str |
| 112 | message: str |
| 113 | agent_id: str |
| 114 | model_id: str |
| 115 | |
| 116 | def to_dict(self) -> dict[str, str]: |
| 117 | """Serialise to a plain dict for JSON embedding. |
| 118 | |
| 119 | Returns: |
| 120 | Dict with all fields as strings. |
| 121 | """ |
| 122 | return { |
| 123 | "commit_id": self.commit_id, |
| 124 | "committed_at": self.committed_at, |
| 125 | "message": self.message, |
| 126 | "agent_id": self.agent_id, |
| 127 | "model_id": self.model_id, |
| 128 | } |
| 129 | |
| 130 | |
| 131 | @dataclass(frozen=True) |
| 132 | class HotNote: |
| 133 | """A frequently-edited note from the recent commit graph. |
| 134 | |
| 135 | Attributes: |
| 136 | path: Vault-relative note path (e.g. ``"notes/session.md"``). |
| 137 | edits: Number of commits in the walk that touched this note. |
| 138 | """ |
| 139 | |
| 140 | path: str |
| 141 | edits: int |
| 142 | |
| 143 | def to_dict(self) -> dict[str, Any]: |
| 144 | """Serialise to a plain dict for JSON embedding. |
| 145 | |
| 146 | Returns: |
| 147 | Dict with ``path`` (str) and ``edits`` (int). |
| 148 | """ |
| 149 | return {"path": self.path, "edits": self.edits} |
| 150 | |
| 151 | |
| 152 | @dataclass(frozen=True) |
| 153 | class PrimeContext: |
| 154 | """Commit-graph-derived bootstrap context for ``knowtation://prime``. |
| 155 | |
| 156 | Attributes: |
| 157 | schema_version: Always :data:`PRIME_CONTEXT_SCHEMA_VERSION`. |
| 158 | commits_scanned: Number of commits examined during the walk. |
| 159 | last_consolidation: Most recent consolidation commit summary, or |
| 160 | ``None`` if no consolidation was found in the |
| 161 | walk window. |
| 162 | hot_notes: Top-N most-edited notes from the walk window, |
| 163 | sorted by edit count descending. |
| 164 | source: ``"muse-commit-graph"`` — identifies this as |
| 165 | commit-graph-derived data so consumers can |
| 166 | distinguish it from the static fallback. |
| 167 | """ |
| 168 | |
| 169 | schema_version: str |
| 170 | commits_scanned: int |
| 171 | last_consolidation: ConsolidationRecord | None |
| 172 | hot_notes: list[HotNote] = field(default_factory=list) |
| 173 | source: str = "muse-commit-graph" |
| 174 | |
| 175 | def to_dict(self) -> dict[str, Any]: |
| 176 | """Serialise to a plain dict suitable for JSON embedding in the prime payload. |
| 177 | |
| 178 | Returns: |
| 179 | Dict with all fields. ``last_consolidation`` is ``null`` (``None``) |
| 180 | when no consolidation was found in the walk window. |
| 181 | """ |
| 182 | return { |
| 183 | "schema_version": self.schema_version, |
| 184 | "source": self.source, |
| 185 | "commits_scanned": self.commits_scanned, |
| 186 | "last_consolidation": ( |
| 187 | self.last_consolidation.to_dict() |
| 188 | if self.last_consolidation is not None |
| 189 | else None |
| 190 | ), |
| 191 | "hot_notes": [n.to_dict() for n in self.hot_notes], |
| 192 | } |
| 193 | |
| 194 | |
| 195 | def _extract_note_paths_from_delta(structured_delta: Any) -> list[str]: |
| 196 | """Extract unique note file paths from a structured delta's op list. |
| 197 | |
| 198 | Iterates the ops, looking for addresses that look like knowtation note |
| 199 | paths (end in ``.md`` or contain ``::section:`` — both indicate the op |
| 200 | belongs to a note file). |
| 201 | |
| 202 | Args: |
| 203 | structured_delta: A :class:`~muse.domain.StructuredDelta` dict or |
| 204 | ``None``. |
| 205 | |
| 206 | Returns: |
| 207 | Deduplicated list of note file paths (vault-relative) mentioned in |
| 208 | the delta. Empty list when *structured_delta* is ``None`` or has no |
| 209 | note-path ops. |
| 210 | """ |
| 211 | if structured_delta is None: |
| 212 | return [] |
| 213 | ops = structured_delta.get("ops", []) |
| 214 | paths: set[str] = set() |
| 215 | for op in ops: |
| 216 | if not isinstance(op, dict): |
| 217 | continue |
| 218 | addr = op.get("address", "") |
| 219 | if not isinstance(addr, str): |
| 220 | continue |
| 221 | # Section-level addresses have the form "notes/file.md::section:…" |
| 222 | if "::" in addr: |
| 223 | path_part = addr.split("::", 1)[0] |
| 224 | if path_part.endswith(".md"): |
| 225 | paths.add(_safe_str(path_part)) |
| 226 | elif addr.endswith(".md"): |
| 227 | paths.add(_safe_str(addr)) |
| 228 | return list(paths) |
| 229 | |
| 230 | |
| 231 | def build_prime_context( |
| 232 | root: Path, |
| 233 | *, |
| 234 | max_commits: int = _DEFAULT_MAX_COMMITS, |
| 235 | hot_note_count: int = _DEFAULT_HOT_NOTE_COUNT, |
| 236 | ) -> PrimeContext: |
| 237 | """Build commit-graph-derived bootstrap context for ``knowtation://prime``. |
| 238 | |
| 239 | Walks the most recent commits on the current branch, extracts: |
| 240 | |
| 241 | * The most recent consolidation commit (``event_type`` in |
| 242 | ``{"consolidation", "consolidation_pass"}``). |
| 243 | * A frequency map of note paths from structured deltas to surface hot notes. |
| 244 | |
| 245 | The walk is bounded by *max_commits* to prevent DoS on long histories. |
| 246 | On any I/O or deserialization error the function returns a |
| 247 | :class:`PrimeContext` with the data collected up to the failure point |
| 248 | (partial results are better than no results for the bootstrap use case). |
| 249 | |
| 250 | Args: |
| 251 | root: Path to the Muse repo root (contains ``.muse/``). |
| 252 | max_commits: Maximum commits to inspect (default 100). |
| 253 | hot_note_count: Number of hot notes to include in the result (default 10). |
| 254 | |
| 255 | Returns: |
| 256 | A :class:`PrimeContext` with schema_version, commits_scanned, |
| 257 | last_consolidation, and hot_notes populated. |
| 258 | """ |
| 259 | try: |
| 260 | branch = read_current_branch(root) |
| 261 | head_id = get_head_commit_id(root, branch) |
| 262 | if head_id is None: |
| 263 | return PrimeContext( |
| 264 | schema_version=PRIME_CONTEXT_SCHEMA_VERSION, |
| 265 | commits_scanned=0, |
| 266 | last_consolidation=None, |
| 267 | hot_notes=[], |
| 268 | ) |
| 269 | |
| 270 | commits, _truncated = walk_commits_bfs(root, head_id, max_commits) |
| 271 | except Exception as exc: |
| 272 | logger.warning("build_prime_context: error reading commit graph: %s", exc) |
| 273 | return PrimeContext( |
| 274 | schema_version=PRIME_CONTEXT_SCHEMA_VERSION, |
| 275 | commits_scanned=0, |
| 276 | last_consolidation=None, |
| 277 | hot_notes=[], |
| 278 | ) |
| 279 | |
| 280 | note_edit_counts: dict[str, int] = {} |
| 281 | last_consolidation: ConsolidationRecord | None = None |
| 282 | |
| 283 | for commit in commits: |
| 284 | # Check for consolidation event type. |
| 285 | event_type = commit.metadata.get("event_type", "") |
| 286 | if last_consolidation is None and event_type in _CONSOLIDATION_KINDS: |
| 287 | last_consolidation = ConsolidationRecord( |
| 288 | commit_id=_safe_str(commit.commit_id), |
| 289 | committed_at=_safe_str(commit.committed_at.isoformat()), |
| 290 | message=_safe_str(commit.message), |
| 291 | agent_id=_safe_str(commit.agent_id), |
| 292 | model_id=_safe_str(commit.model_id), |
| 293 | ) |
| 294 | |
| 295 | # Accumulate note-path edit counts from the structured delta. |
| 296 | for path in _extract_note_paths_from_delta(commit.structured_delta): |
| 297 | note_edit_counts[path] = note_edit_counts.get(path, 0) + 1 |
| 298 | |
| 299 | hot = sorted(note_edit_counts.items(), key=lambda kv: (-kv[1], kv[0])) |
| 300 | hot_notes = [HotNote(path=p, edits=c) for p, c in hot[:hot_note_count]] |
| 301 | |
| 302 | return PrimeContext( |
| 303 | schema_version=PRIME_CONTEXT_SCHEMA_VERSION, |
| 304 | commits_scanned=len(commits), |
| 305 | last_consolidation=last_consolidation, |
| 306 | hot_notes=hot_notes, |
| 307 | ) |
| 308 | |
| 309 | |
| 310 | __all__ = [ |
| 311 | "PRIME_CONTEXT_SCHEMA_VERSION", |
| 312 | "ConsolidationRecord", |
| 313 | "HotNote", |
| 314 | "PrimeContext", |
| 315 | "build_prime_context", |
| 316 | ] |