fix(retrieval): suppress weak lexical fusion hits

packages/infinity_context_adapters/infinity_context_adapters/postgres/locator_retrieval.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+import re
+import unicodedata
 from dataclasses import dataclass
 from uuid import uuid4
 
+from infinity_context_core.application.normalize import normalize_text
 from infinity_context_core.features.context_building.public import (
     CanonicalHydrationInvariantError,
     CanonicalLocatorCandidate,
@@ -13,7 +16,7 @@
     LocatorProviderResult,
     LocatorRetrievalRequest,
 )
-from sqlalchemy import case, cast, not_, or_, select, text
+from sqlalchemy import case, cast, false, literal_column, not_, or_, select, text
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
@@ -22,6 +25,22 @@
 )
 from infinity_context_adapters.postgres.models import MemoryChunkRow
 
+_WORD_RE = re.compile(r"\w+%?", re.UNICODE)
+
+# These words carry little retrieval intent on their own.  Keep this deliberately
+# small and language-generic: the lexical lane should remove query scaffolding,
+# not attempt stemming or language-specific semantic analysis.
+_LOW_INFORMATION_WORDS = frozenset(
+    # English function words and common request scaffolding.
+    """a about an and are as at be by can did do does for from has have how in
+    d i info information is it know ll m me of on or please re s t tell that the
+    this to ve was what when where which who why with would you""".split()  # noqa: SIM905
+    # Russian equivalents.
+    + """а без был была в во вы где для до есть и из или известно информация как
+    к когда кто ли мне мы на но о об от по почему при про расскажи с сведения со то
+    у что это я""".split()  # noqa: SIM905
+)
+
 
 @dataclass(frozen=True, slots=True)
 class PostgresLocatorCandidateProvider:
@@ -101,29 +120,57 @@ async def hydrate_final_locator_read(
 
 
 def _candidate_statement(request: LocatorRetrievalRequest, query: str):
-    terms = tuple(dict.fromkeys(term.casefold() for term in query.split() if term))
+    terms = _keyword_terms(query)
     matches = tuple(
-        MemoryChunkRow.normalized_text.contains(term, autoescape=True) for term in terms
+        or_(*(MemoryChunkRow.normalized_text.contains(alias, autoescape=True) for alias in aliases))
+        for aliases in terms
     )
-    relevance = sum((case((match, 1), else_=0) for match in matches), start=0)
+    relevance = sum((case((match, 1), else_=0) for match in matches), start=literal_column("0"))
     conditions = list(_hard_sql_conditions(request))
     if matches:
+        # Keep the disjunction explicit as an indexable prefilter.  PostgreSQL can
+        # use each LIKE arm with the normalized_text trigram index before applying
+        # the stricter summed minimum-match qualification.
         conditions.append(or_(*matches))
+        # For a one-keyword query, that keyword is necessarily the whole lexical
+        # intent.  For longer queries, require two distinct informative words so
+        # one incidental overlap cannot lend an RRF contribution to a dense hit.
+        conditions.append(relevance >= min(2, len(matches)))
+    else:
+        conditions.append(false())
+    labeled_relevance = relevance.label("relevance")
     return (
         select(
             MemoryChunkRow.id.label("id"),
             MemoryChunkRow.retrieval_version.label("retrieval_version"),
-            relevance.label("relevance"),
+            labeled_relevance,
         )
         .where(*conditions)
         .order_by(
-            relevance.desc(),
+            labeled_relevance.desc(),
             MemoryChunkRow.retrieval_sequence_ordinal,
             MemoryChunkRow.id,
         )
     )
 
 
+def _keyword_terms(query: str) -> tuple[tuple[str, ...], ...]:
+    """Return stable logical signals with canonical and Unicode-folded aliases."""
+
+    words: list[tuple[str, ...]] = []
+    seen: set[str] = set()
+    for canonical_word in _WORD_RE.findall(normalize_text(query)):
+        folded = unicodedata.normalize("NFKC", canonical_word).casefold()
+        if not folded or folded in _LOW_INFORMATION_WORDS or folded in seen:
+            continue
+        seen.add(folded)
+        # Existing canonical rows use normalize_text (lower), while casefold/NFKC
+        # is needed for deterministic Unicode query equivalence.  Query both forms
+        # as one logical word so aliases never inflate relevance.
+        words.append(tuple(dict.fromkeys((canonical_word, folded))))
+    return tuple(words)
+
+
 def _hard_sql_conditions(request: LocatorRetrievalRequest) -> tuple[object, ...]:
     scope = request.scope
     filters = request.hard_filters

packages/infinity_context_core/infinity_context_core/features/context_building/tests/test_locator_retrieval_scoring.py

@@ -1,5 +1,7 @@
 """Exact integer scoring boundary tests for Retrieval."""
 
+import asyncio
+
 import pytest
 
 from infinity_context_core.features.context_building.application.locator_retrieval import (
@@ -15,7 +17,12 @@
 )
 from infinity_context_core.features.context_building.tests.test_locator_retrieval import (
     FINGERPRINT,
+    _canonical,
+    _hit,
+    _Hydrator,
     _Provider,
+    _request,
+    _retrieve,
 )
 
 
@@ -45,6 +52,42 @@ def test_integer_rrf_exact_halves_use_round_half_even() -> None:
     assert _rrf_contribution_score_picos(100_005, 100_000, 100_000, 324) == 260_429_688
 
 
+def test_empty_lexical_lane_preserves_dense_only_results_and_strong_hits_fuse() -> None:
+    dense = _Provider((_hit("dense-only", rank=1), _hit("shared", rank=2)))
+    canonical = _Hydrator((_canonical("dense-only"), _canonical("shared")))
+    registrations = (
+        LocatorProviderRegistration("dense", dense),
+        LocatorProviderRegistration("lexical", _Provider(())),
+    )
+
+    dense_only = asyncio.run(_retrieve(registrations, canonical).execute(_request()))
+    assert tuple(item.canonical_identity for item in dense_only.candidates) == (
+        "dense-only",
+        "shared",
+    )
+    assert all(
+        tuple(value.provider_id for value in item.contributions) == ("dense",)
+        for item in dense_only.candidates
+    )
+
+    fused = asyncio.run(
+        _retrieve(
+            (
+                registrations[0],
+                LocatorProviderRegistration(
+                    "lexical", _Provider((_hit("shared", provider="lexical"),))
+                ),
+            ),
+            canonical,
+        ).execute(_request())
+    )
+    assert fused.candidates[0].canonical_identity == "shared"
+    assert tuple(value.provider_id for value in fused.candidates[0].contributions) == (
+        "dense",
+        "lexical",
+    )
+
+
 def test_preference_evidence_rejects_cross_dimension_weight_swap() -> None:
     with pytest.raises(ValueError, match="dimension evidence"):
         LocatorPreferenceEvidence(

tests/adapters/test_locator_parent_lifecycle.py

@@ -23,7 +23,7 @@ def test_parent_lifecycle_and_binding_are_canonical_for_every_locator_read() ->
     asyncio.run(_assert_parent_authority())
 
 
-def test_locator_keyword_matching_preserves_case_unicode_and_literal_wildcards() -> None:
+def test_locator_keyword_matching_normalizes_words_and_rejects_weak_hits() -> None:
     asyncio.run(_assert_keyword_matching_semantics())
 
 
@@ -74,6 +74,12 @@ async def _assert_keyword_matching_semantics() -> None:
     sessions = async_sessionmaker(engine, expire_on_commit=False)
     cases = (
         ("mixed-unicode", "Café Δέλτα", "café δέλτα"),
+        ("capital-sharp-s", "A STRAẞE landmark", "a straße landmark"),
+        ("generic-landmark", "A generic landmark", "a generic landmark"),
+        ("english-strong", "PostgreSQL roadmap accepted", "postgresql roadmap accepted"),
+        ("english-weak", "PostgreSQL unrelated note", "postgresql unrelated note"),
+        ("russian-strong", "Релиз roadmap утвержден", "релиз roadmap утвержден"),
+        ("russian-weak", "Релиз перенесен", "релиз перенесен"),
         ("literal-percent", "Budget 100% complete", "budget 100% complete"),
         ("literal-underscore", "release_candidate ready", "release_candidate ready"),
         ("percent-decoy", "Budget 1000 complete", "budget 1000 complete"),
@@ -93,9 +99,15 @@ async def _assert_keyword_matching_semantics() -> None:
 
     provider = PostgresLocatorCandidateProvider(sessions)
     expected = {
-        "CAFÉ ΔΈΛΤΑ": ["chunk-mixed-unicode"],
+        "CAFÉ, ΔΈΛΤΑ!": ["chunk-mixed-unicode"],
+        "STRAẞE": ["chunk-capital-sharp-s"],
         "100%": ["chunk-literal-percent"],
         "release_candidate": ["chunk-literal-underscore"],
+        "Please tell me information about the PostgreSQL roadmap?": ["chunk-english-strong"],
+        "Что известно о релиз, и roadmap?": ["chunk-russian-strong"],
+        "Please, what is this about?": [],
+        "I tell you what's this?": [],
+        "Где и что это?": [],
     }
     for query, identities in expected.items():
         result = await provider.retrieve_locator_candidates(_request(query))

tests/adapters/test_locator_retrieval_adapters.py

@@ -15,6 +15,7 @@
     _candidate_statement,
     _canonical_rows,
     _hard_sql_conditions,
+    _keyword_terms,
 )
 from infinity_context_adapters.postgres.mappers import chunk_row_to_domain
 from infinity_context_adapters.postgres.retrieval_projection_mapping import (
@@ -110,22 +111,54 @@ def test_postgres_array_filters_compile_to_jsonb_containment() -> None:
 
 def test_postgres_keyword_match_targets_the_indexed_normalized_column() -> None:
     compiled = _candidate_statement(
-        _request(), "CAFÉ 100% release_candidate"
-    ).compile(
-        dialect=postgresql.dialect()
-    )
+        _request(), "CAFÉ, roadmap! 100% release_candidate"
+    ).compile(dialect=postgresql.dialect())
     statement = str(compiled)
 
     assert "lower(" not in statement
-    assert statement.count("memory_chunks.normalized_text LIKE") == 9
-    assert [compiled.params[f"normalized_text_{ordinal}"] for ordinal in range(1, 4)] == [
+    # Each predicate appears in the indexable OR prefilter and the summed CASE
+    # qualification.  Bound values remain escaped rather than interpolated.
+    assert statement.count("memory_chunks.normalized_text LIKE") == 12
+    assert [compiled.params[f"normalized_text_{ordinal}"] for ordinal in range(1, 5)] == [
         "café",
+        "roadmap",
         "100/%",
         "release/_candidate",
     ]
     assert "ESCAPE '/'" in statement
 
 
+def test_keyword_terms_are_unicode_folded_punctuation_free_and_deterministic() -> None:
+    query = "PLEASE—what is STRAẞE, ＲＯＡＤＭＡＰ? Straße"
+    expected = (("straße", "strasse"), ("ｒｏａｄｍａｐ", "roadmap"))
+
+    assert _keyword_terms(query) == expected
+    assert all(_keyword_terms(query) == expected for _ in range(20))
+
+
+def test_keyword_terms_remove_generic_english_and_russian_scaffolding() -> None:
+    assert _keyword_terms("Please tell me information about the PostgreSQL roadmap?") == (
+        ("postgresql",),
+        ("roadmap",),
+    )
+    assert _keyword_terms("Что известно о релизе, и roadmap?") == (("релизе",), ("roadmap",))
+
+
+def test_keyword_terms_preserve_literal_percent_and_reject_contraction_scaffolding() -> None:
+    assert _keyword_terms("Was it 100% or 1000?") == (("100%",), ("1000",))
+    assert _keyword_terms("I tell you what's it") == ()
+    assert _keyword_terms("you've roadmap") == (("roadmap",),)
+
+
+def test_single_informative_keyword_still_builds_a_lexical_lane() -> None:
+    compiled = _candidate_statement(_request(), "I know the roadmap").compile(
+        dialect=postgresql.dialect()
+    )
+
+    assert "memory_chunks.normalized_text LIKE" in str(compiled)
+    assert "false" not in str(compiled).lower()
+
+
 def test_qdrant_provider_preserves_raw_score_rank_and_version() -> None:
     search = _Search()
     result = asyncio.run(

tests/adapters/test_postgres_canonical_keyword_trigram.py

@@ -578,8 +578,6 @@ async def _assert_locator_semantic_parity(session: AsyncSession) -> None:
         ("locator-unicode-all", 3),
         ("locator-literal-percent", 2),
         ("locator-literal-underscore", 2),
-        ("locator-percent-decoy", 1),
-        ("locator-underscore-decoy", 1),
     ]
     assert raw_rows == expected
     assert [