feat(discovery): balance the starter feed per interest and give it a recent lane

Measured 2026-09-30 for a reader who ticked NLP, CV, ML, AI, Robotics and
Software Engineering, the 200-paper starter pool held 147 papers from 2024,
53 from early 2025 and none from 2026, and Computer Vision got 22 of 200.

- Deal per ticked interest, not per arXiv code. Two-code interests (ML is
cs.LG + stat.ML) were getting twice the share of one-code ones.
- Rank the established lane by citations per month since publication
(floored at three months). Raw totals inside a 24-month window always
favour the oldest papers in it.
- Every third slot per interest comes from a new fresh lane: papers from
the corpus's last three months, most-cited first. A few-weeks-old paper
cannot win on citations however they are ranked.

Onboarding suggestions use the same grouping.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/db.py

@@ -633,6 +633,20 @@ async def get_user_category_filter(user_id: str) -> set[str]:
     return expand_category_groups(state["selected_categories"])
 
 
+async def get_user_category_groups(user_id: str) -> dict[str, set[str]]:
+    """The reader's selected interests, each mapped to its arXiv codes.
+
+    Same source as get_user_category_filter, without flattening: starter feeds
+    balance per interest, and flattening loses which codes belong together.
+    """
+    state = await get_onboarding_state(user_id)
+    if state is None:
+        return {}
+    from app.config import expand_category_groups
+    return {key: codes for key in state["selected_categories"]
+            if (codes := expand_category_groups([key]))}
+
+
 # ── Phase 6.5 B3: Cluster snapshot versioning ─────────────────────────────────
 
 async def save_cluster_snapshot(user_id: str, clusters: list[dict]) -> str:

app/discovery_svc.py

@@ -2,39 +2,88 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Mapping
 from itertools import zip_longest
 
 from app import local_meta, turso_svc
 
+# Every third starter slot within an interest goes to a paper from the corpus's
+# last three months. Ranked by citations alone, those can never compete: a
+# paper a few weeks old has almost none yet.
+_FRESH_EVERY = 3
 
-async def starter_papers(categories: set[str], limit: int = 30) -> list[dict]:
-    """Give thin categories room without eight remote full-table scans.
 
-    With the indexed sidecar, query each category and round-robin its ranked
-    results. Without it, retain the existing single cached remote fallback.
-    These are citation/popularity candidates, not measured live trends.
+def _interleave(lists: list[list[dict]]) -> list[dict]:
+    return [p for row in zip_longest(*lists) for p in row if p]
+
+
+def _with_fresh(established: list[dict], fresh: list[dict]) -> list[dict]:
+    out, e, f = [], iter(established), iter(fresh)
+    while True:
+        batch = [p for p in (next(e, None) for _ in range(_FRESH_EVERY - 1)) if p]
+        batch += [p for p in [next(f, None)] if p]
+        if not batch:
+            return out + list(e) + list(f)
+        out += batch
+
+
+async def starter_papers(
+    categories: set[str] | Mapping[str, set[str]], limit: int = 30,
+) -> list[dict]:
+    """Deal starter candidates out evenly across the reader's interests.
+
+    `categories` is either a mapping of interest -> arXiv codes (what onboarding
+    stores) or a flat set of codes, which is treated as one interest per code.
+    Balancing per interest rather than per code matters: "Machine Learning"
+    spans cs.LG and stat.ML while "Robotics" is only cs.RO, and dealing per code
+    gave the two-code interests twice the share. Measured 2026-09-30 for six
+    interests, Computer Vision got 22 of 200 starter papers and language-model
+    papers 86.
+
+    Within an interest, codes are round-robined and every third slot is a
+    recent paper (see _FRESH_EVERY). These are citation/popularity candidates,
+    not measured live trends. Without the sidecar, one cached remote query
+    serves everything and there is no fresh lane.
     """
-    if not categories or limit <= 0:
+    groups = ({k: set(v) for k, v in categories.items() if v}
+              if isinstance(categories, Mapping)
+              else {code: {code} for code in categories})
+    if not groups or limit <= 0:
         return []
-    if len(categories) == 1 or not local_meta.is_available():
-        return await turso_svc.fetch_trending_by_categories(categories, limit=limit)
+    all_codes = set().union(*groups.values())
+    if not local_meta.is_available():
+        return await turso_svc.fetch_trending_by_categories(all_codes, limit=limit)
     semaphore = asyncio.Semaphore(4)
 
-    async def one(code: str) -> list[dict]:
+    async def guarded(coro):
         async with semaphore:
-            return await turso_svc.fetch_trending_by_categories({code}, limit=limit)
+            try:
+                return await coro
+            except Exception as e:  # one bad lane must not empty the feed
+                print(f"[discovery] starter lane failed: {e}")
+                return []
+
+    codes = sorted(all_codes)
+    names = sorted(groups)
+    results = await asyncio.gather(
+        *(guarded(turso_svc.fetch_trending_by_categories({c}, limit=limit)) for c in codes),
+        *(guarded(turso_svc.fetch_fresh_by_categories(groups[g], limit=limit)) for g in names),
+    )
+    trending = dict(zip(codes, results[:len(codes)]))
+    fresh = dict(zip(names, results[len(codes):]))
 
-    results = await asyncio.gather(*(one(code) for code in sorted(categories)),
-                                   return_exceptions=True)
-    pools = [r for r in results if isinstance(r, list)]
+    per_group = [
+        _with_fresh(_interleave([trending[c] for c in sorted(groups[g])]), fresh[g])
+        for g in names
+    ]
     seen: set[str] = set()
     papers: list[dict] = []
-    for row in zip_longest(*pools):
-        for paper in row:
-            if not paper or not paper.get("arxiv_id") or paper["arxiv_id"] in seen:
-                continue
-            seen.add(paper["arxiv_id"])
-            papers.append(paper)
-            if len(papers) == limit:
-                return papers
+    for paper in _interleave(per_group):
+        aid = paper.get("arxiv_id")
+        if not aid or aid in seen:
+            continue
+        seen.add(aid)
+        papers.append(paper)
+        if len(papers) == limit:
+            break
     return papers

app/local_meta.py

@@ -209,26 +209,61 @@ def newest_update_date() -> str | None:
     return _max_date
 
 
+def _published_since(conn: sqlite3.Connection, codes: set[str],
+                     cutoff: tuple[int, int]) -> list[tuple[str, int, tuple[int, int]]]:
+    """(arxiv_id, citations, publication month) for papers in `codes` published
+    at or after `cutoff`.
+
+    Filters on the (code, update_date) index first. update_date is the last
+    revision, which is never earlier than publication, so it is a superset of
+    the window; the identifier then gives the true publication month. Only ids
+    and counts are read here, so a 24-month cs.LG window stays cheap.
+    """
+    ph = ",".join("?" * len(codes))
+    rows = conn.execute(
+        f"""SELECT arxiv_id, MAX(citation_count) FROM paper_categories
+            WHERE code IN ({ph}) AND update_date >= ?
+            GROUP BY arxiv_id""",
+        (*codes, f"{cutoff[0]:04d}-{cutoff[1]:02d}-01"),
+    ).fetchall()
+    out = []
+    for aid, cit in rows:
+        ym = pub_year_month(aid)
+        if ym is not None and ym >= cutoff:
+            out.append((aid, cit or 0, ym))
+    return out
+
+
+def _rows_in_order(conn: sqlite3.Connection, ids: list[str]) -> list[dict]:
+    got = {r["arxiv_id"]: r for r in fetch_rows(ids)}
+    return [got[i] for i in ids if i in got]
+
+
 def fetch_trending(
     codes: set[str],
     limit: int = 10,
     recency_months: int = 24,
 ) -> list[dict]:
     """
-    Well-cited *recent* papers in any of `codes`.
+    Papers in any of `codes` that are gathering citations fastest.
 
     Ordering by all-time citations returns the same canonical papers to every
     user forever — for cs.LG that is Adam (2014), scikit-learn (2011) and
     BatchNorm (2015).  Those are famous, not trending, and this feeds Tier 0,
-    which is the very first thing a new user sees.  Restricting to a recent
-    window over the same data returns Llama 3, DPO and DeepSeek-R1 instead.
+    which is the very first thing a new user sees.
+
+    Within the window, papers are ranked by citations per month since
+    publication, not by raw totals. Raw totals inside a 24-month window still
+    hand the top slots to the oldest papers in it, which have had two years to
+    accumulate; measured 2026-09-30, the starter pool for six ML interests held
+    147 papers from 2024 and none from 2026. The rate is floored at three
+    months of age so a week-old paper with two citations does not outrank
+    everything.
 
     Two details that matter:
 
       * The window is measured back from the newest paper in the corpus, not
-        from today.  The corpus is a static snapshot ending 2025-05-30, so an
-        absolute cutoff would silently empty out; anchoring to the data means
-        this keeps working if ingestion is added later.
+        from today, so a corpus whose ingestion lags keeps working.
       * Categories are wildly uneven (~302k papers in cs.LG vs ~7.9k in
         q-bio.NC), so a fixed window starves the thin ones.  The window widens
         and finally drops away entirely rather than returning a short list.
@@ -239,6 +274,25 @@ def fetch_trending(
     if conn is None:
         return []
 
+    anchor = newest_update_date()
+    now = _months_before(anchor, 0) if anchor else None
+    if now is not None:
+        try:
+            for months in (recency_months, recency_months * 2, recency_months * 4):
+                cutoff = _months_before(anchor, months)
+                if cutoff is None:
+                    break
+                window = _published_since(conn, codes, cutoff)
+                if len(window) < limit:
+                    continue
+                age = lambda ym: (now[0] - ym[0]) * 12 + (now[1] - ym[1]) + 1
+                window.sort(key=lambda r: (-r[1] / max(age(r[2]), 3), -r[1], r[0]))
+                picked = _rows_in_order(conn, [r[0] for r in window[:limit]])
+                if len(picked) >= limit:
+                    return picked
+        except Exception as e:
+            print(f"[local_meta] trending window failed ({e}) — using citation order")
+
     ph = ",".join("?" * len(codes))
     # Read the most-cited papers in these categories in one index-ordered pass,
     # then apply the publication-date window in Python.  The date cannot be
@@ -292,3 +346,28 @@ def fetch_trending(
 
     # No window could fill the slots — fall back to plain citation order.
     return (best or candidates)[:limit]
+
+
+def fetch_fresh(codes: set[str], limit: int = 10, recency_months: int = 3) -> list[dict]:
+    """
+    The newest papers in `codes`: published in the last `recency_months` of the
+    corpus, most-cited first, newest first among ties.
+
+    fetch_trending cannot supply these however it ranks: a paper a few weeks
+    old has almost no citations yet. This lane exists so a new reader's first
+    feed contains the current month of their field at all. Returns [] when the
+    sidecar is unavailable; there is no Turso fallback for this lane.
+    """
+    conn = connection() if codes else None
+    anchor = newest_update_date() if conn is not None else None
+    cutoff = _months_before(anchor, recency_months - 1) if anchor else None
+    if cutoff is None:
+        return []
+    try:
+        window = _published_since(conn, codes, cutoff)
+        window.sort(key=lambda r: r[0], reverse=True)   # newest id first...
+        window.sort(key=lambda r: -r[1])                # ...then by citations (stable)
+        return _rows_in_order(conn, [r[0] for r in window[:limit]])
+    except Exception as e:
+        print(f"[local_meta] fresh lane failed ({e})")
+        return []

app/routers/onboarding.py

@@ -163,7 +163,7 @@ async def seed_search(
             except Exception as e:
                 print(f"[onboarding] keyword fallback failed: {e}")
     else:
-        categories = await db.get_user_category_filter(user_id)
+        categories = await db.get_user_category_groups(user_id)
         if categories:
             try:
                 papers = await discovery_svc.starter_papers(categories, limit=12)

app/routers/recommendations.py

@@ -418,7 +418,9 @@ async def _build_feed(
 
     # ── Tier 0: category trending (cold start, Phase 5) ──────────────────
     if not state.has_enough_for_recs():
-        category_filter = await db.get_user_category_filter(user_id)
+        # Grouped by interest, so the starter pool balances interests rather
+        # than arXiv codes (see discovery_svc.starter_papers).
+        category_filter = await db.get_user_category_groups(user_id)
         # No categories means the reader skipped onboarding. That used to fall
         # straight through to the empty state and stay there; a reader who told
         # us nothing still gets a feed, just a broader one. See

app/turso_svc.py

@@ -551,3 +551,27 @@ async def fetch_trending_by_categories(
         for p in papers:
             _cache_put(p["arxiv_id"], p)
     return papers
+
+
+async def fetch_fresh_by_categories(categories: set[str], limit: int = 10) -> list[dict]:
+    """
+    Papers published in the corpus's last three months in `categories`, from
+    the sidecar only (see local_meta.fetch_fresh). Turso has no index that
+    makes this affordable, so without the sidecar the lane is simply empty.
+    """
+    if not categories:
+        return []
+    cache_key = ("fresh", tuple(sorted(categories)), limit)
+    cached = _TRENDING_CACHE.get(cache_key)
+    if cached is not None and (time.time() - cached[0]) < _TRENDING_TTL_SECONDS:
+        return cached[1]
+    from app import local_meta
+    if not local_meta.is_available():
+        return []
+    rows = await asyncio.to_thread(local_meta.fetch_fresh, set(categories), limit)
+    papers = [p for p in (_to_paper_dict(r) for r in rows) if p]
+    if papers:
+        _TRENDING_CACHE[cache_key] = (time.time(), papers)
+        for p in papers:
+            _cache_put(p["arxiv_id"], p)
+    return papers

tests/test_discovery_journey.py

@@ -56,7 +56,7 @@ async def test_starter_suggestions_use_selected_categories_and_mark_saved(client
                                        'category':'cs.CL','year':2026}])
     monkeypatch.setattr(discovery_svc, 'starter_papers', starter)
     r = await client.get('/api/onboarding/seed-search')
-    starter.assert_awaited_once_with({'cs.CL','cs.IR'}, limit=12)
+    starter.assert_awaited_once_with({'nlp': {'cs.CL','cs.IR'}}, limit=12)
     assert 'An NLP seed' in r.text and 'Saved' in r.text
     assert 'hx-post=' not in r.text
 
@@ -167,3 +167,40 @@ async def test_cookieless_paper_visit_records_nothing(monkeypatch):
     async with aiosqlite.connect(db.DB_PATH) as conn:
         n = (await (await conn.execute("SELECT COUNT(*) FROM interactions")).fetchone())[0]
     assert n == 0
+
+
+def _lanes(monkeypatch, trending_per_code, fresh_per_group):
+    monkeypatch.setattr(discovery_svc.local_meta, 'is_available', lambda: True)
+
+    async def trending(cats, limit):
+        (code,) = cats
+        return [{'arxiv_id': f'{code}:{i}'} for i in range(trending_per_code.get(code, 0))]
+
+    async def fresh(cats, limit):
+        key = '+'.join(sorted(cats))
+        return [{'arxiv_id': f'fresh:{key}:{i}'} for i in range(fresh_per_group.get(key, 0))]
+
+    monkeypatch.setattr(turso_svc, 'fetch_trending_by_categories', trending)
+    monkeypatch.setattr(turso_svc, 'fetch_fresh_by_categories', fresh)
+
+
+async def test_starters_balance_interests_not_arxiv_codes(monkeypatch):
+    """ML spans two codes and robotics one; each interest still gets half."""
+    _lanes(monkeypatch, {'cs.LG': 50, 'stat.ML': 50, 'cs.RO': 50}, {})
+    papers = await discovery_svc.starter_papers(
+        {'ml': {'cs.LG', 'stat.ML'}, 'robotics': {'cs.RO'}}, limit=20)
+    ids = [p['arxiv_id'] for p in papers]
+    assert sum(a.startswith('cs.RO') for a in ids) == 10
+    assert sum(a.startswith(('cs.LG', 'stat.ML')) for a in ids) == 10
+
+
+async def test_every_third_starter_slot_per_interest_is_recent(monkeypatch):
+    _lanes(monkeypatch, {'cs.RO': 50}, {'cs.RO': 50})
+    papers = await discovery_svc.starter_papers({'robotics': {'cs.RO'}}, limit=9)
+    assert [p['arxiv_id'].startswith('fresh:') for p in papers] == [False, False, True] * 3
+
+
+async def test_starters_fall_back_to_established_when_fresh_lane_is_empty(monkeypatch):
+    _lanes(monkeypatch, {'cs.RO': 50}, {})
+    papers = await discovery_svc.starter_papers({'robotics': {'cs.RO'}}, limit=9)
+    assert len(papers) == 9 and not any(p['arxiv_id'].startswith('fresh:') for p in papers)

tests/test_onboarding.py

@@ -136,3 +136,12 @@ def test_expand_category_groups_unknown_key():
     assert "cs.CL" in result
     # unknown key produced nothing extra
     assert len(result) == 2  # cs.CL + cs.IR
+
+
+async def test_get_user_category_groups_keeps_interests_apart(tmp_db):
+    import app.db as db
+    await db.init_db()
+    await db.save_onboarding_categories("u1", ["ml", "robotics"])
+    groups = await db.get_user_category_groups("u1")
+    assert groups == {"ml": {"cs.LG", "stat.ML"}, "robotics": {"cs.RO"}}
+    assert await db.get_user_category_groups("nobody") == {}