Fix result-set truncation, prefer canonical copies, add name channel

Found via merve's hf-find traces (merve/hf-find-traces).

- Relevance path fetched n=k, so _knn's LIMIT 2k left near-duplicate collapse
no room to backfill and searches silently returned fewer than k (9 of 10 for
"RF-DETR segmentation", 5 of 10 with a param filter). Pool is now k*10 on the
relevance path; the explicit sort modes keep k*4 so widening the pool cannot
trade their relevance for popularity.
- Dup-collapse kept whichever copy scored a hair higher on cosine, which is
routinely a 0-download mirror. It now keeps the most-downloaded member of a
near-duplicate group, in the group's existing rank position.
- Summary embeddings cannot retrieve a repo by name: "iSAID" returned Icelandic
census records and chip-design netlists, while ariG23498/iSAID (whose summary
never says "iSAID") was unreachable. Added an id-matching channel, RRF-fused
at half weight so it rescues a miss without outvoting the vector ranking.
Tokens must look like identifiers and match at an id boundary -- a bare
substring put RASSAISAID/finance-deepseek-prompts above the real iSAID repos.
- Refresh likes/downloads at load from the pipeline's metadata snapshot. Corpus
rows freeze those counts at the moment a card was last summarised, so
dronefreak/UAVid-2020 read 0 downloads against 2454 live. Missing snapshot is
non-fatal.
- Added POPULARITY_WEIGHT, defaulting to OFF. It scores well on a canonical-repo
eval but a blind 3-judge A/B over 24 unseen queries split by domain: mainstream
14-2 for the prior, niche/historical/GLAM 3-5 against it. Rationale in-file.

Verified by test_ranking.py (14 fixture assertions, no corpus or model needed).

app.py

@@ -10,6 +10,7 @@ backend so query vectors stay aligned with the seed document embeddings.
 """
 import logging
 import os
+import re
 import time
 from contextlib import asynccontextmanager
 from typing import List, Optional
@@ -133,11 +134,39 @@ def load_table(cfg: str, table: str, has_param: bool):
           )
         )
     """)
+    _refresh_metadata(cfg, table)
     n = con.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
     STATE["counts"][table] = n
     logger.info(f"loaded {table} (from {cfg}): {n:,} rows @ {EMB_DIM}d")
 
 
+def _refresh_metadata(cfg: str, table: str) -> None:
+    """Overwrite likes/downloads from the pipeline's metadata snapshot, if present.
+
+    Corpus rows carry the counts a repo had when its card was last summarised, and
+    cards rarely change — so a repo can sit at 0 downloads long after it is widely
+    used. Ranking now leans on those counts, so refresh them from the small
+    (id, downloads, likes) file the delta pipeline publishes each run. Absent file
+    or unreadable snapshot is not fatal: the corpus values still work, just stale.
+    """
+    try:
+        path = hf_hub_download(EMBEDDINGS_REPO, f"_state/metadata/{cfg}.parquet", repo_type="dataset")
+    except Exception as e:
+        logger.warning(f"{table}: no metadata snapshot ({e}); using corpus counts")
+        return
+    try:
+        n = con.execute(f"""
+            UPDATE {table} AS t
+            SET downloads = COALESCE(m.downloads, 0)::BIGINT,
+                likes     = COALESCE(m.likes, 0)::BIGINT
+            FROM read_parquet('{path}') AS m
+            WHERE m.id = t.id
+        """).fetchone()
+        logger.info(f"{table}: refreshed metadata for {(n[0] if n else 0):,} rows")
+    except Exception as e:
+        logger.warning(f"{table}: metadata refresh failed ({e}); using corpus counts")
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     t0 = time.time()
@@ -273,23 +302,41 @@ def _vec_literal(vec):
 DUP_SIM = 0.985  # near-identical cards (mirrors/re-uploads) collapse to the top-ranked copy
 
 
+DUP_OVERFETCH = 6  # candidates per requested row, so dup-collapse can be backfilled
+_DL_IDX = 3        # position of `downloads` within COLS (id, summary, likes, downloads, ...)
+
+
 def _knn(table, qvec, k, where_sql, cols, params=None, return_emb=False):
     """Top-k by cosine, with near-duplicate collapse.
 
-    Overfetches 2x, then greedily drops rows whose embedding is ~identical
-    (cos >= DUP_SIM) to a higher-ranked kept row — mirror repos otherwise eat
-    adjacent result slots. Returns (cols..., sim) tuples; with return_emb=True,
-    (cols..., emb, sim).
+    Overfetches DUP_OVERFETCH x, then greedily drops rows whose embedding is
+    ~identical (cos >= DUP_SIM) to a higher-ranked kept row — mirror repos
+    otherwise eat adjacent result slots. When a duplicate is more downloaded than
+    the copy already kept, it replaces it in place: the group keeps its ranking
+    position but is represented by the canonical repo rather than whichever
+    mirror happened to score a hair higher.
+
+    The overfetch has to exceed the dup rate or collapse silently returns fewer
+    than k (popular model families are dominated by near-identical conversion
+    cards); the LIMIT is cheap because the full cosine scan dominates either way.
+    Returns (cols..., sim) tuples; with return_emb=True, (cols..., emb, sim).
     """
     import numpy as np
     q = f"""SELECT {cols}, emb, array_cosine_similarity(emb, {_vec_literal(qvec)}) AS sim
-            FROM {table} {where_sql} ORDER BY sim DESC LIMIT {int(k) * 2}"""
+            FROM {table} {where_sql} ORDER BY sim DESC LIMIT {int(k) * DUP_OVERFETCH}"""
     rows = con.execute(q, params or []).fetchall()
     kept, kept_vecs = [], []
     for r in rows:
         v = np.asarray(r[-2], dtype=np.float32)
-        if kept_vecs and float(np.max(np.stack(kept_vecs) @ v)) >= DUP_SIM:
-            continue
+        if kept_vecs:
+            sims = np.stack(kept_vecs) @ v
+            j = int(np.argmax(sims))
+            if float(sims[j]) >= DUP_SIM:
+                # Same card, different repo: keep whichever copy people actually use.
+                if (r[_DL_IDX] or 0) > (kept[j][_DL_IDX] or 0):
+                    kept[j] = r if return_emb else r[:-2] + (r[-1],)
+                    kept_vecs[j] = v
+                continue
         kept.append(r if return_emb else r[:-2] + (r[-1],))
         kept_vecs.append(v)
         if len(kept) >= int(k):
@@ -319,21 +366,88 @@ def _bm25_ids(kind: str, text: str, n: int) -> list:
     return [r[0] for r in rows]
 
 
-def _hybrid_fuse(table, kind, query, qvec, raw, k, where_sql, wparams):
-    """RRF-fuse vector KNN rows with a BM25 channel over full card text.
+# A query that is short and has no spaces is a repo name, not a description
+# ("iSAID", "rtdetr"). Summary embeddings cannot retrieve those: the summary of
+# ariG23498/iSAID never says "iSAID", and "iSAID" alone retrieves Icelandic
+# census records via ID-ish subword tokens. Matching the id directly is the only
+# channel that can find them.
+_NAME_MAX_TOKENS = 3   # "iSAID instance segmentation" still carries a name token
+_NAME_MIN_CHARS = 4    # shorter tokens ("ocr", "sam") match far too much
+NAME_CHANNEL_WEIGHT = 0.5  # below the vector channel: rescue a miss, never dominate
+
+# Hyphenated descriptors that pass the identifier test but name no repo.
+_NAME_STOPWORDS = frozenset({"real-time", "state-of-the-art", "open-source", "fine-tuned"})
+
+
+def _is_identifier_like(tok: str) -> bool:
+    """True for tokens shaped like a repo/model name rather than an English word.
+
+    Names carry a digit (yolov10), an internal capital (iSAID, RF-DETR), or a
+    separator (rf-detr). Ordinary description words ("herbarium", "element") carry
+    none of these, and matching those against ids returns download-ranked noise.
+    """
+    if tok.lower() in _NAME_STOPWORDS:
+        return False
+    return bool(
+        any(c.isdigit() for c in tok)
+        or any(c in "-_." for c in tok)
+        or re.search(r"[a-z][A-Z]|^[A-Z]{2,}$", tok)
+    )
+
+
+def _name_tokens(query: str) -> list:
+    """Tokens from the query that plausibly name a repo rather than describe one."""
+    if len(query.strip()) > 60 or len(query.split()) > _NAME_MAX_TOKENS:
+        return []
+    return [
+        tok
+        for raw in re.split(r"[\s,]+", query.strip())
+        if len(tok := raw.strip("\"'()[]")) >= _NAME_MIN_CHARS and _is_identifier_like(tok)
+    ]
+
 
-    BM25-only hits are fetched from the in-memory table (with their true cosine
+def _name_ids(table, tokens, n, where_sql, wparams):
+    """Top-n ids whose name contains any of these tokens, most-downloaded first.
+
+    The token must start at an id boundary (start, or after / _ . -). A bare
+    substring match puts `RASSAISAID/finance-deepseek-prompts` above the real
+    `ariG23498/iSAID` for the query "iSAID". The trailing side is deliberately left
+    open so name variants still match — "yolov10" has to find `jameslahm/yolov10x`.
+
+    Ordering by downloads (not cosine) is deliberate: this channel exists to find
+    the repo actually named, and RRF only uses the rank, so the canonical copy of
+    a name should lead its own channel.
+    """
+    if not tokens:
+        return []
+    cond = f"AND {where_sql[6:]}" if where_sql else ""
+    ors = " OR ".join("regexp_matches(id, ?, 'i')" for _ in tokens)
+    pats = [f"(^|[/_.-]){re.escape(t)}" for t in tokens]
+    rows = con.execute(
+        f"""SELECT id FROM {table}
+            WHERE ({ors}) {cond}
+            ORDER BY downloads DESC LIMIT {int(n)}""",
+        pats + list(wparams or []),
+    ).fetchall()
+    return [r[0] for r in rows]
+
+
+def _fuse(table, qvec, raw, extra_ids, k, where_sql, wparams, weight=1.0):
+    """RRF-fuse the vector KNN rows with a second channel's id list.
+
+    Channel-only hits are fetched from the in-memory table (with their true cosine
     similarity) and must pass the same metadata filters as the vector channel.
+    `weight` scales the second channel's contribution, so a speculative channel can
+    rescue a miss without outvoting the vector ranking.
     Returns rows in fused order, same tuple shape as _knn output.
     """
-    bm_ids = _bm25_ids(kind, query, k * 2)
-    if not bm_ids:
+    if not extra_ids:
         return raw
     rrf: dict = {}
     for rank, rid in enumerate(r[0] for r in raw):
         rrf[rid] = rrf.get(rid, 0.0) + 1.0 / (60 + rank)
-    for rank, rid in enumerate(bm_ids):
-        rrf[rid] = rrf.get(rid, 0.0) + 1.0 / (60 + rank)
+    for rank, rid in enumerate(extra_ids):
+        rrf[rid] = rrf.get(rid, 0.0) + weight / (60 + rank)
     order = sorted(rrf, key=rrf.get, reverse=True)
     known = {r[0]: r for r in raw}
     missing = [rid for rid in order if rid not in known][: k * 2]
@@ -349,6 +463,11 @@ def _hybrid_fuse(table, kind, query, qvec, raw, k, where_sql, wparams):
     return [known[rid] for rid in order if rid in known]
 
 
+def _hybrid_fuse(table, kind, query, qvec, raw, k, where_sql, wparams):
+    """RRF-fuse vector KNN rows with a BM25 channel over full card text."""
+    return _fuse(table, qvec, raw, _bm25_ids(kind, query, k * 2), k, where_sql, wparams)
+
+
 async def _sort_results(rows, id_field, sort_by, k):
     """rows: list of dicts with keys id, similarity, summary, likes, downloads, [param_count]."""
     if sort_by == "trending":
@@ -367,10 +486,64 @@ async def _sort_results(rows, id_field, sort_by, k):
         rows.sort(key=lambda r: r.get("last_modified") or "", reverse=True)
         rows = rows[:k]
     else:
-        rows = rows[:k]
+        rows = _apply_popularity_prior(rows)[:k]
     return rows
 
 
+# Optional log-scaled download term, added to cosine to break ties toward the copy
+# people actually use (a 0-download mirror otherwise outranks the canonical repo,
+# since their cards read the same).
+#
+# OFF by default, deliberately. On hf-find's eval/queries.jsonl it looks like a
+# clear win (mean recall 0.363 -> 0.435, zero-download slots 28.3% -> 6.2% at 0.08),
+# but that eval's expected repos were chosen *because* they are canonical, so it
+# cannot help but reward a popularity prior. A blind 3-judge A/B over 24 unseen
+# queries split by domain instead:
+#
+#     mainstream queries          prior 14 - 2 baseline
+#     niche / historical / GLAM   prior  3 - 5 baseline
+#
+# The mechanism: a 150k-download repo gets the full weight while a 50-download
+# exact match gets ~a third of it, so any similarity gap under ~0.054 flips — and
+# niche queries live in exactly that regime. Concretely, at 0.08 the query "OCR for
+# historical printed text" ranks generic dots.ocr-1.5 above
+# wjbmattingly/LightOnOCR-2-1B-cultural-heritage-english.
+#
+# So this is a domain trade, not a free improvement. Enable per-deployment with
+# POPULARITY_WEIGHT=0.08 if the audience is mainstream model/dataset discovery.
+POPULARITY_WEIGHT = float(os.getenv("POPULARITY_WEIGHT", "0"))
+
+
+def _rerank_pool(k: int, sort_by: str) -> int:
+    """Candidate rows to pull before re-ranking down to k.
+
+    The relevance path re-ranks (popularity prior, optional name fusion) and gains
+    from a deeper pool: on eval/queries.jsonl mean recall is 0.363 at k*1, 0.399 at
+    k*4 and 0.435 at k*10, flat thereafter. The cosine scan touches every row
+    regardless, so a larger LIMIT is close to free.
+
+    The explicit sort modes keep the old k*4: they sort hard by likes/downloads, so
+    widening the pool would trade relevance for popularity rather than break ties.
+    """
+    return k * (10 if sort_by == "similarity" else 4)
+
+
+def _apply_popularity_prior(rows):
+    """Re-rank by similarity + a log-scaled, per-result-set-normalised download term."""
+    if POPULARITY_WEIGHT <= 0 or not rows:
+        return rows
+    import math
+    mx = max((r.get("downloads") or 0) for r in rows)
+    if mx <= 0:
+        return rows
+    denom = math.log1p(mx)
+    return sorted(
+        rows,
+        key=lambda r: r["similarity"] + POPULARITY_WEIGHT * (math.log1p(r.get("downloads") or 0) / denom),
+        reverse=True,
+    )
+
+
 def _rows_to_dataset(raw):
     out = []
     for id_, summary, likes, downloads, param, task, license_, language, last_modified, sim in raw:
@@ -447,13 +620,17 @@ async def search_datasets(
 ):
     try:
         qvec = embed_query(f"Instruct: {DS_TASK}\nQuery:{query}")
-        n = k * 4 if sort_by != "similarity" else (k * 2 if hybrid else k)
+        n = _rerank_pool(k, sort_by)
         where_sql, wparams = _where(min_likes, min_downloads,
                                     task=task, license=license, language=language,
                                     modified_after=modified_after)
         raw = _knn("dataset_cards", qvec, n, where_sql, COLS, wparams)
         if hybrid:
             raw = _hybrid_fuse("dataset_cards", "datasets", query, qvec, raw, k, where_sql, wparams)
+        elif (name_toks := _name_tokens(query)):
+            raw = _fuse("dataset_cards", qvec, raw,
+                        _name_ids("dataset_cards", name_toks, k, where_sql, wparams),
+                        k, where_sql, wparams, weight=NAME_CHANNEL_WEIGHT)
         rows = await _sort_results(_rows_to_dataset(raw), "dataset", sort_by, k)
         return QueryResponse(results=[QueryResult(**{k2: v for k2, v in r.items() if k2 != "_id"}) for r in rows])
     except Exception as e:
@@ -512,13 +689,17 @@ async def search_models(
     try:
         use_param = min_param_count > 0 or max_param_count is not None
         qvec = embed_query(f"search_query: {query}")
-        n = k * 4 if sort_by != "similarity" else (k * 2 if hybrid else k)
+        n = _rerank_pool(k, sort_by)
         where_sql, wparams = _where(min_likes, min_downloads, min_param_count, max_param_count, use_param,
                                     task=task, license=license, language=language,
                                     modified_after=modified_after)
         raw = _knn("model_cards", qvec, n, where_sql, COLS, wparams)
         if hybrid:
             raw = _hybrid_fuse("model_cards", "models", query, qvec, raw, k, where_sql, wparams)
+        elif (name_toks := _name_tokens(query)):
+            raw = _fuse("model_cards", qvec, raw,
+                        _name_ids("model_cards", name_toks, k, where_sql, wparams),
+                        k, where_sql, wparams, weight=NAME_CHANNEL_WEIGHT)
         rows = await _sort_results(_rows_to_model(raw), "model", sort_by, k)
         return ModelQueryResponse(results=[ModelQueryResult(**{k2: v for k2, v in r.items() if k2 != "_id"}) for r in rows])
     except Exception as e: