"""Knowtation domain plugin — vault-native version control for Markdown knowledge bases. This plugin implements :class:`~muse.domain.MuseDomainPlugin` for Knowtation-style Markdown vaults. The unit of change is a *note* (its frontmatter, heading sections, outbound links, entity references, and mist attachment IDs) rather than a line of text. Philosophy ---------- A Knowtation vault is a graph of Markdown documents connected by ``[[wikilinks]]``, typed entity references, and shared mist attachment IDs. Two commits that only reformat whitespace in a note produce no semantic delta. Frontmatter tag-set additions auto-merge without conflicts. Section-level changes are detected at heading granularity rather than line granularity. Live State ---------- ``LiveState`` is either a ``pathlib.Path`` pointing at the vault root or a ``SnapshotManifest`` dict. The path form is used by the CLI; the dict form is used by in-memory merge and diff operations. Snapshot Format --------------- A knowtation snapshot is a ``SnapshotManifest``: .. code-block:: json { "files": { "notes/My Note.md": "", "attachments/diagram.svg": "" }, "domain": "knowtation" } The ``files`` values are **raw-bytes SHA-256 hashes** (not semantic hashes). This guarantees the object store can restore files verbatim on ``muse checkout``. Semantic identity (frontmatter + section hashing) is used only inside ``diff()`` when constructing the structured delta. Binary attachments are *not* tracked by this plugin — they live as ``mist`` artifacts and are referenced from note frontmatter via their 12-char mist IDs (``attachments: [, ...]``). See Phase 1.7. Schema ------ The knowtation domain schema declares six dimensions: ``notes`` The note / directory tree — ``TreeSchema`` with GumTree diff. ``frontmatter`` The YAML frontmatter key → value pairs — ``MapSchema`` with set-valued entries. ``sections`` The ordered Markdown heading sections — ``SequenceSchema`` with Myers diff for readable section-level diffs. ``links`` The outbound link set (wikilinks + Markdown refs) — ``SetSchema``. ``entities`` The entity reference set extracted from frontmatter ``entity`` / ``entities`` keys — ``SetSchema`` with ``by_content`` identity. ``attachments`` The mist attachment-ID set from frontmatter ``attachments`` key — ``SetSchema`` with ``by_content`` identity. Added by Phase 1.7; present in the schema skeleton from Phase 1.1 so downstream consumers can rely on its existence. """ from __future__ import annotations import hashlib import logging import os import pathlib import stat as _stat from muse._version import __version__ from muse.core.attributes import load_attributes, resolve_strategy from muse.core.diff_algorithms import snapshot_diff from muse.core.ignore import is_ignored, load_ignore_config, resolve_patterns from muse.core.snapshot import ( _BUILTIN_SECRET_PATTERNS, directories_from_manifest, load_ignore_patterns, ) from muse.core.stat_cache import load_cache from muse.core.schema import ( DimensionSpec, DomainSchema, MapSchema, SequenceSchema, SetSchema, TreeSchema, ) from muse.domain import ( DeleteOp, DomainOp, DriftReport, InsertOp, LiveState, MergeResult, SnapshotManifest, StateDelta, StateSnapshot, StructuredDelta, ) from muse.core.store import Manifest logger = logging.getLogger(__name__) type _AppliedStrategies = dict[str, str] _DOMAIN_NAME = "knowtation" # Vault directories that are never versioned regardless of .museignore. # Includes standard tool-generated dirs (mirrored from code plugin) plus # vault-native config and cache dirs that should never enter the commit graph. _ALWAYS_IGNORE_DIRS: frozenset[str] = frozenset({ # Version control and Muse internals ".git", ".muse", # Python artefacts "__pycache__", ".mypy_cache", ".pytest_cache", ".ruff_cache", # JS / Node artefacts "node_modules", # Virtual environments ".venv", "venv", ".tox", ".nox", # Coverage and build outputs ".coverage", "htmlcov", "dist", "build", ".eggs", # macOS metadata ".DS_Store", # Vault-native tool directories that must not be versioned ".obsidian", # Obsidian IDE workspace config + cache ".logseq", # LogSeq config and cache ".trash", # Obsidian soft-delete trash folder ".knowtation", # Knowtation vector index and stat cache (runtime data) }) class KnowtationPlugin: """Muse domain plugin for Knowtation Markdown vaults. Implements the six core ``MuseDomainPlugin`` protocol methods. Does not yet implement ``StructuredMergePlugin`` (OT merge) or ``CRDTPlugin`` (convergent join) — those are added in Phases 2 and beyond. The plugin is stateless. The module-level singleton :data:`plugin` is the standard entry point registered in ``muse/plugins/registry.py``. """ # ------------------------------------------------------------------ # 1. snapshot # ------------------------------------------------------------------ def snapshot(self, live_state: LiveState) -> StateSnapshot: """Capture the current vault as a content-addressed manifest. Walks all regular files under *live_state*, hashing each one with SHA-256 (raw bytes). Honours ``.museignore`` and always ignores vault-native tool directories (``".obsidian"``, ``".knowtation"``, etc.) and standard build artefacts. Uses ``os.walk`` with in-place ``dirnames`` pruning so that always- ignored and hidden-from-versioning directories are never descended into. The ``StatCache`` is consulted before hashing so unchanged notes are not re-read from disk. Binary attachment blobs should be stored in mist and referenced via frontmatter ``attachments`` keys. They are tracked here as file-level blobs only if they happen to live inside the vault root; the semantic attachment dimension is populated by the note parser in Phase 1.4. Args: live_state: A ``pathlib.Path`` pointing to the vault root, or an existing ``SnapshotManifest`` dict (returned as-is). Returns: A ``SnapshotManifest`` mapping workspace-relative POSIX paths to their SHA-256 raw-bytes digests, with ``domain="knowtation"``. """ if not isinstance(live_state, pathlib.Path): return live_state workdir = live_state patterns = ( _BUILTIN_SECRET_PATTERNS + resolve_patterns(load_ignore_config(workdir), _DOMAIN_NAME) ) cache = load_cache(workdir) files: Manifest = {} musekeep_dirs: set[str] = set() root_str = str(workdir) prefix_len = len(root_str) + 1 for dirpath, dirnames, filenames in os.walk(root_str, followlinks=False): rel_dir = "" if dirpath != root_str: rel_dir = dirpath[prefix_len:] if os.sep != "/": rel_dir = rel_dir.replace(os.sep, "/") # Prune subdirectories: skip _ALWAYS_IGNORE_DIRS, directories # whose contents are entirely ignored by .museignore, and nested # Muse repos (treated as independent units). dirnames[:] = sorted( d for d in dirnames if d not in _ALWAYS_IGNORE_DIRS and not is_ignored( f"{rel_dir}/{d}/_" if rel_dir else f"{d}/_", patterns ) and not os.path.isdir(os.path.join(dirpath, d, ".muse")) ) for fname in sorted(filenames): abs_str = os.path.join(dirpath, fname) try: st = os.lstat(abs_str) except OSError: continue if not _stat.S_ISREG(st.st_mode): continue rel = abs_str[prefix_len:] if os.sep != "/": rel = rel.replace(os.sep, "/") if is_ignored(rel, patterns): continue files[rel] = cache.get_cached( rel, abs_str, st.st_mtime, st.st_size, st.st_ino ) if fname == ".musekeep" and rel_dir: musekeep_dirs.add(rel_dir) cache.prune(set(files)) cache.save() # Tracked directories: all directories that contain at least one # versioned file, plus any directories containing a .musekeep marker. dirs = sorted(set(directories_from_manifest(files)) | musekeep_dirs) return SnapshotManifest(files=files, domain=_DOMAIN_NAME, directories=dirs) # ------------------------------------------------------------------ # 2. diff # ------------------------------------------------------------------ def diff( self, base: StateSnapshot, target: StateSnapshot, *, repo_root: pathlib.Path | None = None, ) -> StateDelta: """Compute the structured delta between two vault snapshots. Phase 1.1 skeleton: delegates to ``snapshot_diff`` for file-level operations using the declared domain schema. Phase 1.4 will add note-level (frontmatter + section) ``PatchOp`` entries when ``repo_root`` is provided. Args: base: Base snapshot (older state). target: Target snapshot (newer state). repo_root: Repository root for object-store access (unused in Phase 1.1 but wired for the Phase 1.4 parser upgrade). Returns: A ``StructuredDelta`` with ``domain="knowtation"``. """ # Phase 1.1: coarse file-level diff via schema-driven snapshot_diff. # Phase 1.4 will upgrade this to note-level PatchOps with # frontmatter + section child_ops when repo_root is provided. delta = snapshot_diff(self.schema(), base, target) return StructuredDelta( domain=_DOMAIN_NAME, ops=delta.get("ops", []), summary=delta.get("summary", ""), ) # ------------------------------------------------------------------ # 3. merge # ------------------------------------------------------------------ def merge( self, base: StateSnapshot, left: StateSnapshot, right: StateSnapshot, *, repo_root: pathlib.Path | None = None, ) -> MergeResult: """Three-way merge at file granularity, respecting ``.museattributes``. Standard three-way logic augmented by per-path strategy overrides declared in ``.museattributes``. Phase 2.4 will upgrade this to section-level OT merge via ``StructuredMergePlugin.merge_ops``. - Both sides agree → consensus wins (including both deleted). - Only one side changed → take that side. - Both sides changed differently → consult ``.museattributes``: - ``ours`` — take left; remove from conflict list. - ``theirs`` — take right; remove from conflict list. - ``base`` — revert to ancestor; remove from conflict list. - ``union`` — keep all additions; prefer left for conflicts. - ``manual`` — force into conflict list regardless. - ``auto`` — default three-way conflict. Args: base: Common ancestor snapshot. left: Our branch snapshot. right: Their branch snapshot. repo_root: Repository root; when provided, ``.museattributes`` is consulted for per-path strategy overrides. Returns: A ``MergeResult`` with the reconciled snapshot, any file-level conflicts, and ``applied_strategies`` recording which rules fired. """ attrs = load_attributes(repo_root, domain=_DOMAIN_NAME) if repo_root else [] base_files = base["files"] left_files = left["files"] right_files = right["files"] merged: Manifest = dict(base_files) conflicts: list[str] = [] applied_strategies: _AppliedStrategies = {} all_paths = set(base_files) | set(left_files) | set(right_files) for path in sorted(all_paths): b = base_files.get(path) l = left_files.get(path) r = right_files.get(path) if l == r: if l is None: merged.pop(path, None) else: merged[path] = l if attrs and resolve_strategy(attrs, path) == "manual": conflicts.append(path) applied_strategies[path] = "manual" elif b == l: if r is None: merged.pop(path, None) else: merged[path] = r if attrs and resolve_strategy(attrs, path) == "manual": conflicts.append(path) applied_strategies[path] = "manual" elif b == r: if l is None: merged.pop(path, None) else: merged[path] = l if attrs and resolve_strategy(attrs, path) == "manual": conflicts.append(path) applied_strategies[path] = "manual" else: strategy = resolve_strategy(attrs, path) if attrs else "auto" if strategy == "ours": merged[path] = l or b or "" applied_strategies[path] = "ours" elif strategy == "theirs": merged[path] = r or b or "" applied_strategies[path] = "theirs" elif strategy == "base": if b is None: merged.pop(path, None) else: merged[path] = b applied_strategies[path] = "base" elif strategy == "union": merged[path] = l or r or b or "" applied_strategies[path] = "union" elif strategy == "manual": conflicts.append(path) merged[path] = l or r or b or "" applied_strategies[path] = "manual" else: conflicts.append(path) merged[path] = l or r or b or "" return MergeResult( merged=SnapshotManifest( files=merged, domain=_DOMAIN_NAME, directories=directories_from_manifest(merged), ), conflicts=conflicts, applied_strategies=applied_strategies, ) # ------------------------------------------------------------------ # 4. drift # ------------------------------------------------------------------ def drift(self, committed: StateSnapshot, live: LiveState) -> DriftReport: """Report how much the vault has drifted from the last commit. Called by ``muse status``. Takes a snapshot of the current live state and diffs it against the committed snapshot. Args: committed: The last committed snapshot. live: Current live state (vault path or snapshot manifest). Returns: A ``DriftReport`` describing what has changed since the last commit. """ current = self.snapshot(live) delta = self.diff(committed, current) # Filter delete ops for files now covered by .museignore that are # still on disk — they were previously tracked but the user stopped # tracking them. Surfacing them as deleted in status is misleading. if isinstance(live, pathlib.Path) and delta.get("ops"): ignore_patterns = load_ignore_patterns(live) filtered = [ op for op in delta["ops"] if not ( op.get("op") == "delete" and "::" not in op.get("address", "") and is_ignored(op["address"], ignore_patterns) and (live / op["address"]).exists() ) ] if len(filtered) != len(delta["ops"]): delta = StructuredDelta( domain=delta["domain"], ops=filtered, summary=delta.get("summary", ""), ) return DriftReport( has_drift=len(delta.get("ops", [])) > 0, summary=delta.get("summary", ""), delta=delta, ) # ------------------------------------------------------------------ # 5. apply # ------------------------------------------------------------------ def apply(self, delta: StateDelta, live_state: LiveState) -> LiveState: """Apply a delta to the vault. Called by ``muse checkout`` after the core engine has already restored file-level objects from the object store. The knowtation plugin has no domain-specific post-processing in Phase 1.1 — index rebuilding will be added in Phase 3.4. Args: delta: The typed operation list (passed through unchanged). live_state: Current live state (returned unchanged). Returns: *live_state* unchanged. """ return live_state # ------------------------------------------------------------------ # 6. schema # ------------------------------------------------------------------ def schema(self) -> DomainSchema: """Declare the structural schema of the knowtation domain. Returns: A ``DomainSchema`` with six semantic dimensions: ``notes``, ``frontmatter``, ``sections``, ``links``, ``entities``, and ``attachments``. Note: ``merge_mode`` is ``"three_way"`` for Phase 1.1. Phase 2 will upgrade ``sections`` to OT-based merge by implementing ``StructuredMergePlugin.merge_ops``. The schema_version tracks the Muse release so downstream consumers can detect schema migrations. """ return self._schema_cached() # ------------------------------------------------------------------ # 7. conflict_fingerprint — RererePlugin sub-protocol (Phase 2.5) # ------------------------------------------------------------------ def conflict_fingerprint( self, path: str, ours_id: str, theirs_id: str, repo_root: pathlib.Path, ) -> str: """Return a knowtation-aware conflict fingerprint for the :class:`muse.domain.RererePlugin` protocol. Implements the runtime-checkable :class:`muse.domain.RererePlugin` sub-protocol so ``muse rerere`` and ``muse merge`` recognise this plugin as domain-aware and use the knowtation note-fingerprint instead of the default content fingerprint. The protocol passes only the *object IDs* of the conflicting blobs. This method loads the bytes from the local object store and delegates to the bytes-based :func:`muse.plugins.knowtation.rerere.conflict_fingerprint`. The merge-base bytes are not available at the protocol layer (the MERGE_STATE base commit is not threaded through the rerere call sites) so an empty ``base`` is passed — the resulting fingerprint is still stable across cosmetic rewrites because section bodies are sourced from the canonicalised inputs. Loading errors (blob missing from the local store, integrity check failure, oversize) propagate as exceptions; the core engine catches those and falls back to the default content fingerprint (see :func:`muse.core.rerere.compute_fingerprint`). Args: path: Vault-relative POSIX path of the conflicting note (used only for diagnostic logging). ours_id: SHA-256 object ID of the ours blob. theirs_id: SHA-256 object ID of the theirs blob. repo_root: Repository root holding ``.muse/objects/``. Returns: 64-character lowercase hex SHA-256 fingerprint suitable for indexing into ``.muse/rr-cache/``. Raises: FileNotFoundError: When either blob is absent from the local object store. ValueError: When either blob exceeds the 16 MiB cap enforced by the differ. """ from muse.core.object_store import read_object from muse.plugins.knowtation.rerere import conflict_fingerprint ours_bytes = read_object(repo_root, ours_id) if ours_bytes is None: raise FileNotFoundError( f"knowtation rerere: ours blob {ours_id[:8]} for {path!r} " "is not in the local object store" ) theirs_bytes = read_object(repo_root, theirs_id) if theirs_bytes is None: raise FileNotFoundError( f"knowtation rerere: theirs blob {theirs_id[:8]} for {path!r} " "is not in the local object store" ) return conflict_fingerprint(ours_bytes, theirs_bytes, b"") # ------------------------------------------------------------------ # Internal: schema construction # ------------------------------------------------------------------ def _schema_cached(self) -> DomainSchema: """Return the domain schema (separated for readability).""" return DomainSchema( domain=_DOMAIN_NAME, description=( "Vault-native version control for Knowtation Markdown knowledge bases. " "Treats the vault as a structured graph of notes with YAML frontmatter, " "heading sections, wikilinks, entity references, and mist attachment IDs. " "Two commits that only reformat whitespace produce no semantic delta. " "Frontmatter tag additions and outbound link changes are detected at " "key granularity rather than byte level." ), top_level=TreeSchema( kind="tree", node_type="vault", diff_algorithm="gumtree", ), dimensions=[ DimensionSpec( name="notes", description=( "Note / directory tree. Tracks which Markdown files exist " "and how they are organised within the vault folder structure." ), schema=TreeSchema( kind="tree", node_type="note", diff_algorithm="gumtree", ), independent_merge=False, ), DimensionSpec( name="frontmatter", description=( "YAML frontmatter key-value pairs. Tracks the structured " "metadata at the top of each note: title, date, tags, " "project, source_type, source_id, entities, attachments." ), schema=MapSchema( kind="map", key_type="frontmatter_key", value_schema=SetSchema( kind="set", element_type="frontmatter_value", identity="by_content", ), identity="by_key", ), independent_merge=True, ), DimensionSpec( name="sections", description=( "Ordered Markdown heading sections within a note. " "Each section is a (heading_level, heading_text, body_hash) " "tuple. Myers diff produces readable section-level diffs. " "Phase 2.4 upgrades this dimension to OT-based merge." ), schema=SequenceSchema( kind="sequence", element_type="section", identity="by_id", diff_algorithm="myers", alphabet=None, ), independent_merge=True, ), DimensionSpec( name="links", description=( "Outbound link set: wikilinks (``[[Target]]``), typed " "wikilinks (``[[Type::Target]]``), and Markdown hrefs " "(``[text](other-note.md)``). Order is semantically " "irrelevant; tracked as an unordered set." ), schema=SetSchema( kind="set", element_type="link", identity="by_content", ), independent_merge=True, ), DimensionSpec( name="entities", description=( "Entity reference set from frontmatter ``entity`` / " "``entities`` keys. Entity handles are the primary " "hook for knowledge-graph queries and note clustering." ), schema=SetSchema( kind="set", element_type="entity_ref", identity="by_content", ), independent_merge=True, ), DimensionSpec( name="attachments", description=( "Mist attachment-ID set from frontmatter ``attachments`` " "key. Each ID is the 12-char base58 SHA-256 prefix " "assigned by the mist domain when the original binary " "blob (PDF, audio, image) was pushed. Phase 1.7 wires " "the import pipeline to populate this field automatically." ), schema=SetSchema( kind="set", element_type="mist_id", identity="by_content", ), independent_merge=True, ), ], merge_mode="three_way", schema_version=__version__, ) # --------------------------------------------------------------------------- # Module-level singleton # --------------------------------------------------------------------------- #: The singleton plugin instance registered in ``muse/plugins/registry.py``. plugin = KnowtationPlugin()