op_transform.py python
550 lines 21.4 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
1 """Operational transformation for Muse domain operations.
2
3 This module implements the commutativity rules and position-adjustment
4 transforms that allow the merge engine to reason over ``DomainOp`` trees
5 rather than file-path sets. The result is sub-file auto-merge: two agents
6 inserting notes at non-overlapping bars never produce a conflict.
7
8 Theory
9 ------
10 Operational Transformation (OT) is the theory behind real-time collaborative
11 editors (Google Docs, VS Code Live Share). The key insight is that two
12 *concurrent* operations — generated independently against the same base state
13 — can be applied in sequence without conflict if and only if they *commute*:
14 applying them in either order yields the same final state.
15
16 For concurrent operations that commute, OT provides a ``transform`` function
17 that adjusts positions so that the result is identical regardless of which
18 operation is applied first.
19
20 Public API
21 ----------
22 - :class:`MergeOpsResult` — structured result of merging two op lists.
23 - :func:`ops_commute` — commmutativity oracle for any two ``DomainOp``\\s.
24 - :func:`transform` — position-adjusted ``(a', b')`` for commuting ops.
25 - :func:`merge_op_lists` — three-way merge at operation granularity.
26
27 Commutativity rules (summary)
28 ------------------------------
29
30 ============================================= =====================================
31 Op A Op B Commute?
32 ============================================= =====================================
33 InsertOp(pos=i) InsertOp(pos=j) Yes — if i ≠ j or both None (unordered)
34 InsertOp(pos=i) InsertOp(pos=i) **No** — positional conflict
35 InsertOp(addr=A) DeleteOp(addr=B) Yes — if A ≠ B (different containers)
36 InsertOp(addr=A) DeleteOp(addr=A) **No** — same container
37 DeleteOp(addr=A) DeleteOp(addr=B) Yes — always (consensus delete is fine)
38 ReplaceOp(addr=A) ReplaceOp(addr=B) Yes — if A ≠ B
39 ReplaceOp(addr=A) ReplaceOp(addr=A) **No** — concurrent value conflict
40 MoveOp(from=i) MoveOp(from=j) Yes — if i ≠ j
41 MoveOp(from=i) DeleteOp(pos=i) **No** — move-delete conflict
42 PatchOp(addr=A) PatchOp(addr=B) Yes — if A ≠ B; recurse if A == B
43 ============================================= =====================================
44
45 Position adjustment
46 -------------------
47 When two ``InsertOp``\\s at different positions commute, the later-applied one
48 must have its position adjusted. Concretely, if *a* inserts at position *i*
49 and *b* inserts at position *j* with *i < j*:
50
51 - Applying *a* first shifts every element at position ≥ i by one; so *b*
52 must be adjusted to *j + 1*.
53 - Applying *b* first does not affect positions < j; so *a* stays at *i*.
54
55 For *merge_op_lists*, positions are adjusted via the **counting formula**: for
56 each InsertOp in one side's exclusive additions, add the count of the other
57 side's InsertOps that have position ≤ this op's position (on the same
58 address). This is correct for any number of concurrent insertions and avoids
59 the cascading adjustment errors that arise from naive sequential pairwise OT.
60
61 Synchronous guarantee
62 ---------------------
63 All functions are synchronous, pure, and allocation-bounded — no I/O, no
64 async, no external state.
65 """
66
67 from __future__ import annotations
68
69 import logging
70 from dataclasses import dataclass, field
71
72 from muse.domain import (
73
74 DeleteOp,
75 DomainOp,
76 InsertOp,
77 MoveOp,
78 PatchOp,
79 ReplaceOp,
80 StructuredDelta,
81 )
82
83 type _AddrMap = dict[str, list[int]]
84
85 logger = logging.getLogger(__name__)
86
87
88 # ---------------------------------------------------------------------------
89 # Result type
90 # ---------------------------------------------------------------------------
91
92
93 @dataclass
94 class MergeOpsResult:
95 """Result of a three-way operation-level merge.
96
97 ``merged_ops`` contains the operations from both sides that can be applied
98 to the common ancestor to produce the merged state. Positions in any
99 ``InsertOp`` entries have been adjusted so that the ops can be applied in
100 ascending position order to produce a deterministic result.
101
102 ``conflict_ops`` contains pairs ``(our_op, their_op)`` where the two
103 operations cannot be auto-merged. Each pair must be resolved manually
104 (or via ``.museattributes`` strategy) before the merge can complete.
105
106 ``is_clean`` is ``True`` when ``conflict_ops`` is empty.
107 """
108
109 merged_ops: list[DomainOp] = field(default_factory=list)
110 conflict_ops: list[tuple[DomainOp, DomainOp]] = field(default_factory=list)
111
112 @property
113 def is_clean(self) -> bool:
114 """``True`` when no conflicting operation pairs were found."""
115 return len(self.conflict_ops) == 0
116
117
118 # ---------------------------------------------------------------------------
119 # Internal helpers
120 # ---------------------------------------------------------------------------
121
122
123 def _op_key(op: DomainOp) -> tuple[str, ...]:
124 """Return a hashable key uniquely identifying *op* for set membership tests.
125
126 The key captures all semantically significant fields so that two ops with
127 identical effect produce the same key. This is used to detect consensus
128 operations (both sides added the same op independently).
129 """
130 if op["op"] == "insert":
131 return ("insert", op["address"], str(op["position"]), op["content_id"])
132 if op["op"] == "delete":
133 return ("delete", op["address"], str(op["position"]), op["content_id"])
134 if op["op"] == "move":
135 return (
136 "move",
137 op["address"],
138 str(op["from_position"]),
139 str(op["to_position"]),
140 op["content_id"],
141 )
142 if op["op"] == "replace":
143 return (
144 "replace",
145 op["address"],
146 str(op["position"]),
147 op["old_content_id"],
148 op["new_content_id"],
149 )
150 if op["op"] == "mutate":
151 return ("mutate", op["address"], op["entity_id"], op["old_content_id"], op["new_content_id"])
152 # PatchOp — key on address and child_domain; child_ops are not hashed for
153 # performance reasons. Two patch ops on the same container are treated as
154 # the same "slot" for conflict detection purposes.
155 return ("patch", op["address"], op["child_domain"])
156
157
158 # ---------------------------------------------------------------------------
159 # Commutativity oracle
160 # ---------------------------------------------------------------------------
161
162
163 def ops_commute(a: DomainOp, b: DomainOp) -> bool:
164 """Return ``True`` if operations *a* and *b* commute (are auto-mergeable).
165
166 Two operations commute when applying them in either order produces the
167 same final state. This function implements the commutativity rules table
168 for all 25 op-kind pairs.
169
170 For ``PatchOp`` at the same address, commmutativity is determined
171 recursively by checking all child-op pairs.
172
173 Args:
174 a: First domain operation.
175 b: Second domain operation.
176
177 Returns:
178 ``True`` if the two operations can be safely auto-merged.
179 """
180 # ------------------------------------------------------------------
181 # InsertOp + *
182 # ------------------------------------------------------------------
183 if a["op"] == "insert":
184 if b["op"] == "insert":
185 # Different containers always commute — they are completely independent.
186 if a["address"] != b["address"]:
187 return True
188 a_pos, b_pos = a["position"], b["position"]
189 # Unordered collections (position=None) always commute.
190 if a_pos is None or b_pos is None:
191 return True
192 # Ordered sequences within the same container: conflict only at equal positions.
193 return a_pos != b_pos
194 if b["op"] == "delete":
195 # Conservative: inserts and deletes at the same container conflict.
196 return a["address"] != b["address"]
197 if b["op"] == "move":
198 return a["address"] != b["address"]
199 if b["op"] == "replace":
200 return a["address"] != b["address"]
201 # b is PatchOp (exhaustion of DeleteOp | MoveOp | ReplaceOp | PatchOp)
202 return a["address"] != b["address"]
203
204 # ------------------------------------------------------------------
205 # DeleteOp + *
206 # ------------------------------------------------------------------
207 if a["op"] == "delete":
208 if b["op"] == "insert":
209 return a["address"] != b["address"]
210 if b["op"] == "delete":
211 # Consensus delete (same or different address) always commutes.
212 # Two branches that both removed the same element produce the same
213 # result: the element is absent.
214 return True
215 if b["op"] == "move":
216 # Conflict if the delete's position matches the move's source.
217 a_pos = a["position"]
218 if a_pos is None:
219 return True # unordered collection: no positional conflict
220 return a_pos != b["from_position"]
221 if b["op"] == "replace":
222 return a["address"] != b["address"]
223 # b is PatchOp
224 return a["address"] != b["address"]
225
226 # ------------------------------------------------------------------
227 # MoveOp + *
228 # ------------------------------------------------------------------
229 if a["op"] == "move":
230 if b["op"] == "insert":
231 return a["address"] != b["address"]
232 if b["op"] == "delete":
233 b_pos = b["position"]
234 if b_pos is None:
235 return True
236 return a["from_position"] != b_pos
237 if b["op"] == "move":
238 # Two moves from different source positions commute.
239 return a["from_position"] != b["from_position"]
240 if b["op"] == "replace":
241 return a["address"] != b["address"]
242 # b is PatchOp
243 return a["address"] != b["address"]
244
245 # ------------------------------------------------------------------
246 # ReplaceOp + *
247 # ------------------------------------------------------------------
248 if a["op"] == "replace":
249 if b["op"] == "insert":
250 return a["address"] != b["address"]
251 if b["op"] == "delete":
252 return a["address"] != b["address"]
253 if b["op"] == "move":
254 return a["address"] != b["address"]
255 if b["op"] == "replace":
256 # Two replaces at the same address conflict (concurrent value change).
257 return a["address"] != b["address"]
258 # b is PatchOp
259 return a["address"] != b["address"]
260
261 # ------------------------------------------------------------------
262 # MutateOp + * (a["op"] == "mutate" commutes with everything at a
263 # different address; same-entity concurrent mutations conflict)
264 # ------------------------------------------------------------------
265 if a["op"] == "mutate":
266 if b["op"] == "mutate":
267 return a["entity_id"] != b["entity_id"]
268 return a["address"] != b["address"]
269
270 # ------------------------------------------------------------------
271 # PatchOp + * (a["op"] == "patch" after all checks above)
272 # ------------------------------------------------------------------
273 if b["op"] == "insert":
274 return a["address"] != b["address"]
275 if b["op"] == "delete":
276 return a["address"] != b["address"]
277 if b["op"] == "move":
278 return a["address"] != b["address"]
279 if b["op"] == "replace":
280 return a["address"] != b["address"]
281 if b["op"] == "mutate":
282 return a["address"] != b["address"]
283 # b is PatchOp
284 if a["address"] != b["address"]:
285 return True
286 # Same address: recurse into child ops — all child pairs must commute.
287 for child_a in a["child_ops"]:
288 for child_b in b["child_ops"]:
289 if not ops_commute(child_a, child_b):
290 return False
291 return True
292
293
294 # ---------------------------------------------------------------------------
295 # OT transform
296 # ---------------------------------------------------------------------------
297
298
299 def transform(a: DomainOp, b: DomainOp) -> tuple[DomainOp, DomainOp]:
300 """Return ``(a', b')`` such that ``apply(apply(base, a), b') == apply(apply(base, b), a')``.
301
302 This is the core OT transform function. It should only be called when
303 :func:`ops_commute` has confirmed that *a* and *b* commute. For all
304 commuting pairs except ordered InsertOp+InsertOp, the identity transform
305 is returned — the operations do not interfere with each other's positions.
306
307 For the InsertOp+InsertOp case with integer positions (the most common
308 case in practice), positions are adjusted so the diamond property holds:
309 the same final sequence is produced regardless of application order.
310
311 Args:
312 a: First domain operation.
313 b: Second domain operation (must commute with *a*).
314
315 Returns:
316 A tuple ``(a', b')`` where:
317
318 - *a'* is the version of *a* to apply when *b* has already been applied.
319 - *b'* is the version of *b* to apply when *a* has already been applied.
320 """
321 if a["op"] == "insert" and b["op"] == "insert":
322 a_pos, b_pos = a["position"], b["position"]
323 if a_pos is not None and b_pos is not None and a_pos != b_pos:
324 if a_pos < b_pos:
325 # a inserts before b's original position → b shifts up by 1.
326 b_prime = InsertOp(
327 op="insert",
328 address=b["address"],
329 position=b_pos + 1,
330 content_id=b["content_id"],
331 content_summary=b["content_summary"],
332 )
333 return a, b_prime
334 else:
335 # b inserts before a's original position → a shifts up by 1.
336 a_prime = InsertOp(
337 op="insert",
338 address=a["address"],
339 position=a_pos + 1,
340 content_id=a["content_id"],
341 content_summary=a["content_summary"],
342 )
343 return a_prime, b
344
345 # All other commuting pairs: identity transform.
346 return a, b
347
348
349 # ---------------------------------------------------------------------------
350 # Three-way merge at operation granularity
351 # ---------------------------------------------------------------------------
352
353
354 def _adjust_insert_positions(
355 ops: list[DomainOp],
356 other_ops: list[DomainOp],
357 ) -> list[DomainOp]:
358 """Adjust ``InsertOp`` positions in *ops* to account for *other_ops*.
359
360 For each ``InsertOp`` with a non-``None`` position in *ops*, the adjusted
361 position is ``original_position + count`` where ``count`` is the number of
362 ``InsertOp``\\s in *other_ops* that share the same ``address`` and have
363 ``position ≤ original_position``.
364
365 This implements the *counting formula* for multi-op position adjustment.
366 It is correct for any number of concurrent insertions on each side,
367 producing the same final sequence regardless of application order.
368
369 Non-``InsertOp`` entries and unordered inserts (``position=None``) pass
370 through unchanged.
371
372 Args:
373 ops: The list of ops whose positions need adjustment.
374 other_ops: The concurrent operations from the other branch.
375
376 Returns:
377 A new list with adjusted ``InsertOp``\\s; all other entries are copied
378 unchanged.
379 """
380 # Collect other-side InsertOp positions, grouped by address.
381 other_by_addr: _AddrMap = {}
382 for op in other_ops:
383 if op["op"] == "insert" and op["position"] is not None:
384 addr = op["address"]
385 if addr not in other_by_addr:
386 other_by_addr[addr] = []
387 other_by_addr[addr].append(op["position"])
388
389 result: list[DomainOp] = []
390 for op in ops:
391 if op["op"] == "insert" and op["position"] is not None:
392 addr = op["address"]
393 pos = op["position"]
394 others = other_by_addr.get(addr, [])
395 shift = sum(1 for p in others if p <= pos)
396 if shift:
397 result.append(
398 InsertOp(
399 op="insert",
400 address=addr,
401 position=pos + shift,
402 content_id=op["content_id"],
403 content_summary=op["content_summary"],
404 )
405 )
406 else:
407 result.append(op)
408 else:
409 result.append(op)
410
411 return result
412
413
414 def merge_op_lists(
415 base_ops: list[DomainOp],
416 ours_ops: list[DomainOp],
417 theirs_ops: list[DomainOp],
418 ) -> MergeOpsResult:
419 """Three-way merge at operation granularity.
420
421 Implements the standard three-way merge algorithm applied to typed domain
422 operations rather than file-path sets. The inputs represent:
423
424 - *base_ops*: operations present in the common ancestor.
425 - *ours_ops*: operations present on our branch (superset of base for
426 kept ops, plus our new additions).
427 - *theirs_ops*: operations present on their branch (same structure).
428
429 Algorithm
430 ---------
431 1. **Kept from base** — ops in base that both sides retained are included
432 unchanged.
433 2. **Consensus additions** — ops added independently by both sides (same
434 key) are included exactly once (idempotent).
435 3. **Exclusive additions** — ops added by only one side enter the
436 commmutativity check:
437
438 - Any pair (ours_exclusive, theirs_exclusive) where
439 :func:`ops_commute` returns ``False`` is recorded as a conflict.
440 - Exclusive additions not involved in any conflict are included in
441 ``merged_ops``, with ``InsertOp`` positions adjusted via
442 :func:`_adjust_insert_positions`.
443
444 Position adjustment note
445 ------------------------
446 The adjusted ``InsertOp`` positions in ``merged_ops`` are *absolute
447 positions in the final merged sequence* — meaning they already account for
448 all insertions from both sides. Callers applying the merged ops to the
449 base state should apply ``InsertOp``\\s in ascending position order to
450 obtain the correct final sequence.
451
452 Args:
453 base_ops: Operations in the common ancestor delta.
454 ours_ops: Operations on our branch.
455 theirs_ops: Operations on their branch.
456
457 Returns:
458 A :class:`MergeOpsResult` with merged and conflicting op lists.
459 """
460 base_key_set = {_op_key(op) for op in base_ops}
461 ours_key_set = {_op_key(op) for op in ours_ops}
462 theirs_key_set = {_op_key(op) for op in theirs_ops}
463
464 # 1. Ops both sides kept from the base.
465 kept: list[DomainOp] = [
466 op
467 for op in base_ops
468 if _op_key(op) in ours_key_set and _op_key(op) in theirs_key_set
469 ]
470
471 # 2. New ops — not present in base.
472 ours_new = [op for op in ours_ops if _op_key(op) not in base_key_set]
473 theirs_new = [op for op in theirs_ops if _op_key(op) not in base_key_set]
474
475 ours_new_keys = {_op_key(op) for op in ours_new}
476 theirs_new_keys = {_op_key(op) for op in theirs_new}
477 consensus_keys = ours_new_keys & theirs_new_keys
478
479 # Consensus additions: both sides added the same op → include once.
480 consensus: list[DomainOp] = [
481 op for op in ours_new if _op_key(op) in consensus_keys
482 ]
483
484 # 3. Each side's exclusive new additions.
485 ours_exclusive = [op for op in ours_new if _op_key(op) not in consensus_keys]
486 theirs_exclusive = [op for op in theirs_new if _op_key(op) not in consensus_keys]
487
488 # Conflict detection: any pair from both sides that does not commute.
489 conflict_ops: list[tuple[DomainOp, DomainOp]] = []
490 conflicting_ours_keys: set[tuple[str, ...]] = set()
491 conflicting_theirs_keys: set[tuple[str, ...]] = set()
492
493 for our_op in ours_exclusive:
494 for their_op in theirs_exclusive:
495 if not ops_commute(our_op, their_op):
496 conflict_ops.append((our_op, their_op))
497 conflicting_ours_keys.add(_op_key(our_op))
498 conflicting_theirs_keys.add(_op_key(their_op))
499
500 # 4. Clean ops: not involved in any conflict.
501 clean_ours = [
502 op for op in ours_exclusive if _op_key(op) not in conflicting_ours_keys
503 ]
504 clean_theirs = [
505 op for op in theirs_exclusive if _op_key(op) not in conflicting_theirs_keys
506 ]
507
508 # 5. Position adjustment using the counting formula.
509 clean_ours_adjusted = _adjust_insert_positions(clean_ours, clean_theirs)
510 clean_theirs_adjusted = _adjust_insert_positions(clean_theirs, clean_ours)
511
512 merged_ops: list[DomainOp] = (
513 list(kept) + list(consensus) + clean_ours_adjusted + clean_theirs_adjusted
514 )
515
516 logger.debug(
517 "merge_op_lists: kept=%d consensus=%d ours=%d theirs=%d conflicts=%d",
518 len(kept),
519 len(consensus),
520 len(clean_ours_adjusted),
521 len(clean_theirs_adjusted),
522 len(conflict_ops),
523 )
524
525 return MergeOpsResult(merged_ops=merged_ops, conflict_ops=conflict_ops)
526
527
528 def merge_structured(
529 base_delta: StructuredDelta,
530 ours_delta: StructuredDelta,
531 theirs_delta: StructuredDelta,
532 ) -> MergeOpsResult:
533 """Merge two structured deltas against a common base delta.
534
535 A convenience wrapper over :func:`merge_op_lists` that accepts
536 :class:`~muse.domain.StructuredDelta` objects directly.
537
538 Args:
539 base_delta: Delta representing the common ancestor's operations.
540 ours_delta: Delta produced by our branch.
541 theirs_delta: Delta produced by their branch.
542
543 Returns:
544 A :class:`MergeOpsResult` describing the merged and conflicting ops.
545 """
546 return merge_op_lists(
547 base_delta["ops"],
548 ours_delta["ops"],
549 theirs_delta["ops"],
550 )
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 33 days ago