fix(merge): keep max-weight tested_by edge in Pass 1 dedup (#113)

Codex P2: link_tests Pass 1 dropped duplicate (production, test) pairs
purely by arrival order — when two batches both emitted a tested_by
edge for the same pair with different confidences (0.3 vs 0.9), the
edge that happened to iterate first won. The general Step 6 deduper
at line 762 mirrors `weight > existing.weight` semantics but it only
ever saw one of the duplicates, so it couldn't rescue the heavier one.

Refactor Pass 1 to mirror Step 6's weight comparison locally:

  - Track `pair_to_idx` mapping each kept (prod, test) pair to its
    slot in the compacted edges list. On a duplicate, look up the
    existing kept edge and compare weights; if the new edge is
    strictly heavier, swap (if needed) and replace the slot. Tie or
    lighter → drop the new edge.
  - Defer the swap operation until we know an edge will survive — no
    point canonicalizing a doomed duplicate.
  - Track surviving swap pairs in a separate `swapped_pairs` set so
    the `swapped` counter reflects the FINAL output, not the wasted
    work on edges that were later replaced. This means: replacing a
    swapped edge with a heavier canonical one drops the swap from
    the count; replacing a canonical edge with a heavier swapped one
    adds it.
  - Extract the swap-in-place mutation into `_swap_tested_by_in_place`
    so it can be invoked from both code paths.

Five new unit tests cover all four weight-vs-direction combinations
plus a tie case (existing test_drops_duplicate_canonical_edges, which
still passes — tie → keep first, no swap counted).

microservices-demo regression check unchanged: 7 → 7 edges, 3 swapped,
0 dropped, 7 tagged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

understand-anything-plugin/skills/understand/merge-batch-graphs.py

@@ -487,6 +487,24 @@ def _file_node_path(node: dict[str, Any]) -> str | None:
     return nid[len("file:"):]
 
 
+def _swap_tested_by_in_place(
+    edge: dict[str, Any], original_src: str, original_tgt: str
+) -> None:
+    """Flip an inverted `tested_by` edge so source becomes production and
+    target becomes the test file. Mutates `edge` in place; appends a
+    `[direction corrected]` audit marker to `description`.
+    """
+    edge["source"] = original_tgt
+    edge["target"] = original_src
+    edge["direction"] = "forward"
+    prev = edge.get("description")
+    edge["description"] = (
+        "Direction corrected (was test → production)"
+        if not prev
+        else f"{prev} [direction corrected]"
+    )
+
+
 def _ensure_tested_tag(node: dict[str, Any]) -> bool:
     """Append "tested" to `node["tags"]`, coercing malformed `tags` to a
     fresh list. Returns True if the tag was newly added.
@@ -557,11 +575,21 @@ def link_tests(
             node_id_to_classification[node["id"]] = "prod"
 
     # ── Pass 1: walk existing tested_by edges, canonicalize or drop.
-    # `covered` tracks (production_id, test_id) pairs already represented
-    # by a tested_by edge after this pass — used to suppress duplicate
-    # supplements in Pass 2 and to deduplicate within Pass 1 itself.
+    # `covered` tracks (production_id, test_id) pairs that have a kept edge
+    # after this pass — used both to deduplicate within Pass 1 and to
+    # suppress duplicate supplements in Pass 2.
+    # `pair_to_idx` maps each kept pair to its slot in the compacted edges
+    # list, so a duplicate that arrives later with a higher weight can
+    # replace the earlier slot in place (mirrors Step 6's
+    # `weight > existing.weight` rule — without this, a 0.3-weight edge
+    # from batch 1 would silently outrank a 0.9-weight edge from batch 2
+    # because Step 6 only ever sees one of them).
+    # `swapped_pairs` records which surviving pairs came from a flipped
+    # edge, so the `swapped` counter reflects the FINAL output and
+    # doesn't double-count work done on edges that were later replaced.
     covered: set[tuple[str, str]] = set()
-    swapped = 0
+    pair_to_idx: dict[tuple[str, str], int] = {}
+    swapped_pairs: set[tuple[str, str]] = set()
     dropped = 0
     write_idx = 0
     for edge in edges:
@@ -580,35 +608,45 @@ def link_tests(
         # has no recoverable meaning — drop it.
         if (src_class, tgt_class) == ("prod", "test"):
             pair = (src, tgt)
-            if pair in covered:
-                # Duplicate canonical edge — drop the dup, keep the first.
-                dropped += 1
-                continue
-            covered.add(pair)
-            edges[write_idx] = edge
-            write_idx += 1
+            needs_swap = False
         elif (src_class, tgt_class) == ("test", "prod"):
             pair = (tgt, src)
-            if pair in covered:
-                dropped += 1
-                continue
-            covered.add(pair)
-            # Flip in place; mark provenance so reviewers can audit.
-            edge["source"] = tgt
-            edge["target"] = src
-            edge["direction"] = "forward"
-            prev = edge.get("description")
-            edge["description"] = (
-                "Direction corrected (was test → production)"
-                if not prev
-                else f"{prev} [direction corrected]"
-            )
-            swapped += 1
-            edges[write_idx] = edge
-            write_idx += 1
+            needs_swap = True
         else:
             dropped += 1
+            continue
+
+        if pair in covered:
+            # Duplicate pair: keep the heavier-weight edge (mirrors the
+            # weight-aware dedup in Step 6, which can't help here because
+            # only one of the duplicates would reach it).
+            existing_idx = pair_to_idx[pair]
+            existing = edges[existing_idx]
+            if _num(edge.get("weight", 0)) > _num(existing.get("weight", 0)):
+                # Heavier — replace existing slot. Apply the swap (or not)
+                # only on the survivor, so we never spend cycles canonicalizing
+                # an edge we're about to drop.
+                if needs_swap:
+                    _swap_tested_by_in_place(edge, src, tgt)
+                    swapped_pairs.add(pair)
+                else:
+                    # Replacement is canonical — if the previous winner came
+                    # from a swap, the surviving slot is no longer a swap.
+                    swapped_pairs.discard(pair)
+                edges[existing_idx] = edge
+            # else: existing is heavier or equal — keep it, drop the new edge.
+            dropped += 1
+            continue
+
+        if needs_swap:
+            _swap_tested_by_in_place(edge, src, tgt)
+            swapped_pairs.add(pair)
+        covered.add(pair)
+        pair_to_idx[pair] = write_idx
+        edges[write_idx] = edge
+        write_idx += 1
     del edges[write_idx:]
+    swapped = len(swapped_pairs)
 
     # ── Pass 2: path-convention supplement for tests not yet paired.
     paired_test_ids = {test_id for (_prod_id, test_id) in covered}

understand-anything-plugin/skills/understand/test_merge_batch_graphs.py

@@ -413,6 +413,122 @@ class LinkTestsTests(unittest.TestCase):
         self.assertEqual((added, dropped, tagged, swapped), (0, 1, 0, 0))
         self.assertEqual([e for e in edges if e["type"] == "tested_by"], [])
 
+    def test_dup_keeps_higher_weight_canonical(self) -> None:
+        # Two canonical tested_by edges for the same pair, weights 0.3 and
+        # 0.9. The heavier one must be kept — mirroring the weight-aware
+        # dedup at Step 6 (which never sees the discarded duplicate).
+        nodes_by_id = {
+            "file:src/foo.ts": _file_node("src/foo.ts"),
+            "file:src/foo.test.ts": _file_node("src/foo.test.ts"),
+        }
+        edges: list[dict[str, Any]] = [
+            {"source": "file:src/foo.ts", "target": "file:src/foo.test.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.3},
+            {"source": "file:src/foo.ts", "target": "file:src/foo.test.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.9},
+        ]
+        added, dropped, tagged, swapped = mbg.link_tests(nodes_by_id, edges)
+        self.assertEqual((added, dropped, swapped), (0, 1, 0))
+        tested_by_edges = [e for e in edges if e["type"] == "tested_by"]
+        self.assertEqual(len(tested_by_edges), 1)
+        self.assertEqual(tested_by_edges[0]["weight"], 0.9)
+
+    def test_dup_lighter_inverted_dropped_no_swap_counted(self) -> None:
+        # Heavier canonical first, lighter inverted second. The lighter
+        # inverted edge is dropped without being swapped — no point
+        # canonicalizing an edge that's about to die in the dedup.
+        nodes_by_id = {
+            "file:src/foo.ts": _file_node("src/foo.ts"),
+            "file:src/foo.test.ts": _file_node("src/foo.test.ts"),
+        }
+        edges: list[dict[str, Any]] = [
+            {"source": "file:src/foo.ts", "target": "file:src/foo.test.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.9},
+            {"source": "file:src/foo.test.ts", "target": "file:src/foo.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.3},
+        ]
+        added, dropped, tagged, swapped = mbg.link_tests(nodes_by_id, edges)
+        self.assertEqual((added, dropped, swapped), (0, 1, 0))
+        tested_by_edges = [e for e in edges if e["type"] == "tested_by"]
+        self.assertEqual(len(tested_by_edges), 1)
+        self.assertEqual(tested_by_edges[0]["weight"], 0.9)
+        # Surviving edge is the original canonical — no audit marker.
+        self.assertNotIn(
+            "direction corrected",
+            (tested_by_edges[0].get("description") or "").lower(),
+        )
+
+    def test_dup_replaces_with_heavier_inverted(self) -> None:
+        # Lighter canonical first, heavier inverted second. The inverted
+        # edge gets swapped AND replaces the kept slot, since it's heavier.
+        nodes_by_id = {
+            "file:src/foo.ts": _file_node("src/foo.ts"),
+            "file:src/foo.test.ts": _file_node("src/foo.test.ts"),
+        }
+        edges: list[dict[str, Any]] = [
+            {"source": "file:src/foo.ts", "target": "file:src/foo.test.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.3},
+            {"source": "file:src/foo.test.ts", "target": "file:src/foo.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.9},
+        ]
+        added, dropped, tagged, swapped = mbg.link_tests(nodes_by_id, edges)
+        self.assertEqual(added, 0)
+        self.assertEqual(dropped, 1)
+        self.assertEqual(swapped, 1)  # surviving edge IS a swap
+        tested_by_edges = [e for e in edges if e["type"] == "tested_by"]
+        self.assertEqual(len(tested_by_edges), 1)
+        edge = tested_by_edges[0]
+        self.assertEqual(edge["source"], "file:src/foo.ts")
+        self.assertEqual(edge["target"], "file:src/foo.test.ts")
+        self.assertEqual(edge["weight"], 0.9)
+        self.assertIn("direction corrected", edge["description"].lower())
+
+    def test_dup_swapped_then_canonical_heavier_clears_swapped_count(self) -> None:
+        # Inverted lighter first (swap is applied, swapped_pairs={pair}),
+        # then canonical heavier replaces — the surviving edge is canonical
+        # so `swapped` must drop back to 0.
+        nodes_by_id = {
+            "file:src/foo.ts": _file_node("src/foo.ts"),
+            "file:src/foo.test.ts": _file_node("src/foo.test.ts"),
+        }
+        edges: list[dict[str, Any]] = [
+            {"source": "file:src/foo.test.ts", "target": "file:src/foo.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.3},
+            {"source": "file:src/foo.ts", "target": "file:src/foo.test.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.9},
+        ]
+        added, dropped, tagged, swapped = mbg.link_tests(nodes_by_id, edges)
+        self.assertEqual(added, 0)
+        self.assertEqual(dropped, 1)
+        self.assertEqual(swapped, 0)  # surviving edge is canonical, not a swap
+        tested_by_edges = [e for e in edges if e["type"] == "tested_by"]
+        self.assertEqual(len(tested_by_edges), 1)
+        self.assertEqual(tested_by_edges[0]["weight"], 0.9)
+
+    def test_dup_two_inverted_keeps_heavier_swapped_once(self) -> None:
+        # Both inverted, different weights. The heavier one wins the slot
+        # after both get swapped; `swapped` reflects the surviving edge,
+        # not the wasted swap on the dropped lighter one.
+        nodes_by_id = {
+            "file:src/foo.ts": _file_node("src/foo.ts"),
+            "file:src/foo.test.ts": _file_node("src/foo.test.ts"),
+        }
+        edges: list[dict[str, Any]] = [
+            {"source": "file:src/foo.test.ts", "target": "file:src/foo.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.3},
+            {"source": "file:src/foo.test.ts", "target": "file:src/foo.ts",
+             "type": "tested_by", "direction": "forward", "weight": 0.9},
+        ]
+        added, dropped, tagged, swapped = mbg.link_tests(nodes_by_id, edges)
+        self.assertEqual(added, 0)
+        self.assertEqual(dropped, 1)
+        self.assertEqual(swapped, 1)
+        tested_by_edges = [e for e in edges if e["type"] == "tested_by"]
+        self.assertEqual(len(tested_by_edges), 1)
+        edge = tested_by_edges[0]
+        self.assertEqual(edge["weight"], 0.9)
+        self.assertIn("direction corrected", edge["description"].lower())
+
     def test_drops_duplicate_canonical_edges(self) -> None:
         # Two LLM edges describing the same (production, test) pair — keep
         # one, drop the other.