"""Knowtation four-layer note differ — Phase 2.1. Produces a structured :class:`~muse.domain.StructuredDelta` between two knowtation note byte-strings and applies that delta deterministically with :func:`apply`. The differ operates in four sequential layers: * **Layer 1 — Title.** Compares the YAML ``title`` field. Emits a single :class:`~muse.domain.ReplaceOp` when the title changes; otherwise no op. * **Layer 2 — Frontmatter.** For every other frontmatter key, emits fine-grained :class:`~muse.domain.InsertOp` / :class:`~muse.domain.DeleteOp` per element for set-valued dimensions (``tags``, ``entity``, ``follows``, ``summarizes``, ``attachments``) and :class:`~muse.domain.ReplaceOp` (or Insert / Delete when the field appears / disappears) for scalar fields. * **Layer 3 — Sections.** Splits each note body by Markdown heading, runs Myers diff (:func:`muse.core.diff_algorithms.lcs.myers_ses`) on the section ID sequence, and emits Insert / Delete / Move ops per section. Section IDs are level-, title-, and occurrence-qualified so that duplicate headings in the same note remain individually addressable. * **Layer 4 — Body lines.** For each section whose ID is present in both notes, runs Myers diff on the line sequence and emits Insert / Delete ops per changed line. Round-trip invariant -------------------- For any two notes ``A`` and ``B`` in *canonical form*:: apply(diff_notes(A, B), A) == B Canonical form is produced by :func:`canonicalize`: * LF-only line endings (CRLF is normalised). * Frontmatter serialised via :func:`yaml.safe_dump` with ``sort_keys=False``, ``default_flow_style=False``, ``allow_unicode=True``, ``width=10000``; ``title`` is hoisted to the first key, every other key is sorted alphabetically; set-valued list fields are deduplicated and sorted. * Frontmatter delimiters ``---\\n`` on both sides; no leading whitespace. Tests in ``tests/test_knowtation_differ.py`` always pre-canonicalise both sides before asserting the round-trip. Security and robustness ----------------------- * YAML parsing uses :func:`yaml.safe_load` exclusively — no arbitrary Python object instantiation. * Input size is bounded by :data:`_MAX_NOTE_BYTES`; oversize inputs raise :class:`ValueError`. * Invalid UTF-8 is decoded with ``errors="replace"`` and never raises. * Binary content (e.g. leading NUL bytes) is gracefully treated as a body with no frontmatter. No new dependencies are introduced. The module relies only on stdlib, ``pyyaml`` (already a runtime dependency), and existing ``muse`` modules. """ from __future__ import annotations import hashlib import logging from dataclasses import dataclass from typing import Any import yaml from muse.core.diff_algorithms.lcs import detect_moves, myers_ses from muse.domain import ( DeleteOp, DomainOp, InsertOp, MoveOp, ReplaceOp, StructuredDelta, ) from muse.plugins.knowtation.symbols import split_sections logger = logging.getLogger(__name__) # ────────────────────────────────────────────────────────────────────────────── # Configuration constants # ────────────────────────────────────────────────────────────────────────────── #: Maximum note size accepted by :func:`_parse_note`. Inputs larger than this #: are rejected with :class:`ValueError` to defend against runaway YAML parsing #: and pathological inputs. 16 MiB is generous for any realistic vault note #: while remaining small enough to bound algorithm complexity. _MAX_NOTE_BYTES: int = 16 * 1024 * 1024 #: Frontmatter fields treated as unordered sets when diffing. Insertions and #: deletions are emitted per element rather than as a single scalar replace. _SET_FIELDS: frozenset[str] = frozenset( {"tags", "entity", "follows", "summarizes", "attachments"} ) #: The title field is handled separately as Layer 1. _TITLE_KEY: str = "title" #: Address namespace for Layer 1 (title). _TITLE_ADDR: str = "title" #: Address prefix for Layer 2 (frontmatter). _FM_PREFIX: str = "frontmatter:" #: Address prefix for Layer 3 (sections). _SECTION_PREFIX: str = "section:" #: Address infix used by Layer 4 to disambiguate per-line body ops from the #: section addresses that contain them. The full address shape is #: ``"section::#<occurrence>::line:<n>"``. _LINE_INFIX: str = "::line:" #: The domain name returned in every :class:`StructuredDelta`. _DOMAIN: str = "knowtation" # ────────────────────────────────────────────────────────────────────────────── # Hash and stringify helpers # ────────────────────────────────────────────────────────────────────────────── def _content_id(value: str) -> str: """Return a deterministic SHA-256 hex digest of *value* (UTF-8 encoded). Used to populate ``content_id`` / ``old_content_id`` / ``new_content_id`` on every op so that downstream merge engines can rely on content-addressed equality without re-hashing. Args: value: The text whose content ID is required. Returns: 64-character lowercase hexadecimal string. """ return hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest() def _stringify(value: Any) -> str: """Render a YAML scalar (any type) to a stable string representation. The result is used both as the human-readable ``content_summary`` / ``old_summary`` / ``new_summary`` on ops and as the input to :func:`_content_id`. Booleans are rendered as the YAML literals ``"true"`` / ``"false"`` so that ``True`` and ``"true"`` produce identical content IDs. ``None`` becomes the empty string. Args: value: Any YAML-loadable scalar (str, int, float, bool, None, …). Returns: A deterministic string representation. """ if value is None: return "" if isinstance(value, bool): return "true" if value else "false" return str(value) # ────────────────────────────────────────────────────────────────────────────── # Note parsing and canonical re-serialisation # ────────────────────────────────────────────────────────────────────────────── @dataclass class _ParsedNote: """Internal representation of a note used by the differ. ``text`` is the full note text with CRLF normalised to LF. ``frontmatter`` is the parsed YAML mapping (always a ``dict``; empty when no frontmatter is present). ``body`` is the text following the frontmatter block, or the full text when no frontmatter exists. """ text: str frontmatter: dict[str, Any] body: str def _normalize_text(content: bytes) -> str: """Decode bytes to text and normalise line endings to LF only. Args: content: Raw note bytes. Returns: Decoded UTF-8 text with CRLF and bare CR converted to LF. Raises: ValueError: If *content* exceeds :data:`_MAX_NOTE_BYTES`. """ if len(content) > _MAX_NOTE_BYTES: raise ValueError( f"Note size {len(content)} bytes exceeds maximum {_MAX_NOTE_BYTES}" ) text = content.decode("utf-8", errors="replace") return text.replace("\r\n", "\n").replace("\r", "\n") def _parse_note(content: bytes) -> _ParsedNote: """Parse note bytes into a :class:`_ParsedNote`. Recognises a YAML frontmatter block delimited by ``---`` on both ends (with ``...`` also accepted as the closing delimiter, per the YAML spec). When parsing fails the entire content is treated as body text with no frontmatter — the function never raises on malformed input (apart from the size check inside :func:`_normalize_text`). Args: content: Raw note bytes. Returns: A :class:`_ParsedNote` with normalised text, parsed frontmatter, and the body text. """ text = _normalize_text(content) if not text.startswith("---"): return _ParsedNote(text=text, frontmatter={}, body=text) rest = text[3:] if not rest.startswith("\n"): return _ParsedNote(text=text, frontmatter={}, body=text) end = -1 for delim in ("\n---", "\n..."): pos = rest.find(delim) if pos != -1 and (end == -1 or pos < end): end = pos if end == -1: return _ParsedNote(text=text, frontmatter={}, body=text) fm_yaml = rest[:end] close_end = end + 4 if close_end < len(rest) and rest[close_end] == "\n": close_end += 1 body = rest[close_end:] try: data = yaml.safe_load(fm_yaml) except yaml.YAMLError as exc: logger.debug("YAMLError parsing frontmatter, treating as body: %s", exc) return _ParsedNote(text=text, frontmatter={}, body=text) if not isinstance(data, dict): return _ParsedNote(text=text, frontmatter={}, body=text) return _ParsedNote(text=text, frontmatter=data, body=body) def _order_frontmatter_keys(fm: dict[str, Any]) -> dict[str, Any]: """Reorder a frontmatter dict into canonical key order. Canonical order: ``title`` first when present, all remaining keys sorted alphabetically. This guarantees that two notes with identical key / value pairs in different YAML orderings canonicalise to the same bytes. Args: fm: Frontmatter mapping. Returns: A new ``dict`` with canonically-ordered keys. """ result: dict[str, Any] = {} if _TITLE_KEY in fm: result[_TITLE_KEY] = fm[_TITLE_KEY] for k in sorted(fm): if k != _TITLE_KEY: result[k] = fm[k] return result def _canonicalize_frontmatter(fm: dict[str, Any]) -> dict[str, Any]: """Apply canonical-form rules to a frontmatter dict. * Key order via :func:`_order_frontmatter_keys`. * Set-valued list fields (:data:`_SET_FIELDS`) are deduplicated and sorted. Args: fm: Frontmatter mapping. Returns: A new ``dict`` in canonical form. """ out = dict(fm) for key in _SET_FIELDS: val = out.get(key) if isinstance(val, list): unique_sorted = sorted({_stringify(x) for x in val}) if unique_sorted: out[key] = unique_sorted else: out.pop(key, None) return _order_frontmatter_keys(out) def _serialize_note(frontmatter: dict[str, Any], body: str) -> bytes: """Re-serialise a parsed note back to canonical bytes. When *frontmatter* is empty the output is just the encoded body (no delimiter lines). Otherwise the frontmatter is dumped via :func:`yaml.safe_dump` with the canonical option set (see module docstring) and wrapped in ``---`` delimiters. Args: frontmatter: Parsed frontmatter mapping. body: Body text (post-frontmatter). Returns: Canonical-form note bytes. """ canonical_fm = _canonicalize_frontmatter(frontmatter) if frontmatter else {} if not canonical_fm: return body.encode("utf-8") yaml_text = yaml.safe_dump( canonical_fm, sort_keys=False, default_flow_style=False, allow_unicode=True, width=10000, ) return f"---\n{yaml_text}---\n{body}".encode("utf-8") def canonicalize(content: bytes) -> bytes: """Return the canonical-form bytes for a note. Idempotent: ``canonicalize(canonicalize(x)) == canonicalize(x)``. Use this on both sides of a round-trip test to absorb formatting differences (CRLF vs LF, unsorted tags, alternative YAML key orders) that the round-trip invariant intentionally normalises away. Args: content: Raw note bytes. Returns: Canonical-form bytes. """ note = _parse_note(content) return _serialize_note(note.frontmatter, note.body) # ────────────────────────────────────────────────────────────────────────────── # Section model # ────────────────────────────────────────────────────────────────────────────── @dataclass class _Section: """A heading section with an occurrence-disambiguated stable ID. ``text`` is the verbatim section text, including the heading line itself and every line up to (but not including) the next heading. Concatenating every section's ``text`` reproduces the full body. """ sid: str level: int title: str text: str def _section_id(level: int, title: str, occurrence: int) -> str: """Compute a unique section ID from level, title, and occurrence index. Two sections with the same level and title within the same note are disambiguated by their ``occurrence`` (0-based count of the (level, title) pair so far). This keeps section IDs unique inside any single note while remaining deterministic across parses. Args: level: Heading level (0 for preamble, 1-6 for H1-H6). title: Heading text (no ``#`` characters). occurrence: 0-based index among sections sharing the same level/title. Returns: Address string, e.g. ``"section:2:Background#0"``. """ return f"{_SECTION_PREFIX}{level}:{title}#{occurrence}" def _split_sections_typed(body: str) -> list[_Section]: """Split *body* into a list of :class:`_Section` with stable IDs. Reuses :func:`muse.plugins.knowtation.symbols.split_sections` for the raw splitting logic, then assigns occurrence-indexed IDs. Always returns a list (possibly empty) — never raises. Args: body: Body text with frontmatter already stripped. Returns: Ordered list of sections; empty if *body* is blank. """ try: raw = split_sections(body) except Exception as exc: logger.debug("split_sections raised; treating as no sections: %s", exc) return [] occurrence: dict[tuple[int, str], int] = {} sections: list[_Section] = [] for level, title, sec_body, _line in raw: key = (level, title) idx = occurrence.get(key, 0) occurrence[key] = idx + 1 sections.append( _Section( sid=_section_id(level, title, idx), level=level, title=title, text=sec_body, ) ) return sections def _section_from_address(address: str, text: str) -> _Section: """Reconstruct a :class:`_Section` from a Layer 3 op's address and content. The address shape is ``"section:<level>:<title>#<occurrence>"`` (see :func:`_section_id`). Parsing is best-effort: malformed addresses default to ``level=0`` / ``title="(preamble)"`` so that apply() remains crash-free on adversarial input. Args: address: Section address from an op. text: Full section text from the op's ``content_summary``. Returns: A reconstructed :class:`_Section`. """ level = 0 title = "(preamble)" if address.startswith(_SECTION_PREFIX): rest = address[len(_SECTION_PREFIX) :] level_str, _, name_part = rest.partition(":") try: level = int(level_str) except ValueError: level = 0 # rpartition handles titles that themselves contain '#' characters. head, sep, _occ = name_part.rpartition("#") title = head if sep else name_part return _Section(sid=address, level=level, title=title, text=text) # ────────────────────────────────────────────────────────────────────────────── # Layer 1 — title diff # ────────────────────────────────────────────────────────────────────────────── def _diff_title(a: _ParsedNote, b: _ParsedNote) -> list[DomainOp]: """Emit at most one :class:`ReplaceOp` capturing the title change. The title is treated as a string ('' when the field is absent). No op is emitted when both sides agree. Args: a: Parsed base note. b: Parsed target note. Returns: ``[ReplaceOp]`` when the title changed, else ``[]``. """ title_a = a.frontmatter.get(_TITLE_KEY) title_b = b.frontmatter.get(_TITLE_KEY) if title_a == title_b: return [] old = _stringify(title_a) new = _stringify(title_b) return [ ReplaceOp( op="replace", address=_TITLE_ADDR, position=None, old_content_id=_content_id(old), new_content_id=_content_id(new), old_summary=old, new_summary=new, ) ] # ────────────────────────────────────────────────────────────────────────────── # Layer 2 — frontmatter diff (per-key, set-aware) # ────────────────────────────────────────────────────────────────────────────── def _diff_frontmatter(a: _ParsedNote, b: _ParsedNote) -> list[DomainOp]: """Per-key frontmatter diff, excluding the title (handled by Layer 1). Keys are processed in alphabetical order so that the output is deterministic. For each key: * Present only in *b* → :func:`_diff_field_added`. * Present only in *a* → :func:`_diff_field_removed`. * Present in both with different values → :func:`_diff_field_changed`. * Equal in both → no op. Args: a: Parsed base note. b: Parsed target note. Returns: Ordered list of ops (delete-side first, then insert-side, by key). """ fm_a = {k: v for k, v in a.frontmatter.items() if k != _TITLE_KEY} fm_b = {k: v for k, v in b.frontmatter.items() if k != _TITLE_KEY} ops: list[DomainOp] = [] for key in sorted(set(fm_a) | set(fm_b)): if key in fm_a and key not in fm_b: ops.extend(_diff_field_removed(key, fm_a[key])) elif key not in fm_a and key in fm_b: ops.extend(_diff_field_added(key, fm_b[key])) else: if fm_a[key] != fm_b[key]: ops.extend(_diff_field_changed(key, fm_a[key], fm_b[key])) return ops def _set_field_addr(key: str, item: str) -> str: """Return the per-element address for a set-valued frontmatter field. Args: key: Frontmatter key (e.g. ``"tags"``). item: Element value. Returns: ``"frontmatter:<key>::<item>"``. """ return f"{_FM_PREFIX}{key}::{item}" def _diff_field_added(key: str, value: Any) -> list[DomainOp]: """Emit insert ops for a newly-added frontmatter field. Set-valued fields produce one :class:`InsertOp` per element; scalar fields produce a single :class:`InsertOp` for the whole value. Args: key: Frontmatter key. value: New value. Returns: Ordered list of insert ops. """ if key in _SET_FIELDS and isinstance(value, list): return [ InsertOp( op="insert", address=_set_field_addr(key, _stringify(item)), position=None, content_id=_content_id(_stringify(item)), content_summary=_stringify(item), ) for item in sorted({_stringify(x) for x in value}) ] rendered = _stringify(value) return [ InsertOp( op="insert", address=f"{_FM_PREFIX}{key}", position=None, content_id=_content_id(rendered), content_summary=rendered, ) ] def _diff_field_removed(key: str, value: Any) -> list[DomainOp]: """Emit delete ops for a removed frontmatter field. Mirrors :func:`_diff_field_added` for the inverse direction. Args: key: Frontmatter key. value: Old value. Returns: Ordered list of delete ops. """ if key in _SET_FIELDS and isinstance(value, list): return [ DeleteOp( op="delete", address=_set_field_addr(key, _stringify(item)), position=None, content_id=_content_id(_stringify(item)), content_summary=_stringify(item), ) for item in sorted({_stringify(x) for x in value}) ] rendered = _stringify(value) return [ DeleteOp( op="delete", address=f"{_FM_PREFIX}{key}", position=None, content_id=_content_id(rendered), content_summary=rendered, ) ] def _diff_field_changed(key: str, val_a: Any, val_b: Any) -> list[DomainOp]: """Emit fine-grained ops for an existing field whose value changed. Set fields are diffed element-wise (Insert per added, Delete per removed). Scalar fields become a single :class:`ReplaceOp`. Args: key: Frontmatter key. val_a: Old value. val_b: New value. Returns: Ordered list of ops. """ if key in _SET_FIELDS and isinstance(val_a, list) and isinstance(val_b, list): set_a = {_stringify(x) for x in val_a} set_b = {_stringify(x) for x in val_b} added = sorted(set_b - set_a) removed = sorted(set_a - set_b) ops: list[DomainOp] = [] for item in removed: ops.append( DeleteOp( op="delete", address=_set_field_addr(key, item), position=None, content_id=_content_id(item), content_summary=item, ) ) for item in added: ops.append( InsertOp( op="insert", address=_set_field_addr(key, item), position=None, content_id=_content_id(item), content_summary=item, ) ) return ops old = _stringify(val_a) new = _stringify(val_b) return [ ReplaceOp( op="replace", address=f"{_FM_PREFIX}{key}", position=None, old_content_id=_content_id(old), new_content_id=_content_id(new), old_summary=old, new_summary=new, ) ] # ────────────────────────────────────────────────────────────────────────────── # Layer 3 — section diff # ────────────────────────────────────────────────────────────────────────────── def _diff_sections( sections_a: list[_Section], sections_b: list[_Section], ) -> tuple[list[DomainOp], set[str]]: """Diff two section sequences using Myers on the section ID list. For each Myers edit step: * ``keep`` → record the section ID as stable (used by Layer 4). * ``insert`` → emit :class:`InsertOp` carrying the full section text. * ``delete`` → emit :class:`DeleteOp` carrying the full section text. Adjacent insert / delete pairs sharing a content ID (unchanged section bytes) are collapsed into :class:`MoveOp` via :func:`muse.core.diff_algorithms.lcs.detect_moves`. Args: sections_a: Sections from the base note. sections_b: Sections from the target note. Returns: Tuple ``(ops, stable_section_ids)``. ``stable_section_ids`` is the set of IDs the Myers algorithm matched between the two sides; these sections are eligible for body-line diffing in Layer 4. """ ids_a = [s.sid for s in sections_a] ids_b = [s.sid for s in sections_b] a_by_id = {s.sid: s for s in sections_a} b_by_id = {s.sid: s for s in sections_b} steps = myers_ses(ids_a, ids_b) raw_inserts: list[InsertOp] = [] raw_deletes: list[DeleteOp] = [] stable_ids: set[str] = set() for step in steps: if step.kind == "keep": stable_ids.add(step.item) elif step.kind == "insert": sec = b_by_id[step.item] raw_inserts.append( InsertOp( op="insert", address=sec.sid, position=step.target_index, content_id=_content_id(sec.text), content_summary=sec.text, ) ) elif step.kind == "delete": sec = a_by_id[step.item] raw_deletes.append( DeleteOp( op="delete", address=sec.sid, position=step.base_index, content_id=_content_id(sec.text), content_summary=sec.text, ) ) moves, rem_inserts, rem_deletes = detect_moves(raw_inserts, raw_deletes) ops: list[DomainOp] = [*rem_deletes, *rem_inserts, *moves] return ops, stable_ids # ────────────────────────────────────────────────────────────────────────────── # Layer 4 — per-stable-section body line diff # ────────────────────────────────────────────────────────────────────────────── def _diff_body_lines(sid: str, body_a: str, body_b: str) -> list[DomainOp]: """Run Myers on the line sequences of one stable section's bodies. Each line gets an op whose ``address`` includes the parent section's ID and the per-line offset (``<sid>::line:<n>``). Inserts use the target line index; deletes use the base line index. Args: sid: Stable section ID (present in both A and B). body_a: Section body text from A. body_b: Section body text from B. Returns: Ordered list of ops: every delete first (descending base index is preserved by :func:`apply`), then every insert. """ lines_a = body_a.split("\n") lines_b = body_b.split("\n") if lines_a == lines_b: return [] steps = myers_ses(lines_a, lines_b) inserts: list[InsertOp] = [] deletes: list[DeleteOp] = [] for step in steps: if step.kind == "insert": inserts.append( InsertOp( op="insert", address=f"{sid}{_LINE_INFIX}{step.target_index}", position=step.target_index, content_id=_content_id(step.item), content_summary=step.item, ) ) elif step.kind == "delete": deletes.append( DeleteOp( op="delete", address=f"{sid}{_LINE_INFIX}{step.base_index}", position=step.base_index, content_id=_content_id(step.item), content_summary=step.item, ) ) return [*deletes, *inserts] # ────────────────────────────────────────────────────────────────────────────── # Public diff entry point # ────────────────────────────────────────────────────────────────────────────── def diff_notes(note_a: bytes, note_b: bytes) -> StructuredDelta: """Diff two knowtation notes, returning a typed :class:`StructuredDelta`. Runs all four layers in sequence and concatenates the resulting ops. The ``summary`` field is a short human-readable count of ops per layer. Args: note_a: Base note bytes (older state). note_b: Target note bytes (newer state). Returns: :class:`StructuredDelta` with ``domain="knowtation"`` and an ordered ``ops`` list. Raises: ValueError: If either input exceeds :data:`_MAX_NOTE_BYTES`. """ a = _parse_note(note_a) b = _parse_note(note_b) title_ops = _diff_title(a, b) fm_ops = _diff_frontmatter(a, b) sections_a = _split_sections_typed(a.body) sections_b = _split_sections_typed(b.body) section_ops, stable_ids = _diff_sections(sections_a, sections_b) body_ops: list[DomainOp] = [] a_section_text = {s.sid: s.text for s in sections_a} b_section_text = {s.sid: s.text for s in sections_b} for sid in sorted(stable_ids): text_a = a_section_text.get(sid, "") text_b = b_section_text.get(sid, "") if text_a != text_b: body_ops.extend(_diff_body_lines(sid, text_a, text_b)) ops: list[DomainOp] = [*title_ops, *fm_ops, *section_ops, *body_ops] summary = _summarize_ops(title_ops, fm_ops, section_ops, body_ops) return StructuredDelta(domain=_DOMAIN, ops=ops, summary=summary) def _summarize_ops( title_ops: list[DomainOp], fm_ops: list[DomainOp], section_ops: list[DomainOp], body_ops: list[DomainOp], ) -> str: """Compose a human-readable summary across the four diff layers. Args: title_ops: Ops produced by Layer 1. fm_ops: Ops produced by Layer 2. section_ops: Ops produced by Layer 3. body_ops: Ops produced by Layer 4. Returns: A short comma-separated string, or ``"no changes"`` when every layer is empty. """ parts: list[str] = [] if title_ops: parts.append("title changed") if fm_ops: n_ins = sum(1 for op in fm_ops if op["op"] == "insert") n_del = sum(1 for op in fm_ops if op["op"] == "delete") n_rep = sum(1 for op in fm_ops if op["op"] == "replace") sub: list[str] = [] if n_ins: sub.append(f"{n_ins} frontmatter added") if n_del: sub.append(f"{n_del} frontmatter removed") if n_rep: sub.append(f"{n_rep} frontmatter changed") parts.extend(sub) if section_ops: n_ins = sum(1 for op in section_ops if op["op"] == "insert") n_del = sum(1 for op in section_ops if op["op"] == "delete") n_mov = sum(1 for op in section_ops if op["op"] == "move") sub_s: list[str] = [] if n_ins: sub_s.append(f"{n_ins} sections added") if n_del: sub_s.append(f"{n_del} sections removed") if n_mov: sub_s.append(f"{n_mov} sections moved") parts.extend(sub_s) if body_ops: n_lines = len(body_ops) parts.append(f"{n_lines} body lines changed") return ", ".join(parts) if parts else "no changes" # ────────────────────────────────────────────────────────────────────────────── # apply() # ────────────────────────────────────────────────────────────────────────────── def apply(delta: StructuredDelta, content: bytes) -> bytes: """Apply *delta* to a note's bytes, returning canonical-form new bytes. The four layers are applied in order: title → frontmatter → sections → body lines. The function is pure: it never mutates *content* or *delta* and always returns a new ``bytes`` object. The output is in canonical form (see :func:`canonicalize` and the module docstring). Tests should canonicalise both sides of a round-trip assertion to absorb formatting differences. Args: delta: A :class:`StructuredDelta` produced by :func:`diff_notes`. content: The base note bytes the delta applies to. Returns: The new note bytes after every op has been applied. Raises: ValueError: If *content* exceeds :data:`_MAX_NOTE_BYTES`. """ note = _parse_note(content) fm = dict(note.frontmatter) title_ops, fm_ops, section_ops, body_line_ops = _bucket_ops( delta.get("ops", []) ) fm = _apply_title_ops(fm, title_ops) fm = _apply_fm_ops(fm, fm_ops) sections = _split_sections_typed(note.body) sections = _apply_section_ops(sections, section_ops) sections = _apply_body_line_ops(sections, body_line_ops) new_body = "".join(s.text for s in sections) return _serialize_note(fm, new_body) def _bucket_ops( ops: list[DomainOp], ) -> tuple[ list[DomainOp], list[DomainOp], list[DomainOp], dict[str, list[DomainOp]], ]: """Sort ops into per-layer buckets based on their address shape. Address conventions (see module docstring): * ``"title"`` → Layer 1. * ``"frontmatter:…"`` → Layer 2. * ``"section:…::line:n"`` → Layer 4, keyed by parent section ID. * ``"section:…"`` → Layer 3. Unknown addresses are logged and silently dropped to keep apply() robust against partial / adversarial deltas. Args: ops: The op list from a :class:`StructuredDelta`. Returns: Tuple of ``(title_ops, fm_ops, section_ops, body_line_ops_by_sid)``. """ title_ops: list[DomainOp] = [] fm_ops: list[DomainOp] = [] section_ops: list[DomainOp] = [] body_line_ops: dict[str, list[DomainOp]] = {} for op in ops: addr = op.get("address", "") if addr == _TITLE_ADDR: title_ops.append(op) elif _LINE_INFIX in addr: sid = addr.split(_LINE_INFIX, 1)[0] body_line_ops.setdefault(sid, []).append(op) elif addr.startswith(_FM_PREFIX): fm_ops.append(op) elif addr.startswith(_SECTION_PREFIX): section_ops.append(op) else: logger.debug("Dropping op with unknown address: %r", addr) return title_ops, fm_ops, section_ops, body_line_ops def _apply_title_ops( fm: dict[str, Any], ops: list[DomainOp] ) -> dict[str, Any]: """Apply Layer 1 ops to *fm*, returning a new dict. A ReplaceOp on the title sets / clears the ``title`` key. When the new value is the empty string the key is removed entirely so that the canonical-form serialisation does not include an empty title line. Args: fm: Current frontmatter mapping. ops: Layer 1 ops (zero or one ReplaceOp). Returns: New frontmatter mapping. """ result = dict(fm) for op in ops: if op["op"] != "replace": continue new = op.get("new_summary", "") if new == "": result.pop(_TITLE_KEY, None) else: result[_TITLE_KEY] = new return result def _apply_fm_ops( fm: dict[str, Any], ops: list[DomainOp] ) -> dict[str, Any]: """Apply Layer 2 ops to *fm*, returning a new dict. Set-field addresses (``"frontmatter:<key>::<item>"``) trigger element-wise set updates; scalar-field addresses (``"frontmatter:<key>"``) trigger insert / delete / replace on the value. The resulting set fields are deduplicated and sorted to keep canonical form stable. Args: fm: Current frontmatter mapping. ops: Layer 2 ops (any combination of insert / delete / replace). Returns: New frontmatter mapping. """ result = dict(fm) for op in ops: addr: str = op["address"] if "::" in addr: base_addr, _, item = addr.partition("::") key = base_addr[len(_FM_PREFIX) :] current = result.get(key) current_list: list[str] = ( [str(x) for x in current] if isinstance(current, list) else [] ) kind = op["op"] if kind == "insert": if item not in current_list: current_list.append(item) elif kind == "delete": while item in current_list: current_list.remove(item) current_list = sorted(set(current_list)) if current_list: result[key] = current_list else: result.pop(key, None) else: key = addr[len(_FM_PREFIX) :] kind = op["op"] if kind == "insert": result[key] = op.get("content_summary", "") elif kind == "delete": result.pop(key, None) elif kind == "replace": result[key] = op.get("new_summary", "") return result def _apply_section_ops( sections: list[_Section], ops: list[DomainOp] ) -> list[_Section]: """Apply Layer 3 ops to *sections*, returning a new list. MoveOps are decomposed back into a delete-at-source + insert-at-target pair using the base section list as the content source (a move never changes section bytes by definition). Then all deletes are applied in descending position order, followed by all inserts in ascending target position order. This delete-then-insert ordering is the well-known correct strategy for LCS edit-script replay. Args: sections: Current section list. ops: Layer 3 ops. Returns: New section list. """ cid_to_section: dict[str, _Section] = { _content_id(s.text): s for s in sections } deletes: list[DomainOp] = [] inserts: list[DomainOp] = [] for op in ops: if op["op"] == "delete": deletes.append(op) elif op["op"] == "insert": inserts.append(op) elif op["op"] == "move": move_op: MoveOp = op cid = move_op["content_id"] src = cid_to_section.get(cid) if src is None: logger.debug( "MoveOp references unknown content_id %r; skipping", cid ) continue deletes.append( DeleteOp( op="delete", address=move_op["address"], position=move_op["from_position"], content_id=cid, content_summary=src.text, ) ) inserts.append( InsertOp( op="insert", address=move_op["address"], position=move_op["to_position"], content_id=cid, content_summary=src.text, ) ) intermediate = list(sections) for d in sorted(deletes, key=lambda o: -int(o.get("position") or 0)): pos_val = d.get("position") if pos_val is None: continue pos = int(pos_val) if 0 <= pos < len(intermediate): del intermediate[pos] result = list(intermediate) for ins in sorted(inserts, key=lambda o: int(o.get("position") or 0)): pos_val = ins.get("position") if pos_val is None: pos = len(result) else: pos = int(pos_val) text = ins.get("content_summary", "") sec = _section_from_address(ins["address"], text) if pos < 0: pos = 0 if pos > len(result): pos = len(result) result.insert(pos, sec) return result def _apply_body_line_ops( sections: list[_Section], body_line_ops: dict[str, list[DomainOp]], ) -> list[_Section]: """Apply Layer 4 ops, returning a new section list with edited bodies. For each section ID present in *body_line_ops*, look up the section in *sections* by ID, split its body into lines, then run the standard delete-descending / insert-ascending replay. Sections not present in *body_line_ops* are passed through unchanged. Args: sections: Current section list (after Layer 3 has been applied). body_line_ops: Mapping of section ID → list of ops. Returns: New section list with body-line edits applied. """ if not body_line_ops: return sections sid_to_index: dict[str, int] = {s.sid: i for i, s in enumerate(sections)} result = list(sections) for sid, ops in body_line_ops.items(): idx = sid_to_index.get(sid) if idx is None: logger.debug( "Body line ops reference unknown section %r; skipping", sid ) continue sec = result[idx] lines = sec.text.split("\n") deletes = [op for op in ops if op["op"] == "delete"] inserts = [op for op in ops if op["op"] == "insert"] for d in sorted(deletes, key=lambda o: -int(o.get("position") or 0)): pos_val = d.get("position") if pos_val is None: continue pos = int(pos_val) if 0 <= pos < len(lines): del lines[pos] for ins in sorted(inserts, key=lambda o: int(o.get("position") or 0)): pos_val = ins.get("position") if pos_val is None: pos = len(lines) else: pos = int(pos_val) line = ins.get("content_summary", "") if pos < 0: pos = 0 if pos > len(lines): pos = len(lines) lines.insert(pos, line) new_text = "\n".join(lines) result[idx] = _Section( sid=sec.sid, level=sec.level, title=sec.title, text=new_text ) return result # ────────────────────────────────────────────────────────────────────────────── # Public re-exports # ────────────────────────────────────────────────────────────────────────────── __all__ = [ "diff_notes", "apply", "canonicalize", ]