"""Hierarchical vault snapshot manifests for the Muse knowtation domain. A vault manifest organises a snapshot's notes into a two-level hierarchy:: ProjectManifest ← top-level project grouping (by frontmatter ``project`` key) └─ NoteEntry ← one Markdown note's metadata + hashes This structure enables *partial re-parsing*: when merging, only notes whose ``content_hash`` changed need to be re-parsed by the note engine. Notes in unchanged projects are reused from the cached manifest, making three-way merges on large vaults significantly faster. The ``frontmatter_hash`` in :class:`NoteEntry` is a SHA-256 of the note's *frontmatter JSON* rather than its raw bytes. Two notes that differ only in whitespace inside the body but have identical frontmatter will have the same ``frontmatter_hash``. Public API ---------- - :class:`NoteEntry` — per-note metadata. - :class:`ProjectManifest` — project-level grouping. - :class:`VaultManifest` — complete hierarchical vault manifest. - :func:`build_vault_manifest` — build from a flat snapshot manifest. - :func:`diff_vault_manifests` — find added/removed/modified notes between two. - :func:`write_vault_manifest` — persist to ``.muse/knowtation_manifests/.json``. - :func:`read_vault_manifest` — load from disk. """ from __future__ import annotations import hashlib import json import logging import pathlib from typing import TypedDict from muse.core._types import Manifest logger = logging.getLogger(__name__) type _ProjectNoteMap = dict[str, list[str]] # project → [note_path, ...] type _NoteEntryMap = dict[str, "NoteEntry"] # note_path → NoteEntry # --------------------------------------------------------------------------- # Data types # --------------------------------------------------------------------------- class NoteEntry(TypedDict): """Metadata for a single Markdown note. ``path`` Workspace-relative POSIX path. ``content_hash`` SHA-256 of the raw note bytes (from the object store). ``frontmatter_hash`` SHA-256 of the note's YAML frontmatter JSON. Empty string when the note has no frontmatter. ``project`` Value of the ``project`` frontmatter key, or ``""`` when absent. ``note_type`` File extension classification (``"markdown"`` or ``"other"``). ``section_count`` Number of Markdown headings (H1–H6) in the note. 0 for non-Markdown files or parse errors. ``size_bytes`` Raw file size in bytes (0 if unavailable). """ path: str content_hash: str frontmatter_hash: str project: str note_type: str section_count: int size_bytes: int class ProjectManifest(TypedDict): """Manifest for one project grouping of notes. ``project`` Project name (value of frontmatter ``project`` key, or ``"__root__"`` for notes without a project). ``project_hash`` SHA-256 of all sorted ``content_hash`` values in this project — stable fingerprint for change detection. ``notes`` All :class:`NoteEntry` records in this project. ``total_notes`` Total number of notes tracked in this project. ``markdown_notes`` Number of notes with a ``.md`` extension. """ project: str project_hash: str notes: list[NoteEntry] total_notes: int markdown_notes: int class VaultManifest(TypedDict): """Complete hierarchical manifest for one vault snapshot. ``snapshot_id`` The snapshot this manifest was built from. ``manifest_hash`` SHA-256 of this manifest's JSON — stable cache key. ``projects`` All :class:`ProjectManifest` entries, sorted by project name. ``total_notes`` Total notes across all projects. ``markdown_notes`` Notes with ``.md`` extension. ``total_sections`` Sum of ``section_count`` across all notes. """ snapshot_id: str manifest_hash: str projects: list[ProjectManifest] total_notes: int markdown_notes: int total_sections: int # --------------------------------------------------------------------------- # Note type classification # --------------------------------------------------------------------------- _NOTE_SUFFIXES: frozenset[str] = frozenset({".md", ".markdown", ".mdx"}) def _note_type_of(file_path: str) -> str: """Return the note-type label for *file_path* based on its extension.""" suffix = pathlib.PurePosixPath(file_path).suffix.lower() return "markdown" if suffix in _NOTE_SUFFIXES else "other" def _is_markdown(file_path: str) -> bool: """Return ``True`` if *file_path* has a Markdown extension.""" suffix = pathlib.PurePosixPath(file_path).suffix.lower() return suffix in _NOTE_SUFFIXES # --------------------------------------------------------------------------- # Frontmatter and section parsing (lightweight, no external deps) # --------------------------------------------------------------------------- def _parse_frontmatter_hash(content: bytes) -> tuple[str, str]: """Extract and hash the YAML frontmatter block from *content*. Returns a ``(frontmatter_hash, project)`` tuple. When no frontmatter is present, both values are empty strings. Parsing is intentionally minimal at Phase 1.1 — a proper typed ``FrontmatterSchema`` will be introduced in Phase 1.3. The hash is computed over the raw YAML text (not parsed values) so that semantically identical frontmatter with different whitespace produces different hashes — acceptable for the manifest cache key until Phase 1.3 introduces normalised hashing. """ try: text = content.decode("utf-8", errors="replace") except Exception: return "", "" if not text.startswith("---"): return "", "" # Find the closing "---" marker on its own line. rest = text[3:] end = rest.find("\n---") if end == -1: return "", "" raw_fm = rest[:end].strip() fm_hash = hashlib.sha256(raw_fm.encode()).hexdigest() # Extract ``project`` value — simple key: value scan (no full YAML parse). project = "" for line in raw_fm.splitlines(): stripped = line.strip() if stripped.startswith("project:"): value = stripped[len("project:"):].strip().strip("\"'") project = value break return fm_hash, project def _count_sections(content: bytes) -> int: """Count Markdown headings (# through ######) in *content*. Returns 0 when the note has no headings or when *content* cannot be decoded as UTF-8. """ try: text = content.decode("utf-8", errors="replace") except Exception: return 0 # Skip the frontmatter block before counting headings. body = text if text.startswith("---"): rest = text[3:] end = rest.find("\n---") if end != -1: body = rest[end + 4:] count = 0 for line in body.splitlines(): if line.startswith("#"): # Accept # through ###### at the start of a line (ATX headings). stripped = line.lstrip("#") if stripped and (stripped[0] == " " or stripped[0] == "\t"): count += 1 return count # --------------------------------------------------------------------------- # Builder # --------------------------------------------------------------------------- def build_vault_manifest( snapshot_id: str, flat_manifest: Manifest, repo_root: pathlib.Path, ) -> VaultManifest: """Build a :class:`VaultManifest` from a flat ``{path: content_hash}`` dict. Attempts to parse frontmatter and count sections for Markdown files. Falls back gracefully for binary files or parse errors. All blob reads go through the Muse object store so only committed content is parsed. Args: snapshot_id: The snapshot this manifest represents. flat_manifest: ``{workspace_path: sha256}`` from the snapshot. repo_root: Repository root for object store access. Returns: A fully populated :class:`VaultManifest`. """ from muse.core.object_store import read_object # Group notes by project (frontmatter ``project`` key or ``"__root__"``). entries: list[NoteEntry] = [] total_sections = 0 total_markdown = 0 for file_path in sorted(flat_manifest): content_hash = flat_manifest[file_path] note_type = _note_type_of(file_path) fm_hash = "" project = "__root__" section_count = 0 if note_type == "markdown": source = read_object(repo_root, content_hash) if source is not None: try: fm_hash, raw_project = _parse_frontmatter_hash(source) project = raw_project or "__root__" section_count = _count_sections(source) except Exception: logger.debug("Note parse failed for %s", file_path) total_markdown += 1 total_sections += section_count # size_bytes requires reading the blob; use 0 as a safe default when # the object is not locally cached. Phase 1.4 will hydrate this field. entries.append(NoteEntry( path=file_path, content_hash=content_hash, frontmatter_hash=fm_hash, project=project, note_type=note_type, section_count=section_count, size_bytes=0, )) # Group entries by project. project_map: _ProjectNoteMap = {} entry_map: _NoteEntryMap = {e["path"]: e for e in entries} for entry in entries: project_map.setdefault(entry["project"], []).append(entry["path"]) projects: list[ProjectManifest] = [] for proj_name in sorted(project_map): note_paths = sorted(project_map[proj_name]) proj_notes = [entry_map[p] for p in note_paths] proj_hashes = [e["content_hash"] for e in proj_notes] proj_hash = hashlib.sha256( "|".join(sorted(proj_hashes)).encode() ).hexdigest() md_count = sum(1 for n in proj_notes if n["note_type"] == "markdown") projects.append(ProjectManifest( project=proj_name, project_hash=proj_hash, notes=proj_notes, total_notes=len(proj_notes), markdown_notes=md_count, )) manifest_json = json.dumps( {"snapshot_id": snapshot_id, "projects": projects}, sort_keys=True ) manifest_hash = hashlib.sha256(manifest_json.encode()).hexdigest() return VaultManifest( snapshot_id=snapshot_id, manifest_hash=manifest_hash, projects=projects, total_notes=len(flat_manifest), markdown_notes=total_markdown, total_sections=total_sections, ) # --------------------------------------------------------------------------- # Diff # --------------------------------------------------------------------------- class NoteFileDiff(TypedDict): """Change record from :func:`diff_vault_manifests` for one note.""" path: str change: str # "added" | "removed" | "modified" | "frontmatter_changed" old_hash: str new_hash: str old_fm_hash: str new_fm_hash: str frontmatter_change: bool # True when fm_hash differs (frontmatter mutated) def diff_vault_manifests( base: VaultManifest, target: VaultManifest, ) -> list[NoteFileDiff]: """Produce a per-note change list between two :class:`VaultManifest` objects. Notes with identical ``content_hash`` values are skipped (no change). Notes where only the ``content_hash`` changed but ``frontmatter_hash`` is the same are marked ``"modified"`` with ``frontmatter_change=False``. Notes where ``frontmatter_hash`` changed are ``"frontmatter_changed"`` with ``frontmatter_change=True``. Args: base: Manifest for the earlier state. target: Manifest for the later state. Returns: Sorted list of :class:`NoteFileDiff` records. """ base_notes: _NoteEntryMap = {} for proj in base["projects"]: for note in proj["notes"]: base_notes[note["path"]] = note target_notes: _NoteEntryMap = {} for proj in target["projects"]: for note in proj["notes"]: target_notes[note["path"]] = note diffs: list[NoteFileDiff] = [] all_paths = sorted(set(base_notes) | set(target_notes)) for path in all_paths: bn = base_notes.get(path) tn = target_notes.get(path) if bn is None and tn is not None: diffs.append(NoteFileDiff( path=path, change="added", old_hash="", new_hash=tn["content_hash"], old_fm_hash="", new_fm_hash=tn["frontmatter_hash"], frontmatter_change=True, )) elif bn is not None and tn is None: diffs.append(NoteFileDiff( path=path, change="removed", old_hash=bn["content_hash"], new_hash="", old_fm_hash=bn["frontmatter_hash"], new_fm_hash="", frontmatter_change=True, )) elif bn is not None and tn is not None: if bn["content_hash"] == tn["content_hash"]: continue # No change. fm_changed = bn["frontmatter_hash"] != tn["frontmatter_hash"] diffs.append(NoteFileDiff( path=path, change="frontmatter_changed" if fm_changed else "modified", old_hash=bn["content_hash"], new_hash=tn["content_hash"], old_fm_hash=bn["frontmatter_hash"], new_fm_hash=tn["frontmatter_hash"], frontmatter_change=fm_changed, )) return diffs # --------------------------------------------------------------------------- # Persistence # --------------------------------------------------------------------------- def write_vault_manifest( repo_root: pathlib.Path, manifest: VaultManifest ) -> None: """Persist a :class:`VaultManifest` to ``.muse/knowtation_manifests/.json``. Creates the directory if absent. Skips writing when the file already exists (content-addressed store is immutable). Args: repo_root: Repository root. manifest: The manifest to write. """ store_dir = repo_root / ".muse" / "knowtation_manifests" store_dir.mkdir(parents=True, exist_ok=True) path = store_dir / f"{manifest['manifest_hash']}.json" if not path.exists(): path.write_text(json.dumps(manifest), encoding="utf-8") def read_vault_manifest( repo_root: pathlib.Path, manifest_hash: str ) -> VaultManifest | None: """Load a :class:`VaultManifest` by its hash. Args: repo_root: Repository root. manifest_hash: The ``manifest_hash`` of the target manifest. Returns: The deserialized :class:`VaultManifest`, or ``None`` if not found. """ path = ( repo_root / ".muse" / "knowtation_manifests" / f"{manifest_hash}.json" ) if not path.exists(): return None raw: VaultManifest = json.loads(path.read_text(encoding="utf-8")) return raw