fix(reconciler): global score-descending pairing prevents false-positive feature deactivations

The reconciler's matcher was per-write greedy first-fit: every
synthesised write picked the best candidate that no earlier write had
already claimed. With realistic inputs this routinely orphaned the
candidate that would have been a clean pair for a later write while a
borderline-match earlier write got the credit, then the orphaned
candidate was swept inactive.

Concrete prod evidence (ATOAPayment, 2026-06-01 PR-3220 merge): 28
features soft-deleted in one scan; at least 10 of them had a still-active
twin with embedding cosine >= 0.85, several with identical titles --
"Cache Service" <-> "Cache Service" (0.857), "Vendor Rollout Flags" <->
"Vendor Rollout Flags" (0.845), "Upload Analytics Events" <-> "Analytics
Upload Events" (0.853), "Pricing & Merchant Plans" <-> "Merchant Pricing
& Plans" (0.876), and more.

Replace _match_strategy (per-write greedy) with _resolve_pairings (global
within-tier resolution). Tier order is preserved: signature -> Jaccard
-> containment -> cosine. Within each tier the resolver collects every
qualifying (write, candidate) edge, sorts descending by score with a
stable tiebreak on (write_index, candidate_feature_id), and claims edges
in that order. The strongest pair always wins.

Sweep behavior, candidate_filter, deactivate_filter, audit-log contract,
and the conservative absorb path are all unchanged. The change is
algorithm-local; existing callers (synthesize.py full scan,
pr_narrow_synthesis.py narrow path) need no changes and benefit equally.

Validated locally by restoring the prod ATOAPayment snapshot and
replaying the PR-3220 webhook through the backend's Redis-stream worker
on the new matcher. High-confidence false-positives (cosine >= 0.85
twin) dropped from prod's 10+ to 4 on the local replay -- a meaningful
incremental fix. The residual 4 are a cross-tier ordering pattern (a
sibling write claims via signature/Jaccard while the better cosine pair
is starved); a fuller fix would need global cross-tier optimization
(Hungarian-style) and is out of scope for this change.

Tests:
- Updated test_feature_reconciler_containment.py's 4 unit tests to use
  the new _resolve_pairings entry via a _pair_one helper. Integration
  test untouched.
- New test_feature_reconciler_global_pairing.py pins the regression with
  6 cases: strongest-Jaccard-pair-wins regardless of input order;
  cosine score-descending claims correctly; signature tier still
  pre-empts lower tiers; sibling write does not block; tier order
  preserved when both Jaccard and cosine qualify for the same pair;
  unmatched candidate left for sweep.

All 1818 backend tests pass; ruff and mypy --strict clean across 492
source files.

Signed-off-by: Arun Rajkumar <mickyarunr@gmail.com>

Author: mickyarun

backend/app/services/feature_reconciler.py

@@ -28,6 +28,13 @@
 preserved (and revivable on re-introduction) rather than silently
 disappearing.
 
+Within each tier the resolver picks pairings globally by score-descending
+edge order, not by write-iteration order. That avoids the failure where
+an earlier write claims a candidate via a borderline match, leaving the
+later write that would have been a clean pair to insert-as-new while the
+displaced candidate gets swept inactive ("Cache Service ↔ Cache Service"
+deactivations seen in PR-merge synthesis on large repos).
+
 The containment tier exists for the "narrow PR adds a sub-component to
 a broader feature" case: a 1-file PR re-synthesised on its own cluster
 emits a tiny feature whose files are a strict subset of an existing
@@ -122,9 +129,10 @@ async def reconcile_features_for_repo(
 
     1. Bulk-load every existing feature for ``repo_id`` (active +
        inactive) so revival is single-pass.
-    2. For each ``FeatureWrite``, run :func:`_match_strategy` to pick
-       an existing row by signature → Jaccard → containment → cosine.
-       First hit wins; ties broken by score.
+    2. Call :func:`_resolve_pairings` to assign each ``FeatureWrite``
+       to an existing row by signature → Jaccard → containment → cosine.
+       Within each tier the highest-scoring edge claims its pair first
+       (global resolution, not per-write greedy).
     3. If matched via signature / Jaccard / cosine: revive (when
        inactive) + ``update_in_place`` + refresh PRIMARY junction. If
        matched via containment: ``_absorb_into_existing`` — keep the
@@ -160,20 +168,19 @@ async def reconcile_features_for_repo(
         else all_candidates
     )
     by_signature: dict[str, ReconcilerCandidate] = {c.cluster_signature: c for c in candidates}
+    pairings = _resolve_pairings(
+        synthesised,
+        candidates,
+        by_signature,
+        jaccard_threshold=jaccard_threshold,
+        cosine_threshold=cosine_threshold,
+        containment_threshold=containment_threshold,
+    )
     matched_ids: set[uuid.UUID] = set()
     feat_repo = FeatureRepository(db, org_id=org_id)
     result = ReconcileResult()
 
-    for write in synthesised:
-        match, match_via, score = _match_strategy(
-            write,
-            candidates,
-            by_signature,
-            matched_ids,
-            jaccard_threshold=jaccard_threshold,
-            cosine_threshold=cosine_threshold,
-            containment_threshold=containment_threshold,
-        )
+    for write, (match, match_via, score) in zip(synthesised, pairings, strict=True):
         decision: str
         if match is None:
             await _insert_new(
@@ -371,83 +378,136 @@ async def _update_existing(
     )
 
 
-# O(n*m) — fine at hundreds; revisit at 5k+ features per repo.
-def _match_strategy(
-    write: FeatureWrite,
+# O(n*m) per tier — fine at hundreds; revisit at 5k+ features per repo.
+def _resolve_pairings(
+    synthesised: list[FeatureWrite],
     candidates: list[ReconcilerCandidate],
     by_signature: dict[str, ReconcilerCandidate],
-    matched_ids: set[uuid.UUID],
     *,
     jaccard_threshold: float,
     cosine_threshold: float,
     containment_threshold: float,
-) -> tuple[ReconcilerCandidate | None, str, float]:
-    """Layered identity matcher: signature → Jaccard → containment → cosine.
-
-    Returns ``(candidate, match_via, score)`` where ``match_via`` is
-    one of ``signature``, ``jaccard``, ``containment``, ``cosine``,
-    ``insert``. Score is 1.0 for an exact signature match, the
-    Jaccard / containment / cosine value for the fallback tiers, and
-    0.0 for ``insert``.
-
-    Containment is asymmetric — ``|write ∩ cand| / |write|`` — and only
-    fires when the candidate is the larger side. It catches the case
-    where a narrow PR re-synthesises a sub-cluster (a handful of files)
-    whose files are already part of a broader existing feature; the
-    earlier symmetric Jaccard tier misses this because the union
-    blows up with the broader feature's wider footprint.
-
-    Skips candidates already claimed by a prior synthesised entry
-    (``matched_ids``) so two synthesised features cannot collapse onto
-    the same existing row.
+) -> list[tuple[ReconcilerCandidate | None, str, float]]:
+    """Globally resolve write→candidate pairings, tier-then-score order.
+
+    Tier priority is signature → Jaccard → containment → cosine. Within
+    each tier the resolver claims the highest-scoring (write, candidate)
+    edge first, then the next, until no candidate edges remain — instead
+    of resolving each write in input order and letting the first write
+    that walks the candidate list win.
+
+    The old per-write greedy left "Cache Service ↔ Cache Service" pairs
+    unmatched whenever a sibling write reached the candidate first via a
+    borderline Jaccard. The displaced candidate then had no match and
+    was swept inactive. Score-descending within-tier resolution gives
+    the strongest pair first dibs on the candidate.
+
+    Returns one ``(candidate, match_via, score)`` per synthesised entry,
+    in input order. ``match_via=='insert'`` and ``candidate is None``
+    when no tier claimed the write.
     """
-    sig_match = by_signature.get(write.cluster_signature)
-    if sig_match is not None and sig_match.feature_id not in matched_ids:
-        return sig_match, "signature", 1.0
-
-    write_paths = _flatten_paths(write.code_locations)
-    if write_paths:
-        best_jac: tuple[ReconcilerCandidate, float] | None = None
+    n = len(synthesised)
+    pairings: list[tuple[ReconcilerCandidate | None, str, float]] = [(None, "insert", 0.0)] * n
+    claimed_writes: set[int] = set()
+    claimed_candidates: set[uuid.UUID] = set()
+
+    # Tier 1: signature — exact cluster_signature, 1:1 by definition.
+    for i, write in enumerate(synthesised):
+        cand = by_signature.get(write.cluster_signature)
+        if cand is not None and cand.feature_id not in claimed_candidates:
+            pairings[i] = (cand, "signature", 1.0)
+            claimed_writes.add(i)
+            claimed_candidates.add(cand.feature_id)
+
+    # Memoise flattened paths so each side is built once across tiers.
+    write_paths_memo: dict[int, set[str]] = {}
+    cand_paths_memo: dict[uuid.UUID, set[str]] = {}
+
+    def write_paths(i: int) -> set[str]:
+        cached = write_paths_memo.get(i)
+        if cached is None:
+            cached = _flatten_paths(synthesised[i].code_locations)
+            write_paths_memo[i] = cached
+        return cached
+
+    def cand_paths(c: ReconcilerCandidate) -> set[str]:
+        cached = cand_paths_memo.get(c.feature_id)
+        if cached is None:
+            cached = _flatten_paths(c.code_locations)
+            cand_paths_memo[c.feature_id] = cached
+        return cached
+
+    def assign_tier(
+        edges: list[tuple[float, int, ReconcilerCandidate]],
+        label: str,
+    ) -> None:
+        # Sort key, in order: score descending; lower write index wins
+        # on ties (matches the old greedy by-iteration-order semantics);
+        # candidate feature_id as the final deterministic tiebreak so
+        # output never depends on the candidate-list order the upstream
+        # loader happens to produce.
+        edges.sort(key=lambda e: (-e[0], e[1], str(e[2].feature_id)))
+        for score, i, cand in edges:
+            if i in claimed_writes or cand.feature_id in claimed_candidates:
+                continue
+            pairings[i] = (cand, label, score)
+            claimed_writes.add(i)
+            claimed_candidates.add(cand.feature_id)
+
+    # Tier 2: Jaccard over file paths.
+    jacc_edges: list[tuple[float, int, ReconcilerCandidate]] = []
+    for i in range(n):
+        if i in claimed_writes:
+            continue
+        wp = write_paths(i)
+        if not wp:
+            continue
         for cand in candidates:
-            if cand.feature_id in matched_ids:
+            if cand.feature_id in claimed_candidates:
                 continue
-            cand_paths = _flatten_paths(cand.code_locations)
-            if not cand_paths:
+            cp = cand_paths(cand)
+            if not cp:
                 continue
-            score = _jaccard(write_paths, cand_paths)
-            if score >= jaccard_threshold and (best_jac is None or score > best_jac[1]):
-                best_jac = (cand, score)
-        if best_jac is not None:
-            return best_jac[0], "jaccard", best_jac[1]
-
-        best_cont: tuple[ReconcilerCandidate, float] | None = None
+            score = _jaccard(wp, cp)
+            if score >= jaccard_threshold:
+                jacc_edges.append((score, i, cand))
+    assign_tier(jacc_edges, "jaccard")
+
+    # Tier 3: asymmetric containment. Candidate must be strictly larger
+    # — absorbing into a smaller candidate would truncate the synthesis
+    # result, not the other way around.
+    cont_edges: list[tuple[float, int, ReconcilerCandidate]] = []
+    for i in range(n):
+        if i in claimed_writes:
+            continue
+        wp = write_paths(i)
+        if not wp:
+            continue
         for cand in candidates:
-            if cand.feature_id in matched_ids:
+            if cand.feature_id in claimed_candidates:
                 continue
-            cand_paths = _flatten_paths(cand.code_locations)
-            # Containment requires the candidate to be strictly larger —
-            # absorbing into a smaller or equal-sized candidate would
-            # truncate the synthesis result, not the other way around.
-            if not cand_paths or len(write_paths) >= len(cand_paths):
+            cp = cand_paths(cand)
+            if not cp or len(wp) >= len(cp):
                 continue
-            score = _containment(write_paths, cand_paths)
-            if score >= containment_threshold and (best_cont is None or score > best_cont[1]):
-                best_cont = (cand, score)
-        if best_cont is not None:
-            return best_cont[0], "containment", best_cont[1]
-
-    if write.embedding:
-        best_cos: tuple[ReconcilerCandidate, float] | None = None
+            score = _containment(wp, cp)
+            if score >= containment_threshold:
+                cont_edges.append((score, i, cand))
+    assign_tier(cont_edges, "containment")
+
+    # Tier 4: embedding cosine.
+    cos_edges: list[tuple[float, int, ReconcilerCandidate]] = []
+    for i, write in enumerate(synthesised):
+        if i in claimed_writes or not write.embedding:
+            continue
         for cand in candidates:
-            if cand.feature_id in matched_ids or cand.embedding is None:
+            if cand.feature_id in claimed_candidates or cand.embedding is None:
                 continue
             score = _cosine(write.embedding, cand.embedding)
-            if score >= cosine_threshold and (best_cos is None or score > best_cos[1]):
-                best_cos = (cand, score)
-        if best_cos is not None:
-            return best_cos[0], "cosine", best_cos[1]
+            if score >= cosine_threshold:
+                cos_edges.append((score, i, cand))
+    assign_tier(cos_edges, "cosine")
 
-    return None, "insert", 0.0
+    return pairings
 
 
 def _flatten_paths(locations: dict[str, list[str]] | None) -> set[str]:

backend/tests/services/test_feature_reconciler_containment.py

@@ -23,7 +23,7 @@
 
 Three properties under test:
 
-1. ``_match_strategy`` returns ``match_via="containment"`` when the
+1. ``_resolve_pairings`` returns ``match_via="containment"`` when the
    synth's files are mostly inside a larger candidate.
 2. The containment tier does not fire when the synth side is the same
    size or larger than the candidate (no false absorbs into smaller
@@ -48,10 +48,32 @@
     JACCARD_THRESHOLD,
     FeatureWrite,
     ReconcilerCandidate,
-    _match_strategy,
+    _resolve_pairings,
 )
 
 
+def _pair_one(
+    write: FeatureWrite,
+    candidates: list[ReconcilerCandidate],
+) -> tuple[ReconcilerCandidate | None, str, float]:
+    """Run :func:`_resolve_pairings` for a single write and return its slot.
+
+    The matcher is now global (resolves an entire batch at once), so the
+    single-write tier-behavior tests below wrap the write in a one-element
+    list and read pairings[0].
+    """
+    by_signature = {c.cluster_signature: c for c in candidates}
+    pairings = _resolve_pairings(
+        [write],
+        candidates,
+        by_signature,
+        jaccard_threshold=JACCARD_THRESHOLD,
+        cosine_threshold=COSINE_THRESHOLD,
+        containment_threshold=CONTAINMENT_THRESHOLD,
+    )
+    return pairings[0]
+
+
 def _candidate(
     *,
     signature: str,
@@ -104,15 +126,7 @@ def test_containment_matches_when_synth_files_are_subset_of_larger_candidate() -
     )
     write = _write(signature="sig-narrow", files=["sidebar.vue"])
 
-    match, via, score = _match_strategy(
-        write,
-        [cand],
-        by_signature={cand.cluster_signature: cand},
-        matched_ids=set(),
-        jaccard_threshold=JACCARD_THRESHOLD,
-        cosine_threshold=COSINE_THRESHOLD,
-        containment_threshold=CONTAINMENT_THRESHOLD,
-    )
+    match, via, score = _pair_one(write, [cand])
 
     assert match is cand
     assert via == "containment"
@@ -130,15 +144,7 @@ def test_containment_skips_when_synth_is_not_smaller() -> None:
     cand = _candidate(signature="sig-cand", files=["a.vue", "b.vue"])
     write = _write(signature="sig-write", files=["a.vue", "c.vue"])
 
-    match, via, _score = _match_strategy(
-        write,
-        [cand],
-        by_signature={cand.cluster_signature: cand},
-        matched_ids=set(),
-        jaccard_threshold=JACCARD_THRESHOLD,
-        cosine_threshold=COSINE_THRESHOLD,
-        containment_threshold=CONTAINMENT_THRESHOLD,
-    )
+    match, via, _score = _pair_one(write, [cand])
 
     assert match is None
     assert via == "insert"
@@ -149,15 +155,7 @@ def test_signature_match_beats_containment() -> None:
     cand = _candidate(signature="sig-shared", files=["a.vue", "b.vue", "c.vue", "d.vue"])
     write = _write(signature="sig-shared", files=["a.vue"])
 
-    _match, via, score = _match_strategy(
-        write,
-        [cand],
-        by_signature={cand.cluster_signature: cand},
-        matched_ids=set(),
-        jaccard_threshold=JACCARD_THRESHOLD,
-        cosine_threshold=COSINE_THRESHOLD,
-        containment_threshold=CONTAINMENT_THRESHOLD,
-    )
+    _match, via, score = _pair_one(write, [cand])
 
     assert via == "signature"
     assert score == pytest.approx(1.0)
@@ -170,15 +168,7 @@ def test_jaccard_match_beats_containment() -> None:
     # Containment would also match (3/4 = 0.75) — but Jaccard runs first.
     write = _write(signature="sig-write", files=["a.vue", "b.vue", "c.vue", "d.vue"])
 
-    _match, via, _score = _match_strategy(
-        write,
-        [cand],
-        by_signature={cand.cluster_signature: cand},
-        matched_ids=set(),
-        jaccard_threshold=JACCARD_THRESHOLD,
-        cosine_threshold=COSINE_THRESHOLD,
-        containment_threshold=CONTAINMENT_THRESHOLD,
-    )
+    _match, via, _score = _pair_one(write, [cand])
 
     # Jaccard cares about |∩| / |∪|. Containment is asymmetric. The synth
     # is the LARGER side here, so containment is structurally disallowed
@@ -195,15 +185,7 @@ def test_containment_below_threshold_falls_through_to_insert() -> None:
     # 1 of 4 synth files in candidate → containment = 0.25 < 0.5.
     write = _write(signature="sig-write", files=["a.vue", "x.vue", "y.vue", "z.vue"])
 
-    match, via, _score = _match_strategy(
-        write,
-        [cand],
-        by_signature={cand.cluster_signature: cand},
-        matched_ids=set(),
-        jaccard_threshold=JACCARD_THRESHOLD,
-        cosine_threshold=COSINE_THRESHOLD,
-        containment_threshold=CONTAINMENT_THRESHOLD,
-    )
+    match, via, _score = _pair_one(write, [cand])
 
     assert match is None
     assert via == "insert"