fix(consolidation): interleave fusion for dedup recall so the twin is always shown

The residual near-duplicate observations were recall misses, not LLM errors:
the existing near-identical observation (the merge 'twin') is semantic rank #1
for the new fact, but shares no source-fact graph link and little lexical
overlap, so RRF averaged it below the per-fact recall budget (512 tokens, ~8
observations) and the LLM never saw it -> created a duplicate. The cross-encoder
demoted it the same way (semantic #1 -> reranked #37).

Add round-robin interleave fusion: take each retrieval arm's #1, then each #2,
... (semantic, bm25, graph, temporal), de-duplicating, until the budget fills.
This guarantees every arm's top hit a slot, so the semantic-#1 twin is always
shown and the LLM UPDATEs instead of creating a duplicate.

Replace the recall 'rerank: bool' / ad-hoc passthrough with a single named
'reranking' strategy: 'cross_encoder' (default), 'rrf', 'interleave'.
Consolidation dedup recall uses 'interleave'. For interleave the fusion order is
authoritative — combined scoring's recency/temporal re-sort is skipped (that
re-sort is what buried the twin).

Measured on an English translation of the hermes transcript (removes the
cross-lingual embedding confound): near-dup observations 4% (@0.97) -> 0% at both
1/10 and 1/4 cuts, coverage 89% -> 94%, no false merges (heavy observations are
coherent single themes; distinct facets stay separate).

Author: nicoloboschi

hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py

@@ -1501,12 +1501,13 @@ async def _find_related_observations(
             include_source_facts=True,  # Embed source facts so we avoid a separate DB fetch
             max_source_facts_tokens=config.consolidation_source_facts_max_tokens,
             max_source_facts_tokens_per_observation=config.consolidation_source_facts_max_tokens_per_observation,
-            # Rank by RRF, skipping the cross-encoder reranker: for consolidation we are
-            # looking for an existing near-identical observation to merge into, and the
-            # cross-encoder was observed to demote that twin far below the budget cutoff
-            # (semantic rank #1 -> reranked #37). RRF keeps strong lexical/semantic
-            # matches near the top so the LLM is shown the twin and UPDATEs it.
-            rerank=False,
+            # Round-robin interleave fusion (no cross-encoder): consolidation is looking
+            # for an existing near-identical observation to merge into. Both the
+            # cross-encoder (semantic #1 -> reranked #37) and RRF (semantic #1 -> outside
+            # the 512-token budget) were measured to bury that twin; interleave guarantees
+            # each retrieval arm's top hits a slot, so the semantic-#1 twin is always shown
+            # to the LLM, which then UPDATEs instead of creating a duplicate.
+            reranking="interleave",
             _quiet=True,  # Suppress logging
         )
     finally:

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -255,6 +255,15 @@ def validate_sql_schema(sql: str) -> None:
 from .search.tags import TagGroup, TagsMatch, build_tag_groups_where_clause, build_tags_where_clause
 from .search.types import ScoredResult
 from .task_backend import TaskBackend
+
+# Recall ranking strategy: how the per-arm (semantic/bm25/graph/temporal) results are
+# fused and reranked into the final order.
+#   "cross_encoder" — RRF fusion + cross-encoder rerank (default, user-facing recall).
+#   "rrf"           — RRF fusion, no cross-encoder (RRF score is the order).
+#   "interleave"    — round-robin interleave fusion, no cross-encoder. Guarantees each
+#                     arm's top hits a slot (used by consolidation dedup recall, where RRF
+#                     buried the near-identical twin below budget). See interleave_fusion.
+RecallReranking = Literal["cross_encoder", "rrf", "interleave"]
 from .token_encoding import get_token_encoding
 
 RetainOutboxCallback = Callable[[asyncpg.Connection], Awaitable[None]]
@@ -3186,7 +3195,7 @@ async def recall_async(
         created_before: datetime | None = None,
         _connection_budget: int | None = None,
         _quiet: bool = False,
-        rerank: bool = True,
+        reranking: RecallReranking = "cross_encoder",
     ) -> RecallResultModel:
         """
         Recall memories using N*4-way parallel retrieval (N fact types × 4 retrieval methods).
@@ -3343,7 +3352,7 @@ async def recall_async(
                             include_source_facts=include_source_facts,
                             max_source_facts_tokens=max_source_facts_tokens,
                             max_source_facts_tokens_per_observation=max_source_facts_tokens_per_observation,
-                            rerank=rerank,
+                            reranking=reranking,
                         )
                         break  # Success - exit retry loop
                     except Exception as e:
@@ -3475,7 +3484,7 @@ async def _search_with_retries(
         include_source_facts: bool = False,
         max_source_facts_tokens: int = 4096,
         max_source_facts_tokens_per_observation: int = -1,
-        rerank: bool = True,
+        reranking: RecallReranking = "cross_encoder",
     ) -> RecallResultModel:
         """
         Search implementation with modular retrieval and reranking.
@@ -3792,9 +3801,13 @@ def to_tuple_format(results):
                     if _dur > 0:
                         tracer.add_phase_metric(f"retrieval_{_method}", _dur)
 
-            # Step 3: Merge with RRF
+            # Step 3: Merge ranked lists. RRF by default; interleave (round-robin) when
+            # requested by consolidation dedup recall — RRF averages a strong-in-one-arm
+            # result down and buried the near-identical "twin" observation below budget
+            # (semantic #1 -> outside the shown set), whereas interleave guarantees each
+            # arm's top hits a slot. See interleave_fusion docstring.
             step_start = time.time()
-            from .search.fusion import reciprocal_rank_fusion
+            from .search.fusion import interleave_fusion, reciprocal_rank_fusion
 
             fusion_span = tracer_otel.start_span("hindsight.recall_fusion")
             fusion_span.set_attribute("hindsight.bank_id", bank_id)
@@ -3805,16 +3818,16 @@ def to_tuple_format(results):
 
             try:
                 # Merge 3 or 4 result lists depending on temporal constraint
+                result_lists = [semantic_results, bm25_results, graph_results]
                 if temporal_results:
-                    merged_candidates = reciprocal_rank_fusion(
-                        [semantic_results, bm25_results, graph_results, temporal_results]
-                    )
-                else:
-                    merged_candidates = reciprocal_rank_fusion([semantic_results, bm25_results, graph_results])
+                    result_lists.append(temporal_results)
+                fuse = interleave_fusion if reranking == "interleave" else reciprocal_rank_fusion
+                merged_candidates = fuse(result_lists)
 
                 step_duration = time.time() - step_start
                 log_buffer.append(
-                    f"  [3] RRF merge: {len(merged_candidates)} unique candidates in {step_duration:.3f}s"
+                    f"  [3] {'interleave' if reranking == 'interleave' else 'RRF'} merge: "
+                    f"{len(merged_candidates)} unique candidates in {step_duration:.3f}s"
                 )
             finally:
                 fusion_span.set_attribute("hindsight.merged_count", len(merged_candidates))
@@ -3849,18 +3862,17 @@ def to_tuple_format(results):
                     pre_filtered_count = len(merged_candidates) - reranker_max_candidates
                     merged_candidates = merged_candidates[:reranker_max_candidates]
 
-                if rerank:
+                if reranking == "cross_encoder":
                     # Ensure reranker is initialized (for lazy initialization mode)
                     await reranker_instance.ensure_initialized()
                     scored_results = await reranker_instance.rerank(query, merged_candidates)
                 else:
-                    # RRF-only: skip the cross-encoder and rank by RRF score. Used by
-                    # consolidation recall, where the cross-encoder was observed to demote
-                    # a near-identical existing observation (the dedup "twin") far below the
-                    # budget cutoff (semantic rank #1 -> reranked #37), causing the LLM to
-                    # never see it and create a duplicate. Passthrough ScoredResults keep the
-                    # downstream recency/temporal boosts working, seeded from RRF rank.
-                    rerank_kind = "rrf-passthrough"
+                    # "rrf" / "interleave": skip the cross-encoder and keep the fusion order
+                    # (rrf_score is descending by fusion position for both). The cross-encoder
+                    # was observed to demote a near-identical existing observation (the dedup
+                    # "twin") far below the budget cutoff (semantic rank #1 -> reranked #37),
+                    # causing the LLM to never see it and create a duplicate.
+                    rerank_kind = f"{reranking}-passthrough"
                     scored_results = [
                         ScoredResult(
                             candidate=mc,
@@ -3888,10 +3900,18 @@ def to_tuple_format(results):
             # is_passthrough_reranker tells the scoring code to seed CE scores
             # from RRF rank — only meaningful when the configured reranker is
             # the slim/passthrough one that returns a constant score per pair.
-            if scored_results:
+            if scored_results and reranking == "interleave":
+                # Interleave order is authoritative for dedup recall: do NOT re-sort by the
+                # recency/temporal boosts — that re-sort is precisely what buried the twin
+                # under RRF. Seed weight from the interleave-position rrf_score so the order
+                # survives Step 5 truncation and the Step 6 token-budget cut.
+                for sr in scored_results:
+                    sr.weight = sr.candidate.rrf_score
+                log_buffer.append("  [4.6] Interleave order preserved (combined scoring skipped)")
+            elif scored_results:
                 ce = reranker_instance.cross_encoder
-                # RRF-only mode is passthrough by construction; so is a configured "rrf" CE.
-                is_passthrough = (not rerank) or (ce is not None and ce.provider_name == "rrf")
+                # "rrf" mode is passthrough by construction; so is a configured "rrf" CE.
+                is_passthrough = (reranking == "rrf") or (ce is not None and ce.provider_name == "rrf")
                 apply_combined_scoring(
                     scored_results,
                     now=_recall_scoring_now(question_date),

hindsight-api-slim/hindsight_api/engine/search/fusion.py

@@ -77,6 +77,66 @@ def reciprocal_rank_fusion(result_lists: list[list[RetrievalResult]], k: int = 6
     return merged_results
 
 
+def interleave_fusion(result_lists: list[list[RetrievalResult]]) -> list[MergedCandidate]:
+    """Round-robin (interleaved) fusion — an alternative to RRF for dedup-style recall.
+
+    RRF scores a doc by the *sum* of its reciprocal ranks across arms, so a result
+    that is #1 in one arm but absent/low in the others gets averaged down. That is
+    exactly the consolidation-dedup failure mode: the near-identical existing
+    observation (the "twin" to merge into) is semantic rank #1, yet shares no
+    source-fact graph link and little lexical overlap, so RRF drops it below the
+    recall budget cutoff and the LLM never sees it → creates a duplicate.
+
+    Interleave instead *guarantees every arm's top hits a slot*: take each arm's
+    #1, then each arm's #2, … in arm-priority order, de-duplicating, until all
+    results are placed. The arm priority is the order of ``result_lists``
+    (semantic, bm25, graph, temporal), so semantic #1 is always first.
+
+    ``rrf_score`` is assigned strictly decreasing by final interleave position so
+    downstream order-by-score sorts preserve the interleave order; ``source_ranks``
+    mirrors the RRF bookkeeping (each doc's rank within every arm it appears in).
+    """
+    source_names = ["semantic", "bm25", "graph", "temporal"]
+    source_ranks: dict[str, dict[str, int]] = {}
+    all_retrievals: dict[str, RetrievalResult] = {}
+
+    for source_idx, results in enumerate(result_lists):
+        source_name = source_names[source_idx] if source_idx < len(source_names) else f"source_{source_idx}"
+        for rank, retrieval in enumerate(results, start=1):
+            if not isinstance(retrieval, RetrievalResult):
+                raise TypeError(
+                    f"Expected RetrievalResult but got {type(retrieval).__name__} in {source_name} results at rank {rank}"
+                )
+            doc_id = retrieval.id
+            all_retrievals.setdefault(doc_id, retrieval)
+            source_ranks.setdefault(doc_id, {})[f"{source_name}_rank"] = rank
+
+    # Round-robin pick across arms in priority order: all #1s, then all #2s, ...
+    ordered_ids: list[str] = []
+    seen: set[str] = set()
+    max_len = max((len(r) for r in result_lists), default=0)
+    for r in range(max_len):
+        for results in result_lists:
+            if r < len(results):
+                doc_id = results[r].id
+                if doc_id not in seen:
+                    seen.add(doc_id)
+                    ordered_ids.append(doc_id)
+
+    n = len(ordered_ids)
+    return [
+        MergedCandidate(
+            retrieval=all_retrievals[doc_id],
+            # Strictly decreasing by interleave position → sorting desc by rrf_score
+            # reproduces the interleave order downstream.
+            rrf_score=float(n - pos),
+            rrf_rank=pos + 1,
+            source_ranks=source_ranks[doc_id],
+        )
+        for pos, doc_id in enumerate(ordered_ids)
+    ]
+
+
 def normalize_scores_on_deltas(results: list[dict[str, Any]], score_keys: list[str]) -> list[dict[str, Any]]:
     """
     Normalize scores based on deltas (min-max normalization within result set).

hindsight-api-slim/tests/test_interleave_fusion.py

@@ -0,0 +1,61 @@
+"""Unit tests for round-robin interleave fusion (consolidation dedup recall)."""
+
+from hindsight_api.engine.search.fusion import interleave_fusion
+from hindsight_api.engine.search.types import RetrievalResult
+
+
+def _r(doc_id: str) -> RetrievalResult:
+    return RetrievalResult(id=doc_id, text=f"text-{doc_id}", fact_type="observation")
+
+
+def _ids(merged) -> list[str]:
+    return [mc.id for mc in merged]
+
+
+def test_round_robin_order_takes_each_arm_in_turn():
+    semantic = [_r("s1"), _r("s2"), _r("s3")]
+    bm25 = [_r("b1"), _r("b2")]
+    graph = [_r("g1")]
+
+    merged = interleave_fusion([semantic, bm25, graph])
+
+    # Round 0: s1, b1, g1 ; round 1: s2, b2 ; round 2: s3
+    assert _ids(merged) == ["s1", "b1", "g1", "s2", "b2", "s3"]
+
+
+def test_semantic_top_hit_is_always_first():
+    # The dedup twin: semantic #1 but absent from every other arm. Must still lead.
+    semantic = [_r("twin"), _r("s2")]
+    bm25 = [_r("b1"), _r("b2"), _r("b3")]
+    graph = [_r("g1")]
+
+    merged = interleave_fusion([semantic, bm25, graph])
+
+    assert merged[0].id == "twin"
+
+
+def test_dedup_keeps_first_occurrence_and_records_all_arm_ranks():
+    # "x" is semantic #1 and bm25 #2; it should appear once, at its first (semantic) slot,
+    # but carry ranks from every arm it appears in.
+    semantic = [_r("x"), _r("s2")]
+    bm25 = [_r("b1"), _r("x")]
+
+    merged = interleave_fusion([semantic, bm25])
+
+    assert _ids(merged) == ["x", "b1", "s2"]
+    x = next(mc for mc in merged if mc.id == "x")
+    assert x.source_ranks == {"semantic_rank": 1, "bm25_rank": 2}
+
+
+def test_rrf_score_strictly_decreasing_preserves_order_on_sort():
+    merged = interleave_fusion([[_r("a"), _r("b")], [_r("c")]])
+    scores = [mc.rrf_score for mc in merged]
+    assert scores == sorted(scores, reverse=True)
+    assert len(set(scores)) == len(scores)  # strictly decreasing, no ties
+    # rrf_rank reflects the interleave position
+    assert [mc.rrf_rank for mc in merged] == [1, 2, 3]
+
+
+def test_empty_inputs():
+    assert interleave_fusion([]) == []
+    assert interleave_fusion([[], []]) == []