From 8a89253fecff4e3121b7eee1f21b51a5aaed40eb Mon Sep 17 00:00:00 2001 From: himanshupatro-334 Date: Fri, 11 Sep 2026 23:52:47 +0530 Subject: [PATCH] Scope deduplication during incremental merges (cherry picked from commit 9e65a7199319f101db2d192a4f12a7adcbcba2a6) --- graphify/build.py | 19 +- graphify/dedup.py | 154 +++++++++-- tests/test_build_merge_dedup_scope.py | 385 ++++++++++++++++++++++++++ 3 files changed, 540 insertions(+), 18 deletions(-) create mode 100644 tests/test_build_merge_dedup_scope.py diff --git a/graphify/build.py b/graphify/build.py index 30b60279b..c553ab043 100644 --- a/graphify/build.py +++ b/graphify/build.py @@ -1402,6 +1402,7 @@ def build( dedup: bool = True, dedup_llm_backend: str | None = None, root: str | Path | None = None, + protected_ids: "set[str] | None" = None, ) -> nx.Graph: """Merge multiple extraction results into one graph. @@ -1411,6 +1412,8 @@ def build( dedup_llm_backend: if set (e.g. "gemini", "claude", or "kimi"), uses LLM to resolve ambiguous pairs in the 75–92 Jaro-Winkler score zone. root: if given, absolute source_file paths are made relative to root (#932). + protected_ids: optional set of node IDs to protect from being collapsed with + other protected nodes during incremental merge (#3477). With dedup disabled, extractions are merged in order and the last node's attributes win (NetworkX add_node overwrites). With dedup enabled, nodes @@ -1454,6 +1457,7 @@ def build( # Hyperedge members reference node ids too, so they need the same # survivor rewiring the edges get (#2805). hyperedges=combined.get("hyperedges"), + protected_ids=protected_ids, ) return build_from_json(combined, directed=directed, root=_root) @@ -2027,8 +2031,21 @@ def build_merge( if had_graph else [] ) + # Untouched existing nodes must not be collapsed with each other during dedup (#3477). + _protected_ids = { + n["id"] for n in existing_nodes + if isinstance(n, dict) and n.get("id") + } if had_graph else None + all_chunks = base + list(new_chunks) - G = build(all_chunks, directed=directed, dedup=dedup, dedup_llm_backend=dedup_llm_backend, root=_eff_root) + G = build( + all_chunks, + directed=directed, + dedup=dedup, + dedup_llm_backend=dedup_llm_backend, + root=_eff_root, + protected_ids=_protected_ids, + ) # Prune nodes and edges from deleted source files if prune_sources: diff --git a/graphify/dedup.py b/graphify/dedup.py index a5ed6dbfc..e609053f3 100644 --- a/graphify/dedup.py +++ b/graphify/dedup.py @@ -555,6 +555,7 @@ def deduplicate_entities( dedup_llm_backend: str | None = None, root: str | Path | None = None, hyperedges: "list[dict] | None" = None, + protected_ids: "set[str] | None" = None, ) -> tuple[list[dict], list[dict]]: """Deduplicate near-identical entities in a knowledge graph. @@ -654,6 +655,22 @@ def deduplicate_entities( uf = _UF() exact_merges = 0 + + protected_set: set[str] = set(protected_ids) if protected_ids is not None else set() + prot_by_root: dict[str, str] = {pid: pid for pid in protected_set} + + def _get_prot(nid: str) -> str | None: + return prot_by_root.get(uf.find(nid)) + + def _union_with_prot(x: str, y: str) -> None: + px = _get_prot(x) + py = _get_prot(y) + uf.union(x, y) + new_root = uf.find(x) + prot = px or py + if prot is not None: + prot_by_root[new_root] = prot + for key, group in norm_to_nodes.items(): if len(group) <= 1: continue @@ -671,10 +688,33 @@ def deduplicate_entities( # collapsing distinct nodes that happen to share a label (#1178). continue if len(file_group) > 1: - winner = _pick_winner(file_group) - for node in file_group: - uf.union(winner["id"], node["id"]) - exact_merges += len(file_group) - 1 + if protected_set: + prot_file = [n for n in file_group if n.get("id") in protected_set] + inc_file = [n for n in file_group if n.get("id") not in protected_set] + if prot_file and not inc_file: + # All nodes belong exclusively to an untouched file — preserve all of them + continue + if prot_file and inc_file: + winner = _pick_winner(prot_file) + for node in inc_file: + px = _get_prot(winner["id"]) + py = _get_prot(node["id"]) + if px is not None and py is not None and px != py: + continue + if uf.find(winner["id"]) != uf.find(node["id"]): + _union_with_prot(winner["id"], node["id"]) + exact_merges += 1 + else: + winner = _pick_winner(file_group) + for node in file_group: + if uf.find(winner["id"]) != uf.find(node["id"]): + _union_with_prot(winner["id"], node["id"]) + exact_merges += 1 + else: + winner = _pick_winner(file_group) + for node in file_group: + uf.union(winner["id"], node["id"]) + exact_merges += len(file_group) - 1 # Cross-file residue: union exact matches across files, but only where # it is provably safe (#2182). `concept` is the one file_type meant to # unify across files (#1284) — code is keyed by ID (#1205) and @@ -703,11 +743,38 @@ def deduplicate_entities( key=lambda n: n["id"], ) if len(mergeable) > 1: - winner = _pick_winner(mergeable) - for node in mergeable: - if uf.find(winner["id"]) != uf.find(node["id"]): - uf.union(winner["id"], node["id"]) - exact_merges += 1 + if protected_set: + prot_members = [n for n in mergeable if n.get("id") in protected_set] + inc_members = [n for n in mergeable if n.get("id") not in protected_set] + if not inc_members: + # All participants belong exclusively to untouched files (#3477): + # NEVER collapse them during incremental merge. + continue + if prot_members: + # Mixed: pick AT MOST ONE protected survivor for incoming nodes to fold into. + # Multiple protected nodes must remain separate independent entities. + canonical_winner = _pick_winner(prot_members) + for inc in inc_members: + px = _get_prot(canonical_winner["id"]) + py = _get_prot(inc["id"]) + if px is not None and py is not None and px != py: + continue + if uf.find(canonical_winner["id"]) != uf.find(inc["id"]): + _union_with_prot(canonical_winner["id"], inc["id"]) + exact_merges += 1 + else: + # Incoming only: merge normally + winner = _pick_winner(inc_members) + for node in inc_members: + if uf.find(winner["id"]) != uf.find(node["id"]): + _union_with_prot(winner["id"], node["id"]) + exact_merges += 1 + else: + winner = _pick_winner(mergeable) + for node in mergeable: + if uf.find(winner["id"]) != uf.find(node["id"]): + uf.union(winner["id"], node["id"]) + exact_merges += 1 # ── pass 2: MinHash/LSH + Jaro-Winkler (high-entropy nodes only) ───────── candidates: list[dict] = [] @@ -757,6 +824,13 @@ def deduplicate_entities( if uf.find(node_id) == uf.find(neighbor_id): continue + if protected_set: + px = _get_prot(node_id) + py = _get_prot(neighbor_id) + if px is not None and py is not None and px != py: + # Prevent protected/protected unions and bridging across protected components + continue + neighbor = candidates_by_id.get(neighbor_id) if neighbor is None: continue @@ -826,14 +900,31 @@ def deduplicate_entities( # from the union of both normalized-label groups pulls # never-compared nodes (same label, different source_file) # into the merge, bypassing the #1046/#1178 guards. - winner = _pick_winner([node, neighbor]) - uf.union(winner["id"], node_id) - uf.union(winner["id"], neighbor_id) + if protected_set: + px = _get_prot(node_id) + py = _get_prot(neighbor_id) + if px is not None and py is not None and px != py: + continue + if node_id in protected_set: + winner = node + elif neighbor_id in protected_set: + winner = neighbor + else: + winner = _pick_winner([node, neighbor]) + _union_with_prot(winner["id"], node_id) + _union_with_prot(winner["id"], neighbor_id) + else: + winner = _pick_winner([node, neighbor]) + uf.union(winner["id"], node_id) + uf.union(winner["id"], neighbor_id) fuzzy_merges += 1 # ── pass 3: LLM tiebreaker for ambiguous pairs (opt-in) ────────────────── if dedup_llm_backend is not None: - _llm_tiebreak(candidates, uf, communities, backend=dedup_llm_backend) + _llm_tiebreak( + candidates, uf, communities, backend=dedup_llm_backend, + protected_set=protected_set, get_prot=_get_prot, union_with_prot=_union_with_prot, + ) # ── build remap table from union-find components ────────────────────────── components = uf.components() @@ -861,7 +952,14 @@ def deduplicate_entities( key=lambda pair: pair[0], ) ] - winner = _pick_winner(group_nodes) if group_nodes else {"id": root} + if protected_set: + prot_in_group = [n for n in group_nodes if n.get("id") in protected_set] + if prot_in_group: + winner = _pick_winner(prot_in_group) + else: + winner = _pick_winner(group_nodes) if group_nodes else {"id": root} + else: + winner = _pick_winner(group_nodes) if group_nodes else {"id": root} winner_id = winner["id"] # Even with the right survivor, dropping the losers wholesale loses # whatever fields only they carried. Fill the survivor's absent @@ -991,6 +1089,9 @@ def _llm_tiebreak( batch_size: int = 30, low: float = 75.0, high: float = 92.0, + protected_set: set[str] | None = None, + get_prot=None, + union_with_prot=None, ) -> None: """Batch-resolve ambiguous pairs (score in [low, high)) via LLM.""" try: @@ -1040,6 +1141,11 @@ def _llm_tiebreak( and min(len(norm_i), len(norm_j)) >= 12): score += _COMMUNITY_BOOST if low <= score < high: + if protected_set and get_prot is not None: + px = get_prot(node["id"]) + py = get_prot(neighbor["id"]) + if px is not None and py is not None and px != py: + continue ambiguous.append((node, neighbor, score)) if not ambiguous: @@ -1086,8 +1192,22 @@ def _llm_tiebreak( answer = parts[1].strip().lower() if answer.startswith("yes"): a, b, _ = batch[idx] - winner = _pick_winner([a, b]) - uf.union(winner["id"], a["id"]) - uf.union(winner["id"], b["id"]) + if protected_set and get_prot is not None and union_with_prot is not None: + px = get_prot(a["id"]) + py = get_prot(b["id"]) + if px is not None and py is not None and px != py: + continue + if a["id"] in protected_set: + winner = a + elif b["id"] in protected_set: + winner = b + else: + winner = _pick_winner([a, b]) + union_with_prot(winner["id"], a["id"]) + union_with_prot(winner["id"], b["id"]) + else: + winner = _pick_winner([a, b]) + uf.union(winner["id"], a["id"]) + uf.union(winner["id"], b["id"]) except Exception as exc: print(f"[graphify] --dedup-llm batch failed: {exc}", flush=True) diff --git a/tests/test_build_merge_dedup_scope.py b/tests/test_build_merge_dedup_scope.py new file mode 100644 index 000000000..341ec1e6f --- /dev/null +++ b/tests/test_build_merge_dedup_scope.py @@ -0,0 +1,385 @@ +"""Regression tests for Graphify issue #3477: +Scoped entity deduplication during incremental build_merge(). + +Ensures: +1. Untouched existing nodes are protected from collapsing with other untouched existing nodes. +2. Incoming nodes can still merge into an untouched node, with the untouched node surviving as canonical. +3. Incoming nodes that match multiple untouched nodes attach to at most one protected survivor without + transitively collapsing the protected nodes. +4. Incoming duplicates deduplicate normally among themselves. +5. Full build() with protected_ids=None retains existing global dedup behavior. +6. Fuzzy dedup respects the same protected invariants. +""" +import json +from pathlib import Path +import pytest +import networkx as nx + +import graphify.build as buildmod +from graphify.build import build, build_merge +from graphify.dedup import deduplicate_entities + + +def _write_graph(graph_path: Path, nodes, edges=(), hyperedges=()) -> None: + graph_path.parent.mkdir(parents=True, exist_ok=True) + graph_path.write_text( + json.dumps( + { + "nodes": list(nodes), + "edges": list(edges), + "hyperedges": list(hyperedges), + } + ), + encoding="utf-8", + ) + + +def test_untouched_duplicate_nodes_survive_incremental_merge(tmp_path): + """#3477: Two duplicate-labeled nodes in two untouched files must not be collapsed + when an unrelated third file is incrementally updated with dedup=True.""" + gp = tmp_path / "graphify-out" / "graph.json" + node_a = { + "id": "a_auth_service", + "label": "Authentication Service Component", + "file_type": "concept", + "source_file": "a.md", + "_origin": "semantic", + } + node_b = { + "id": "b_auth_service", + "label": "Authentication Service Component", + "file_type": "concept", + "source_file": "b.md", + "_origin": "semantic", + } + _write_graph(gp, [node_a, node_b]) + + # Incremental update touches unrelated c.md + chunk_c = { + "nodes": [ + { + "id": "c_worker", + "label": "Background Worker Job", + "file_type": "concept", + "source_file": "c.md", + "_origin": "semantic", + } + ], + "edges": [], + } + + G = build_merge([chunk_c], gp, dedup=True) + + # Both untouched nodes must survive with their original IDs + assert "a_auth_service" in G + assert "b_auth_service" in G + assert "c_worker" in G + assert G.number_of_nodes() == 3 + + +def test_incoming_duplicate_merges_into_untouched_node_as_canonical_survivor(tmp_path): + """#3477: An incoming entity duplicate merges into an untouched entity, and the + untouched node MUST be the canonical survivor, with edges rewired.""" + gp = tmp_path / "graphify-out" / "graph.json" + untouched_cache = { + "id": "a_cache_mgr", + "label": "Memory Cache Manager System", + "file_type": "concept", + "source_file": "a.md", + "attributes": {"tier": "primary"}, + "_origin": "semantic", + } + _write_graph(gp, [untouched_cache]) + + # Incoming extraction from changed b.md has a richer duplicate + incoming_cache = { + "id": "b_cache_mgr", # shorter ID, would normally win on tiebreak + "label": "Memory Cache Manager System", + "file_type": "concept", + "source_file": "b.md", + "summary": "Distributed memory cache manager for session state", + "_origin": "semantic", + } + incoming_caller = { + "id": "b_client", + "label": "Cache Client Worker", + "file_type": "concept", + "source_file": "b.md", + "_origin": "semantic", + } + call_edge = { + "source": "b_client", + "target": "b_cache_mgr", + "relation": "calls", + "confidence": "EXTRACTED", + "source_file": "b.md", + } + chunk_b = { + "nodes": [incoming_cache, incoming_caller], + "edges": [call_edge], + } + + G = build_merge([chunk_b], gp, dedup=True) + + # The untouched node MUST survive as the canonical ID + assert "a_cache_mgr" in G + assert "b_cache_mgr" not in G + assert "b_client" in G + + # Missing fields from incoming loser should be merged into survivor + survivor_attrs = G.nodes["a_cache_mgr"] + assert survivor_attrs.get("summary") == "Distributed memory cache manager for session state" + + # Edge must be rewired to the untouched survivor + assert G.has_edge("b_client", "a_cache_mgr") + + +def test_two_untouched_plus_one_incoming_duplicate_bridge_case(tmp_path): + """#3477: When two untouched duplicate nodes exist and an incoming node shares the same label: + - Both untouched nodes survive as separate entities. + - Incoming node resolves to at most one protected survivor. + - The two protected nodes do not become transitively connected.""" + gp = tmp_path / "graphify-out" / "graph.json" + node_a = { + "id": "a_auth", + "label": "Authentication Service Gateway", + "file_type": "concept", + "source_file": "a.md", + "_origin": "semantic", + } + node_b = { + "id": "b_auth", + "label": "Authentication Service Gateway", + "file_type": "concept", + "source_file": "b.md", + "_origin": "semantic", + } + _write_graph(gp, [node_a, node_b]) + + # Incoming extraction from c.md introduces another duplicate + node_c = { + "id": "c_auth", + "label": "Authentication Service Gateway", + "file_type": "concept", + "source_file": "c.md", + "_origin": "semantic", + } + edge_c = { + "source": "c_client", + "target": "c_auth", + "relation": "uses", + "confidence": "EXTRACTED", + "source_file": "c.md", + } + chunk_c = { + "nodes": [ + node_c, + { + "id": "c_client", + "label": "Gateway Client", + "file_type": "concept", + "source_file": "c.md", + "_origin": "semantic", + }, + ], + "edges": [edge_c], + } + + G = build_merge([chunk_c], gp, dedup=True) + + # BOTH untouched nodes must survive + assert "a_auth" in G + assert "b_auth" in G + + # Incoming node was folded into one of the protected nodes + assert "c_auth" not in G + # Total nodes: 2 protected + 1 client = 3 nodes + assert G.number_of_nodes() == 3 + + +def test_two_incoming_duplicate_nodes_deduplicate_normally(tmp_path): + """#3477: Multiple incoming duplicates still deduplicate normally during build_merge.""" + gp = tmp_path / "graphify-out" / "graph.json" + untouched_node = { + "id": "a_existing", + "label": "Existing Untouched Component", + "file_type": "concept", + "source_file": "a.md", + "_origin": "semantic", + } + _write_graph(gp, [untouched_node]) + + # Re-extracting / adding new files b.md and c.md that both emit a shared concept + chunk_b = { + "nodes": [ + { + "id": "b_telemetry", + "label": "Telemetry Event Dispatcher", + "file_type": "concept", + "source_file": "b.md", + "_origin": "semantic", + } + ], + "edges": [], + } + chunk_c = { + "nodes": [ + { + "id": "c_telemetry", + "label": "Telemetry Event Dispatcher", + "file_type": "concept", + "source_file": "c.md", + "_origin": "semantic", + } + ], + "edges": [], + } + + G = build_merge([chunk_b, chunk_c], gp, dedup=True) + + # Untouched survives + assert "a_existing" in G + # Incoming nodes collapsed into one + surviving_telemetry = [n for n in G.nodes if "telemetry" in n] + assert len(surviving_telemetry) == 1 + assert G.number_of_nodes() == 2 + + +def test_full_build_protected_ids_none_retains_global_dedup(): + """#3477: When build() is called without protected_ids (cold build), normal global dedup occurs.""" + chunk_a = { + "nodes": [ + { + "id": "a_auth", + "label": "Authentication Service Gateway", + "file_type": "concept", + "source_file": "a.md", + "_origin": "semantic", + } + ], + "edges": [], + } + chunk_b = { + "nodes": [ + { + "id": "b_auth", + "label": "Authentication Service Gateway", + "file_type": "concept", + "source_file": "b.md", + "_origin": "semantic", + } + ], + "edges": [], + } + + # Full build with protected_ids=None collapses both into 1 node + G = build([chunk_a, chunk_b], dedup=True, protected_ids=None) + assert G.number_of_nodes() == 1 + + +def test_fuzzy_dedup_protected_nodes_do_not_merge(tmp_path): + """#3477: Pass 2 fuzzy dedup must not collapse two near-identical nodes in untouched files.""" + gp = tmp_path / "graphify-out" / "graph.json" + # Near-identical labels that clear the Jaro threshold + node_a = { + "id": "a_auth_system", + "label": "Authentication Manager Processing Engine", + "file_type": "concept", + "source_file": "a.md", + "_origin": "semantic", + } + node_b = { + "id": "b_auth_system", + "label": "Authentication Manager Processng Engine", # typo on token >= 6 chars + "file_type": "concept", + "source_file": "b.md", + "_origin": "semantic", + } + _write_graph(gp, [node_a, node_b]) + + chunk_c = { + "nodes": [ + { + "id": "c_unrelated", + "label": "Unrelated Database Connector", + "file_type": "concept", + "source_file": "c.md", + "_origin": "semantic", + } + ], + "edges": [], + } + + G = build_merge([chunk_c], gp, dedup=True) + + # Both untouched nodes must survive + assert "a_auth_system" in G + assert "b_auth_system" in G + assert G.number_of_nodes() == 3 + + +def test_fuzzy_dedup_incoming_merges_into_protected_as_canonical(tmp_path): + """#3477: Pass 2 fuzzy dedup merges incoming typo into untouched canonical node.""" + gp = tmp_path / "graphify-out" / "graph.json" + untouched = { + "id": "a_auth_system", + "label": "Authentication Manager Processing Engine", + "file_type": "concept", + "source_file": "a.md", + "_origin": "semantic", + } + _write_graph(gp, [untouched]) + + incoming_typo = { + "id": "b_auth_system", + "label": "Authentication Manager Processng Engine", + "file_type": "concept", + "source_file": "b.md", + "_origin": "semantic", + } + chunk_b = {"nodes": [incoming_typo], "edges": []} + + G = build_merge([chunk_b], gp, dedup=True) + + # Untouched survives as canonical winner + assert "a_auth_system" in G + assert "b_auth_system" not in G + assert G.number_of_nodes() == 1 + + +def test_fuzzy_dedup_incoming_bridge_case(tmp_path): + """#3477: In fuzzy dedup, an incoming node must not bridge two untouched nodes.""" + gp = tmp_path / "graphify-out" / "graph.json" + node_a = { + "id": "a_cluster", + "label": "Distributed Storage Processing Cluster Subsystem", + "file_type": "concept", + "source_file": "a.md", + "_origin": "semantic", + } + node_b = { + "id": "b_cluster", + "label": "Distributed Storage Processng Cluster Subsystem", + "file_type": "concept", + "source_file": "b.md", + "_origin": "semantic", + } + _write_graph(gp, [node_a, node_b]) + + # Incoming node in c.md matching a.md + node_c = { + "id": "c_cluster", + "label": "Distributed Storage Processing Cluster Subsystem", + "file_type": "concept", + "source_file": "c.md", + "_origin": "semantic", + } + chunk_c = {"nodes": [node_c], "edges": []} + + G = build_merge([chunk_c], gp, dedup=True) + + # Both untouched nodes must survive + assert "a_cluster" in G + assert "b_cluster" in G + assert "c_cluster" not in G + assert G.number_of_nodes() == 2