"""Knowtation domain rerere plugin — Phase 2.5. Conflict fingerprinting and replay engine that remembers how note conflicts were resolved and automatically reapplies resolutions when the same logical conflict recurs under cosmetic note rewrites. Design overview --------------- Two conflicts are *the same* when the **logical content** of the conflicting sections matches, even when the surrounding note has been reformatted — whitespace tweaks, frontmatter tag additions, or section reorders outside the conflict zone. This is achieved by: 1. **Canonicalising** every input note through :func:`muse.plugins.knowtation.differ.canonicalize` so YAML key order, line endings, and frontmatter dump style are absorbed. 2. Splitting the canonical body into Markdown heading sections via :func:`muse.plugins.knowtation.symbols.split_sections` and assigning each section a stable, occurrence-disambiguated ID. 3. Computing the conflict tuple set ``(section_id, sha256(ours_body), sha256(theirs_body), sha256(base_body))`` only for sections that exist in both ours and theirs **and** whose bodies differ. 4. Sorting the tuples by ``section_id`` and folding them into a single SHA-256 digest with a length-prefixed binary encoding so that no section title can smuggle a separator character through the fingerprint. Storage layout -------------- Resolutions live under ``/.muse/rerere/`` keyed by fingerprint: * ``/.muse/rerere//.json`` — hash-only resolution record consumed by :func:`replay_resolution`. * ``/.muse/rerere//.full`` — full canonical bytes of the resolved note consumed by :func:`replay_full_resolution`. The two-character shard prevents directory blow-up on vaults with thousands of recorded resolutions. Both files are written atomically via ``tempfile.mkstemp`` + :func:`os.replace`. Security model -------------- * :data:`_MAX_NOTE_BYTES` (16 MiB) hard-caps every byte input and is enforced **before** any parsing — defends against runaway YAML and pathological splitter inputs. Inherits the cap defined by :mod:`muse.plugins.knowtation.differ`. * Fingerprints are validated against :data:`_FINGERPRINT_RE` before any filesystem access — path traversal via ``"../"`` or ``/`` is rejected with :class:`ValueError` rather than silently writing outside the rerere directory. * The canonical fingerprint encoding is **length-prefixed binary**: the section ID is preceded by its UTF-8 byte length packed as a 4-byte big-endian integer, and section body hashes are appended as their raw 32-byte digest (decoded from hex). This makes ``|`` / ``:`` injection impossible — a hostile section title cannot create or merge conflict tuples. * Stored resolution JSON uses :func:`json.loads` only; no arbitrary code execution is possible during lookup. * Full-resolution storage is capped at :data:`_MAX_NOTE_BYTES` on read so a corrupted store cannot trigger an OOM during replay. Plugin integration ------------------ This module exposes :class:`KnowtationRererePlugin`, a standalone helper class with the bytes-based interface described in the Phase 2.5 spec. The class is also exported as the :data:`plugin` module-level singleton. The runtime-checkable :class:`muse.domain.RererePlugin` protocol used by ``muse rerere`` is satisfied by :class:`muse.plugins.knowtation.plugin.KnowtationPlugin`, which loads the ours / theirs blobs from the local object store and delegates to this module's :func:`conflict_fingerprint`. See ``KnowtationPlugin.conflict_fingerprint``. """ from __future__ import annotations import datetime import hashlib import json import logging import os import pathlib import re import tempfile from typing import Final from muse.plugins.knowtation.differ import canonicalize from muse.plugins.knowtation.parser import parse_frontmatter from muse.plugins.knowtation.symbols import _strip_frontmatter, split_sections logger = logging.getLogger(__name__) # ────────────────────────────────────────────────────────────────────────────── # Constants # ────────────────────────────────────────────────────────────────────────────── #: Domain slug — matches ``KnowtationPlugin``'s ``_DOMAIN_NAME``. DOMAIN: Final[str] = "knowtation" #: 16 MiB — must mirror the cap in :mod:`muse.plugins.knowtation.differ`. #: Inputs larger than this are rejected with :class:`ValueError` *before* #: any parsing to defend against pathological YAML or splitter inputs. _MAX_NOTE_BYTES: Final[int] = 16 * 1024 * 1024 #: Maximum bytes accepted when reading a stored ``.full`` resolution back #: from disk. Equal to the input cap so a stored resolution cannot exceed #: the size we would have accepted on input. _MAX_FULL_RESOLUTION_BYTES: Final[int] = _MAX_NOTE_BYTES #: Maximum bytes accepted when reading a stored ``.json`` resolution-hash #: record back from disk. The hash record is tiny (under 200 bytes); we #: pick 4 KiB to leave room for future metadata without enabling abuse. _MAX_HASH_RECORD_BYTES: Final[int] = 4096 #: Strict 64-char lowercase hex regex for fingerprint validation. Rejects #: any string containing ``"/"``, ``".."``, or whitespace before it can #: reach :func:`pathlib.Path` and trigger a path-traversal or symlink walk. _FINGERPRINT_RE: Final[re.Pattern[str]] = re.compile(r"^[0-9a-f]{64}$") #: Subdirectory under ``.muse`` where rerere resolutions are stored. This #: is *separate* from the core ``.muse/rr-cache/`` used by #: :mod:`muse.core.rerere` so the two layers do not interfere. _RERERE_DIR: Final[str] = "rerere" #: Suffix for the JSON hash-only resolution record. _HASH_SUFFIX: Final[str] = ".json" #: Suffix for the full canonical bytes resolution. _FULL_SUFFIX: Final[str] = ".full" # ────────────────────────────────────────────────────────────────────────────── # Note parsing — section extraction with stable IDs # ────────────────────────────────────────────────────────────────────────────── def _check_size(label: str, content: bytes) -> None: """Raise :class:`ValueError` if *content* exceeds :data:`_MAX_NOTE_BYTES`. Called *before* any parsing of byte input so an oversized adversarial note never reaches the YAML or section parser. """ if len(content) > _MAX_NOTE_BYTES: raise ValueError( f"{label} size {len(content)} bytes exceeds maximum " f"{_MAX_NOTE_BYTES} bytes" ) def _section_id(level: int, title: str, occurrence: int) -> str: """Return the occurrence-disambiguated stable section ID. Mirrors the convention used by :mod:`muse.plugins.knowtation.differ` and the merger so the same logical section produces the same ID across independent code paths. Args: level: Heading level (``0`` for preamble, ``1``–``6`` for H1–H6). title: Heading text without ``#`` characters. occurrence: 0-based count among sections sharing ``(level, title)``. Returns: Address string, e.g. ``"section:2:Background#0"``. """ return f"section:{level}:{title}#{occurrence}" def _normalize_section_body(section_body: str) -> bytes: """Return a position-independent canonical form of a section body. The :func:`split_sections` algorithm includes the heading line plus every line up to (but not including) the next heading. When a section is reordered within the note its trailing whitespace therefore varies — a section followed by another heading retains a trailing blank line, while the final section retains only a single ``\\n``. Both shapes carry the *same logical content*, so we strip trailing whitespace before hashing to make the fingerprint stable across non-conflicting section reorders. Internal blank lines are preserved verbatim — only the last block of whitespace at the end of the section is normalised. The result is encoded to UTF-8 so the caller can hash it directly. Args: section_body: Raw section text from :func:`split_sections`. Returns: UTF-8 bytes of the section text with trailing whitespace removed and a single ``\\n`` appended (so a body followed by another heading and a body that is the last in the note both end the same way). """ return (section_body.rstrip() + "\n").encode("utf-8") def _section_map(content: bytes) -> dict[str, bytes]: """Parse *content* and return ``{section_id: normalised_body_bytes}``. The note is canonicalised first so cosmetic rewrites — CRLF vs LF, YAML key reorderings, alternative quote styles, frontmatter tag additions — do not change the section bodies hashed downstream. Each section body is then run through :func:`_normalize_section_body` so trailing-whitespace differences introduced by the splitter when sections are reordered cannot perturb the fingerprint. The frontmatter block itself is **not** included in the map; fingerprinting deliberately ignores frontmatter so a tag addition that leaves every section body untouched does not perturb the fingerprint. Args: content: Raw note bytes (any size up to :data:`_MAX_NOTE_BYTES`). Returns: Mapping from stable section ID to that section's normalised body bytes. Empty when the note has no sections. Raises: ValueError: When *content* exceeds :data:`_MAX_NOTE_BYTES`. """ _check_size("note", content) canonical = canonicalize(content) text = canonical.decode("utf-8", errors="replace") body, _ = _strip_frontmatter(text) sections = split_sections(body) result: dict[str, bytes] = {} seen: dict[tuple[int, str], int] = {} for level, title, section_body, _line in sections: occurrence = seen.get((level, title), 0) seen[(level, title)] = occurrence + 1 sid = _section_id(level, title, occurrence) result[sid] = _normalize_section_body(section_body) # Touch parser to ensure the canonical frontmatter is well-formed. # We deliberately discard the result — frontmatter does not enter the # fingerprint by design. This call also exercises the YAML safe_load # path so a malformed frontmatter does not propagate further. parse_frontmatter(canonical) return result # ────────────────────────────────────────────────────────────────────────────── # Public fingerprint # ────────────────────────────────────────────────────────────────────────────── def _hash_bytes(data: bytes) -> str: """Return SHA-256 hex digest of *data* (UTF-8 hex, lowercase).""" return hashlib.sha256(data).hexdigest() def conflict_fingerprint(ours: bytes, theirs: bytes, base: bytes) -> str: """Return a stable SHA-256 fingerprint identifying this conflict. The fingerprint is computed over the **set of conflicting sections** only — sections present in both *ours* and *theirs* whose canonical bodies differ. Each conflicting section contributes a tuple ``(section_id, sha256(ours_body), sha256(theirs_body), sha256(base_body))``; tuples are sorted by ``section_id`` and folded into a single SHA-256 digest with a length-prefixed binary encoding that makes separator- injection attacks impossible. Stability properties (proof in module docstring): * Cosmetic frontmatter changes (key order, tag additions, quote style) cannot affect the fingerprint because frontmatter is excluded from the section map and the bodies are taken from canonicalised input. * Section reorders outside the conflict zone cannot affect the fingerprint because tuples are sorted by ``section_id``. * CRLF / LF differences cannot affect the fingerprint because :func:`canonicalize` rewrites every line ending to LF before sectioning. Commutativity: this function is **not** commutative in ``ours`` and ``theirs`` — swapping them produces a different fingerprint. This is intentional: the resolution recorded for ``(ours, theirs)`` is a different artefact than the resolution recorded for the mirrored conflict. Callers that need commutativity should sort the inputs by ``sha256(canonicalize(side))`` before calling. Args: ours: Raw bytes of the ours-side note. theirs: Raw bytes of the theirs-side note. base: Raw bytes of the merge-base note (empty bytes when unavailable — common for the protocol-level call where only ours and theirs are known). Returns: 64-character lowercase hexadecimal SHA-256 digest. Raises: ValueError: When any input exceeds :data:`_MAX_NOTE_BYTES`. """ ours_sections = _section_map(ours) theirs_sections = _section_map(theirs) base_sections = _section_map(base) if base else {} conflict_sids = sorted( sid for sid in ours_sections if sid in theirs_sections and ours_sections[sid] != theirs_sections[sid] ) h = hashlib.sha256() for sid in conflict_sids: ours_body = ours_sections[sid] theirs_body = theirs_sections[sid] base_body = base_sections.get(sid, b"") sid_bytes = sid.encode("utf-8") # Length-prefixed binary encoding — unconditionally injection-proof. # The 4-byte big-endian length tells us *exactly* how many bytes of # section_id follow, so a malicious title containing "|" or ":" or # "::" cannot merge into a neighbouring tuple. Hash bytes are a # fixed-size 32-byte digest, no separator required. h.update(len(sid_bytes).to_bytes(4, "big")) h.update(sid_bytes) h.update(hashlib.sha256(ours_body).digest()) h.update(hashlib.sha256(theirs_body).digest()) h.update(hashlib.sha256(base_body).digest()) return h.hexdigest() # ────────────────────────────────────────────────────────────────────────────── # Storage primitives # ────────────────────────────────────────────────────────────────────────────── def _validate_fingerprint(fingerprint: str) -> None: """Raise :class:`ValueError` unless *fingerprint* is 64 lowercase hex chars. Defends the rerere directory against path traversal: a fingerprint containing ``"/"`` or ``".."`` would let a caller write outside the intended store, so the validator runs *before* any path construction. """ if not isinstance(fingerprint, str) or not _FINGERPRINT_RE.match(fingerprint): raise ValueError( f"Invalid knowtation rerere fingerprint {fingerprint!r} — " "expected exactly 64 lowercase hex characters." ) def _shard_dir(root: pathlib.Path, fingerprint: str) -> pathlib.Path: """Return the per-fingerprint shard directory under ``.muse/rerere/``. The directory is sharded by the fingerprint's first two hex chars to keep any single directory under ~256 children even when thousands of resolutions accumulate. Re-validates the fingerprint on every call so the directory cannot escape the rerere root. """ _validate_fingerprint(fingerprint) return root / ".muse" / _RERERE_DIR / fingerprint[:2] def _resolution_path(root: pathlib.Path, fingerprint: str, suffix: str) -> pathlib.Path: """Return the file path for *fingerprint* + *suffix* (``.json`` / ``.full``).""" return _shard_dir(root, fingerprint) / f"{fingerprint[2:]}{suffix}" def _write_atomic(dest: pathlib.Path, content: bytes) -> None: """Write *content* to *dest* atomically (temp file + rename). Uses :func:`tempfile.mkstemp` in *dest*'s parent directory so the rename is a single ``os.replace`` system call — atomic on every POSIX platform and on Windows since 1607. Concurrent readers see either the old file or the complete new file, never a partial write. """ dest.parent.mkdir(parents=True, exist_ok=True) fd, tmp_str = tempfile.mkstemp(dir=dest.parent, prefix=".rr-tmp-") tmp = pathlib.Path(tmp_str) try: with os.fdopen(fd, "wb") as fh: fh.write(content) os.replace(tmp, dest) except Exception: tmp.unlink(missing_ok=True) raise # ────────────────────────────────────────────────────────────────────────────── # Hash-only resolution: record / lookup / replay # ────────────────────────────────────────────────────────────────────────────── def record_resolution( root: pathlib.Path, fingerprint: str, resolved: bytes, ) -> None: """Record a hash-only resolution for *fingerprint*. Stores ``{"fingerprint": fp, "resolved_hash": sha256(canonicalize(resolved)), "recorded_at": iso8601}`` at ``/.muse/rerere//.json``. The full resolved bytes are *not* stored — see :func:`record_full_resolution` for the bytes-preserving variant. The hash-only record is sufficient for the conservative replay path (:func:`replay_resolution`) which only confirms whether the *current* note already matches a known good resolution. Args: root: Repository root (parent of ``.muse/``). fingerprint: 64-char hex fingerprint produced by :func:`conflict_fingerprint`. resolved: Raw bytes of the resolved note. Raises: ValueError: When *fingerprint* fails :func:`_validate_fingerprint` or *resolved* exceeds :data:`_MAX_NOTE_BYTES`. """ _validate_fingerprint(fingerprint) _check_size("resolved note", resolved) canonical = canonicalize(resolved) payload = { "fingerprint": fingerprint, "resolved_hash": _hash_bytes(canonical), "recorded_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } dest = _resolution_path(root, fingerprint, _HASH_SUFFIX) _write_atomic(dest, json.dumps(payload, sort_keys=True, indent=2).encode("utf-8")) logger.debug( "knowtation rerere: recorded hash-only resolution %s (resolved_hash=%s)", fingerprint[:8], payload["resolved_hash"][:8], ) def lookup_resolution(root: pathlib.Path, fingerprint: str) -> str | None: """Return ``resolved_hash`` for *fingerprint* or ``None`` when absent. Read-only — never mutates the store. Returns ``None`` when the fingerprint is not recorded, when the JSON record is malformed, or when the record exceeds :data:`_MAX_HASH_RECORD_BYTES`. Args: root: Repository root. fingerprint: 64-char hex fingerprint. Returns: The stored ``resolved_hash`` (64-char hex) or ``None``. Raises: ValueError: When *fingerprint* fails :func:`_validate_fingerprint`. """ _validate_fingerprint(fingerprint) path = _resolution_path(root, fingerprint, _HASH_SUFFIX) if not path.exists(): return None try: size = path.stat().st_size except OSError as exc: logger.warning( "⚠️ knowtation rerere: stat failed on %s: %s", path, exc ) return None if size > _MAX_HASH_RECORD_BYTES: logger.warning( "⚠️ knowtation rerere: resolution record %s is %d bytes — " "exceeds %d-byte cap; treating as missing", fingerprint[:8], size, _MAX_HASH_RECORD_BYTES, ) return None try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: logger.warning( "⚠️ knowtation rerere: failed to read resolution record %s: %s", fingerprint[:8], exc, ) return None if not isinstance(data, dict): return None resolved_hash = data.get("resolved_hash") if not isinstance(resolved_hash, str) or not _FINGERPRINT_RE.match(resolved_hash): # Validate the stored hash too — corrupted disk content must not # silently leak through as a "resolution". return None return resolved_hash def replay_resolution( root: pathlib.Path, fingerprint: str, current: bytes, ) -> bytes | None: """Conservatively replay a hash-only resolution. Looks up the stored ``resolved_hash`` for *fingerprint*; if ``sha256(canonicalize(current)) == resolved_hash`` returns *current* (the note already matches the recorded resolution). Otherwise returns ``None`` because the hash-only record cannot reconstruct the resolved bytes from the hash alone. For full byte-level replay use :func:`replay_full_resolution`. Args: root: Repository root. fingerprint: 64-char hex fingerprint. current: Current note bytes to be checked against the record. Returns: *current* (unchanged) when its canonical SHA-256 matches the recorded ``resolved_hash``; ``None`` otherwise. Raises: ValueError: When *fingerprint* fails :func:`_validate_fingerprint` or *current* exceeds :data:`_MAX_NOTE_BYTES`. """ _check_size("current note", current) expected_hash = lookup_resolution(root, fingerprint) if expected_hash is None: return None actual_hash = _hash_bytes(canonicalize(current)) if actual_hash == expected_hash: return current return None # ────────────────────────────────────────────────────────────────────────────── # Full-bytes resolution: record / replay # ────────────────────────────────────────────────────────────────────────────── def record_full_resolution( root: pathlib.Path, fingerprint: str, resolved: bytes, ) -> None: """Record the full canonical bytes of a resolution for *fingerprint*. Persists :func:`canonicalize` of *resolved* at ``/.muse/rerere//.full`` so that :func:`replay_full_resolution` can return the resolved bytes directly without needing the user to re-resolve. This is the variant the rerere CLI uses when auto-resolving a recurring conflict. Args: root: Repository root. fingerprint: 64-char hex fingerprint. resolved: Raw resolved-note bytes. Raises: ValueError: When *fingerprint* fails :func:`_validate_fingerprint` or *resolved* exceeds :data:`_MAX_NOTE_BYTES`. """ _validate_fingerprint(fingerprint) _check_size("resolved note", resolved) canonical = canonicalize(resolved) dest = _resolution_path(root, fingerprint, _FULL_SUFFIX) _write_atomic(dest, canonical) logger.debug( "knowtation rerere: recorded full resolution %s (%d bytes canonical)", fingerprint[:8], len(canonical), ) def replay_full_resolution( root: pathlib.Path, fingerprint: str, current: bytes, ) -> bytes | None: """Return the previously recorded full-bytes resolution, or ``None``. Reads ``/.muse/rerere//.full`` and returns its bytes verbatim (the canonicalised resolved note). Returns ``None`` when the file is absent, unreadable, or exceeds :data:`_MAX_FULL_RESOLUTION_BYTES`. The *current* parameter is accepted for API parity with :func:`replay_resolution` and as a hook for future evolution (e.g. a three-way merge against the stored resolution). In Phase 2.5 it is deliberately unused: byte-perfect replay always returns the recorded canonical resolution regardless of what *current* contains. Args: root: Repository root. fingerprint: 64-char hex fingerprint. current: Current note bytes (size-checked but otherwise unused — see note above). Returns: Canonical resolved-note bytes from the store, or ``None`` when no full resolution is recorded for *fingerprint*. Raises: ValueError: When *fingerprint* fails :func:`_validate_fingerprint` or *current* exceeds :data:`_MAX_NOTE_BYTES`. """ _check_size("current note", current) _validate_fingerprint(fingerprint) path = _resolution_path(root, fingerprint, _FULL_SUFFIX) if not path.exists(): return None try: size = path.stat().st_size except OSError as exc: logger.warning( "⚠️ knowtation rerere: stat failed on %s: %s", path, exc ) return None if size > _MAX_FULL_RESOLUTION_BYTES: logger.warning( "⚠️ knowtation rerere: full resolution %s is %d bytes — " "exceeds %d-byte cap; refusing to replay", fingerprint[:8], size, _MAX_FULL_RESOLUTION_BYTES, ) return None try: return path.read_bytes() except OSError as exc: logger.warning( "⚠️ knowtation rerere: failed to read full resolution %s: %s", fingerprint[:8], exc, ) return None # ────────────────────────────────────────────────────────────────────────────── # Plugin class # ────────────────────────────────────────────────────────────────────────────── class KnowtationRererePlugin: """Domain-aware rerere plugin for knowtation note conflicts. Standalone helper class with the bytes-based interface required by Phase 2.5. All methods delegate to the module-level functions of the same name so the plugin instance is a thin, stateless façade. The CLI integration (``muse rerere``) is achieved separately by :class:`muse.plugins.knowtation.plugin.KnowtationPlugin`, which implements the runtime-checkable :class:`muse.domain.RererePlugin` protocol method ``conflict_fingerprint(path, ours_id, theirs_id, repo_root)`` and delegates to :func:`conflict_fingerprint` after loading the ours / theirs blobs from the local object store. Thread safety: every method is independent of mutable state. Multiple threads may call any method concurrently; collisions on the rerere store are resolved by the atomic write pattern in :func:`_write_atomic`. """ #: Domain slug — used by orchestrators that key plugins by name. domain: str = DOMAIN # ------------------------------------------------------------------ # Fingerprinting # ------------------------------------------------------------------ def conflict_fingerprint( self, ours: bytes, theirs: bytes, base: bytes, ) -> str: """Return the bytes-based knowtation conflict fingerprint. Thin delegate to the module-level :func:`conflict_fingerprint`. See that function's docstring for stability guarantees and edge cases. Args: ours: Raw ours-side note bytes. theirs: Raw theirs-side note bytes. base: Raw merge-base note bytes (or ``b""`` when unknown). Returns: 64-char lowercase hex SHA-256 fingerprint. Raises: ValueError: When any input exceeds :data:`_MAX_NOTE_BYTES`. """ return conflict_fingerprint(ours, theirs, base) # ------------------------------------------------------------------ # Hash-only resolution # ------------------------------------------------------------------ def record_resolution( self, root: pathlib.Path, fingerprint: str, resolved: bytes, ) -> None: """Persist a hash-only resolution record — see :func:`record_resolution`. """ record_resolution(root, fingerprint, resolved) def lookup_resolution( self, root: pathlib.Path, fingerprint: str, ) -> str | None: """Return the recorded resolved-hash or ``None`` — see :func:`lookup_resolution`. """ return lookup_resolution(root, fingerprint) def replay_resolution( self, root: pathlib.Path, fingerprint: str, current: bytes, ) -> bytes | None: """Conservative hash-only replay — see :func:`replay_resolution`.""" return replay_resolution(root, fingerprint, current) # ------------------------------------------------------------------ # Full-bytes resolution # ------------------------------------------------------------------ def record_full_resolution( self, root: pathlib.Path, fingerprint: str, resolved: bytes, ) -> None: """Persist the canonical resolved bytes — see :func:`record_full_resolution`. """ record_full_resolution(root, fingerprint, resolved) def replay_full_resolution( self, root: pathlib.Path, fingerprint: str, current: bytes, ) -> bytes | None: """Return the canonical resolved bytes — see :func:`replay_full_resolution`. """ return replay_full_resolution(root, fingerprint, current) #: Module-level singleton — used by the registry hook in #: :mod:`muse.plugins.knowtation.plugin`. plugin: Final[KnowtationRererePlugin] = KnowtationRererePlugin() __all__ = [ "DOMAIN", "KnowtationRererePlugin", "conflict_fingerprint", "lookup_resolution", "plugin", "record_full_resolution", "record_resolution", "replay_full_resolution", "replay_resolution", ]