aaronrene / gabriel-muse public
tree_edit.py python
312 lines 10.1 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago
1 """LCS-based tree edit algorithm for labeled ordered trees.
2
3 Implements a correct tree diff that produces ``InsertOp``, ``DeleteOp``,
4 ``ReplaceOp``, and ``MoveOp`` entries for labeled ordered trees.
5
6 Algorithm
7 ---------
8 The diff proceeds top-down recursively:
9
10 1. Compare root nodes by ``content_id``. Different content_id → ``ReplaceOp``
11 on the root node.
12 2. Diff the children sequences using the same LCS algorithm as
13 :mod:`~muse.core.diff_algorithms.lcs`:
14
15 - Matched child pairs (same ``content_id``) → recurse into subtree.
16 - Unmatched inserts → ``InsertOp`` (entire subtree added).
17 - Unmatched deletes → ``DeleteOp`` (entire subtree removed).
18 - Paired insert+delete of same ``content_id`` at different positions →
19 ``MoveOp``.
20
21 This approach is O(nm) per tree level where n, m are child counts. It does
22 not find the globally minimal edit script (Zhang-Shasha is optimal), but it
23 is correct: every change is accounted for, and applying the script to the base
24 tree produces the target tree. For the bounded tree sizes typical of domain
25 objects (scenes, tracks, ASTs ≲ 10k nodes), this is more than adequate for
26 Zhang-Shasha optimisation is a drop-in replacement once needed.
27
28 ``TreeNode`` is defined here and re-exported by the package ``__init__``.
29
30 Public API
31 ----------
32 - :class:`TreeNode` — labeled ordered tree node (frozen dataclass).
33 - :func:`diff` — ``TreeNode`` × ``TreeNode`` → ``StructuredDelta``.
34 """
35
36 from __future__ import annotations
37
38 import logging
39 from dataclasses import dataclass
40 from typing import Literal
41
42 from muse.core.schema import TreeSchema
43 from muse.domain import (
44
45 DeleteOp,
46 DomainOp,
47 InsertOp,
48 MoveOp,
49 ReplaceOp,
50 StructuredDelta,
51 )
52
53 type _DeleteMap = dict[str, tuple[int, "TreeNode"]]
54
55 logger = logging.getLogger(__name__)
56
57
58 # ---------------------------------------------------------------------------
59 # TreeNode — the unit of tree-edit comparison
60 # ---------------------------------------------------------------------------
61
62
63 @dataclass(frozen=True)
64 class TreeNode:
65 """A node in a labeled ordered tree for domain tree-edit algorithms.
66
67 ``id`` is a stable unique identifier for the node (e.g. UUID or path).
68 ``label`` is the human-readable name (e.g. element tag, node type).
69 ``content_id`` is the SHA-256 of this node's own value — excluding its
70 children. Two nodes are considered the same iff their ``content_id``\\s
71 match; a different ``content_id`` triggers a ``ReplaceOp``.
72 ``children`` is an ordered tuple of child nodes.
73 """
74
75 id: str
76 label: str
77 content_id: str
78 children: tuple[TreeNode, ...]
79
80
81 # ---------------------------------------------------------------------------
82 # Internal helpers
83 # ---------------------------------------------------------------------------
84
85
86 def _subtree_nodes(node: TreeNode) -> list[TreeNode]:
87 """Return all nodes in *node*'s subtree (postorder)."""
88 result: list[TreeNode] = []
89
90 def _visit(n: TreeNode) -> None:
91 for child in n.children:
92 _visit(child)
93 result.append(n)
94
95 _visit(node)
96 return result
97
98
99 def _lcs_children(
100 base_children: tuple[TreeNode, ...],
101 target_children: tuple[TreeNode, ...],
102 ) -> list[tuple[Literal["keep", "insert", "delete"], int, int]]:
103 """LCS shortest-edit script on two sequences of child nodes.
104
105 Comparison is by ``id`` — children with the same id are matched (a "keep"),
106 even if their ``content_id`` differs. A kept pair that has a different
107 ``content_id`` will produce a ``ReplaceOp`` when recursed into by
108 :func:`_diff_nodes`.
109
110 Unmatched children produce insert / delete ops.
111
112 Returns a list of ``(kind, base_idx, target_idx)`` triples.
113 """
114 n, m = len(base_children), len(target_children)
115 base_ids = [c.id for c in base_children]
116 target_ids = [c.id for c in target_children]
117
118 dp: list[list[int]] = [[0] * (m + 1) for _ in range(n + 1)]
119 for i in range(n - 1, -1, -1):
120 for j in range(m - 1, -1, -1):
121 if base_ids[i] == target_ids[j]:
122 dp[i][j] = dp[i + 1][j + 1] + 1
123 else:
124 dp[i][j] = max(dp[i + 1][j], dp[i][j + 1])
125
126 result: list[tuple[Literal["keep", "insert", "delete"], int, int]] = []
127 i, j = 0, 0
128 while i < n or j < m:
129 if i < n and j < m and base_ids[i] == target_ids[j]:
130 result.append(("keep", i, j))
131 i += 1
132 j += 1
133 elif j < m and (i >= n or dp[i][j + 1] >= dp[i + 1][j]):
134 result.append(("insert", i, j))
135 j += 1
136 else:
137 result.append(("delete", i, j))
138 i += 1
139
140 return result
141
142
143 def _diff_nodes(
144 base: TreeNode,
145 target: TreeNode,
146 *,
147 domain: str,
148 address: str,
149 ) -> list[DomainOp]:
150 """Recursively diff two tree nodes, returning a flat op list."""
151 ops: list[DomainOp] = []
152 node_addr = f"{address}/{base.id}" if address else base.id
153
154 # Root node comparison
155 if base.content_id != target.content_id:
156 ops.append(
157 ReplaceOp(
158 op="replace",
159 address=node_addr,
160 position=None,
161 old_content_id=base.content_id,
162 new_content_id=target.content_id,
163 old_summary=f"{base.label} (prev)",
164 new_summary=f"{target.label} (new)",
165 )
166 )
167
168 if not base.children and not target.children:
169 return ops
170
171 # Diff children via LCS
172 script = _lcs_children(base.children, target.children)
173
174 raw_inserts: list[tuple[int, TreeNode]] = [] # (target_idx, node)
175 raw_deletes: list[tuple[int, TreeNode]] = [] # (base_idx, node)
176
177 for kind, bi, ti in script:
178 if kind == "keep":
179 # Recurse into the matched child pair
180 ops.extend(
181 _diff_nodes(
182 base.children[bi],
183 target.children[ti],
184 domain=domain,
185 address=node_addr,
186 )
187 )
188 elif kind == "insert":
189 raw_inserts.append((ti, target.children[ti]))
190 else:
191 raw_deletes.append((bi, base.children[bi]))
192
193 # Move detection: paired insert+delete of the same node id at different positions.
194 # Node identity is tracked by id, not content_id, so a repositioned node
195 # is detected as a move even if its content also changed.
196 delete_by_id: _DeleteMap = {}
197 for bi, node in raw_deletes:
198 if node.id not in delete_by_id:
199 delete_by_id[node.id] = (bi, node)
200
201 consumed_ids: set[str] = set()
202 for ti, node in raw_inserts:
203 nid = node.id
204 if nid in delete_by_id and nid not in consumed_ids:
205 from_idx, _ = delete_by_id[nid]
206 if from_idx != ti:
207 ops.append(
208 MoveOp(
209 op="move",
210 address=node_addr,
211 from_position=from_idx,
212 to_position=ti,
213 content_id=node.content_id,
214 )
215 )
216 consumed_ids.add(nid)
217 continue
218 # True insert — recursively add the entire subtree's nodes
219 for sub_node in _subtree_nodes(node):
220 ops.append(
221 InsertOp(
222 op="insert",
223 address=node_addr,
224 position=ti,
225 content_id=sub_node.content_id,
226 content_summary=f"{sub_node.label} added",
227 )
228 )
229
230 for bi, node in raw_deletes:
231 if node.id in consumed_ids:
232 continue
233 # True delete — recursively remove the entire subtree's nodes
234 for sub_node in _subtree_nodes(node):
235 ops.append(
236 DeleteOp(
237 op="delete",
238 address=node_addr,
239 position=bi,
240 content_id=sub_node.content_id,
241 content_summary=f"{sub_node.label} removed",
242 )
243 )
244
245 return ops
246
247
248 # ---------------------------------------------------------------------------
249 # Top-level diff entry point
250 # ---------------------------------------------------------------------------
251
252
253 def diff(
254 schema: TreeSchema,
255 base: TreeNode,
256 target: TreeNode,
257 *,
258 domain: str,
259 address: str = "",
260 ) -> StructuredDelta:
261 """Diff two labeled ordered trees, returning a ``StructuredDelta``.
262
263 Produces ``ReplaceOp`` for node relabels, ``InsertOp`` / ``DeleteOp``
264 for subtree insertions and deletions, and ``MoveOp`` for repositioned
265 subtrees (detected as paired delete+insert of the same content).
266
267 Args:
268 schema: The ``TreeSchema`` declaring node type and diff algorithm.
269 base: Root of the base (ancestor) tree.
270 target: Root of the target (newer) tree.
271 domain: Domain tag for the returned ``StructuredDelta``.
272 address: Address prefix for generated op entries.
273
274 Returns:
275 A ``StructuredDelta`` with typed ops and a human-readable summary.
276 """
277 # Fast path: identical trees
278 if base.content_id == target.content_id and base.children == target.children:
279 return StructuredDelta(
280 domain=domain,
281 ops=[],
282 summary=f"no {schema['node_type']} changes",
283 )
284
285 ops = _diff_nodes(base, target, domain=domain, address=address)
286
287 n_replace = sum(1 for op in ops if op["op"] == "replace")
288 n_insert = sum(1 for op in ops if op["op"] == "insert")
289 n_delete = sum(1 for op in ops if op["op"] == "delete")
290 n_move = sum(1 for op in ops if op["op"] == "move")
291
292 parts: list[str] = []
293 if n_replace:
294 parts.append(f"{n_replace} relabelled")
295 if n_insert:
296 parts.append(f"{n_insert} added")
297 if n_delete:
298 parts.append(f"{n_delete} removed")
299 if n_move:
300 parts.append(f"{n_move} moved")
301 summary = ", ".join(parts) if parts else f"no {schema['node_type']} changes"
302
303 logger.debug(
304 "tree_edit.diff: +%d -%d ~%d r%d ops on %r",
305 n_insert,
306 n_delete,
307 n_move,
308 n_replace,
309 address,
310 )
311
312 return StructuredDelta(domain=domain, ops=ops, summary=summary)
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago