keep applicant count fix surgical

app/agent/session.py

@@ -184,6 +184,18 @@ class MatchmakingAgentSession:
                 search.run_sweep(prefs, fresh=fresh),
                 ["Scanning live roles across the boards…", "Pulling the latest Naukri + LinkedIn + Foundit roles…",
                  "Gathering fresh postings for you…"])
+            if sweep.get("all_boards_failed"):
+                await self.push(
+                    "agent_error",
+                    action="search_failed",
+                    message="All enabled job boards are temporarily unavailable.",
+                    data={
+                        "code": "job_boards_unavailable",
+                        "message": "All enabled job boards are temporarily unavailable.",
+                        "failed_boards": sweep.get("failed_boards", 0),
+                    },
+                )
+                return
             new_count = await _repo.pool_save(self.user_id, sig, sweep["opportunities"])
             # tapped out? a sweep that added almost nothing new → flag it so we stop block-fetching dupes.
             if new_count < s.POOL_MIN_NEW_PER_REFILL:
@@ -261,10 +273,12 @@ class MatchmakingAgentSession:
             dedupe_key=f"matchmaking:generated:{self.user_id}:{sig}",
         )
 
-        # The user already has their deck — if the pool dipped below the reorder mark, top it up OUT OF
-        # BAND so the next search stays free and the shelf never runs thin. (No await: fire-and-forget.)
+        # The persisted feed intentionally drops resume context, but a refill must use the same
+        # retrieval inputs as the blocking sweep whose signature owns the pool. Preserve only
+        # the adapter's top-skill input; never retain the rest of the resume context.
         if decision == "pool+refill":
-            asyncio.create_task(self._background_refill(clean_prefs, sig))
+            refill_prefs = {**clean_prefs, "_skills": (user_context or {}).get("skills") or prefs.get("_skills") or []}
+            asyncio.create_task(self._background_refill(refill_prefs, sig))
 
     async def _background_refill(self, prefs: dict, sig: str):
         """Proactive pool top-up: re-fetch the freshest jobs in the background + bank the new ones, so the
@@ -275,6 +289,9 @@ class MatchmakingAgentSession:
         try:
             from app.config import get_settings
             sweep = await search.run_sweep(prefs, fresh=True)
+            if sweep.get("all_boards_failed"):
+                logger.warning("pool refill skipped: all configured job boards failed, sig=%s", sig[:8])
+                return
             new_count = await _repo.pool_save(self.user_id, sig, sweep["opportunities"])
             if new_count < get_settings().POOL_MIN_NEW_PER_REFILL:   # tapped out → relax the floor
                 await _repo.mark_pool_exhausted(self.user_id, sig)

app/config.py

@@ -89,13 +89,13 @@ class Settings(BaseSettings):
     WELLFOUND_MAX_JOBS: int = 15     # blackfalcondata/wellfound-scraper — registered, OFF (US-heavy)
     WORKINDIA_MAX_JOBS: int = 15     # shahidirfan/workindia-jobs-scraper — India blue/grey-collar (niche, low yield)
 
-    # QScore (competence proxy) — consumed, not computed. Dashboard Momentum/QX reads it via REST.
+    # RQ Score (competence proxy) — consumed, not computed. Dashboard Momentum reads it via REST.
     QSCORE_BASE_URL: str = "http://localhost:8004"
     QSCORE_AUTH_TOKEN: str | None = None
     QSCORE_ORG_ID: str = "growqr"
     # user-service — dashboard Day-streak reads metadata.current_streak via GET /api/state/{clerk_id}
     USER_SERVICE_BASE_URL: str = "http://localhost:8003"
-    # dashboard-service — Q-Score trend series via GET /api/qscore-history/{clerk_id}
+    # dashboard-service — RQ Score trend series via GET /api/qscore-history/{clerk_id}
     DASHBOARD_SERVICE_BASE_URL: str = "http://localhost:8005"
     A2A_OUTBOUND_KEY: str = "dev-a2a-key"   # Bearer token presented to sibling services
 

app/db/repo.py

@@ -10,6 +10,7 @@ from sqlalchemy import func, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from app.db.models import Base, OpportunityState, UserFeed, UserJobPool
+from app.engine.board_adapters import coerce as C
 from app.db.session import engine, session_factory
 
 logger = logging.getLogger(__name__)
@@ -17,14 +18,37 @@ logger = logging.getLogger(__name__)
 # fields a caller may set on an opportunity's state (status + the generated document ids + activity flags)
 _STATE_FIELDS = ("status", "tailored_resume_id", "tailored_version_id", "cover_letter_id", "seen", "viewed", "saved")
 
-# the search-defining pref fields — same values → same signature → cursor advances (new jobs);
-# change any → new signature → cursor resets to page 1.
-_SIG_FIELDS = ("title", "role", "location", "industry", "experience", "years", "work_mode", "remote")
+def _effective_title(prefs: dict) -> str:
+    """Match the board adapter's title precedence for cache identity."""
+    target_titles = prefs.get("targetTitles") or []
+    if target_titles and str(target_titles[0]).strip():
+        return str(target_titles[0]).strip()
+    title = str(prefs.get("title") or "").strip()
+    if title:
+        return title
+    for role in prefs.get("role") or []:
+        if role and role != "Open to any":
+            return str(role)
+    return ""
 
 
 def search_signature(prefs: dict) -> str:
-    """Stable short hash of the search-defining prefs (ignores transient flags / user_context)."""
-    payload = {k: prefs.get(k) for k in _SIG_FIELDS}
+    """Stable short hash of every preference that changes raw board retrieval."""
+    # The pool stores unranked jobs. Ranking and curation rerun on every request,
+    # so only adapter inputs belong in this identity. Canonicalize through the
+    # same helpers used by the board adapters to avoid cache-only semantics.
+    user_context = prefs.get("user_context") or {}
+    skills = user_context.get("skills") or prefs.get("_skills") or []
+    retrieval_skill = str(skills[0]).strip().lower() if skills else ""
+    payload = {
+        "title": C.title_of(prefs),
+        "locations": C.parsed_locations(prefs),
+        "workMode": prefs.get("workMode"),
+        "industry": prefs.get("industry"),
+        "experience": C.seniority_band(prefs),
+        "years": C.years_of(prefs),
+        "retrieval_skill": retrieval_skill,
+    }
     blob = json.dumps(payload, sort_keys=True, default=str)
     return hashlib.sha1(blob.encode()).hexdigest()[:16]
 

app/engine/search.py

@@ -116,27 +116,34 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict:
     # fresh=True forces a live fetch + cache overwrite (dev "fresh search" toggle);
     # otherwise honor the configured cache mode (dev replays cached results for $0).
     cache_mode = "refresh" if fresh else None
-    enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip() in BOARDS]
+    configured = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip()]
+    enabled = [b for b in configured if b in BOARDS]
     cap = get_settings().BOARD_TIMEOUT_S
 
     async def _timed(k: str):
         # A single slow/hung board must NOT hold the whole search hostage — cap it and degrade.
         try:
-            return await asyncio.wait_for(_fetch_board(k, prefs, cache_mode), timeout=cap)
+            key, jobs = await asyncio.wait_for(_fetch_board(k, prefs, cache_mode), timeout=cap)
+            return key, jobs, False
         except asyncio.TimeoutError:
             logger.warning("board %s exceeded %ss — skipped (search stays fast)", k, cap)
-            return k, []
+            return k, [], True
 
     results = await asyncio.gather(*(_timed(k) for k in enabled), return_exceptions=True)
 
     merged: list[dict] = []
     seen: set[str] = set()
     sources: dict[str, int] = {}
+    failed_boards = 0
     for r in results:
         if isinstance(r, Exception):
+            failed_boards += 1
             logger.warning("board fetch failed: %s", r)
             continue
-        key, jobs = r
+        key, jobs, failed = r
+        if failed:
+            failed_boards += 1
+            continue
         sources[key] = len(jobs)
         for j in jobs:
             dk = (j.get("apply_url") or "").strip() or f"{j['organization']}|{j['title']}|{j.get('location_city')}".lower()
@@ -146,7 +153,14 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict:
             merged.append(j)
 
     # No ranking here — the engine (app/engine/rank.py) scores + selects after the sweep.
-    return {"opportunities": merged, "sources": sources}
+    return {
+        "opportunities": merged,
+        "sources": sources,
+        "all_boards_failed": not enabled or failed_boards == len(enabled),
+        "failed_boards": failed_boards,
+        "enabled_boards": len(enabled),
+        "configured_boards": configured,
+    }
 
 
 def has_mvq(prefs: dict) -> bool:

app/engine/stats/__init__.py

@@ -4,7 +4,7 @@ Honest sources (SCOUT_UI_SPEC):
 - feed stats (matches / top / range / salary / competition)  ← the persisted feed
 - funnel + engagement rank                                   ← opportunity_state activity (this DB)
 - Apply window (freshness urgency)                           ← listing posting-age
-- Momentum/QX + Readiness trend                              ← qscore-service   (Phase 1)
+- Momentum/RQ Score + Readiness trend                       ← qscore-service   (Phase 1)
 - Day streak                                                 ← dashboard-service (Phase 1)
 
 Anything without a real source is omitted — the UI cuts or locks it, never fakes it.
@@ -13,6 +13,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import math
 from datetime import datetime, timezone
 
 from app.db import repo as _repo
@@ -78,7 +79,17 @@ async def build_scout_stats(user_id: str, *, user_uuid: str | None = None) -> di
     opps = feed.get("opportunities") or []
 
     scores = _match_scores(opps)
-    apps = [j.get("applicants") for j in opps if j.get("applicants")]
+    apps = []
+    for opportunity in opps:
+        raw_applicants = opportunity.get("applicants")
+        if isinstance(raw_applicants, bool):
+            continue
+        try:
+            applicants = float(raw_applicants)
+        except (TypeError, ValueError):
+            continue
+        if math.isfinite(applicants) and applicants >= 0:
+            apps.append(applicants)
     matches = len(opps)
 
     return {
@@ -109,6 +120,6 @@ async def build_scout_stats(user_id: str, *, user_uuid: str | None = None) -> di
         # ── cross-service (real; None → the UI omits the card, never fakes it) ──
         "qxNow": (qx or {}).get("qx"),
         "quotients": (qx or {}).get("quotients"),
-        "qxTrend": qx_trend,            # Q-Score series (dashboard-service) → Readiness-trend sparkline
+        "qxTrend": qx_trend,            # RQ Score series (qscore-service) → Readiness-trend sparkline
         "dayStreak": streak,
     }

app/engine/stats/clients.py

@@ -1,4 +1,4 @@
-"""Cross-service reads for the dashboard — QScore (Momentum/QX) + user-service (Day streak).
+"""Cross-service reads for the dashboard — RQ Score (Momentum/QX) + user-service (Day streak).
 
 Direct REST, Clerk-mesh-authed. Both degrade to None on any failure (timeout / 404 / service down) —
 a missing KPI is omitted by the UI, never faked, and never crashes the dashboard.
@@ -6,6 +6,8 @@ a missing KPI is omitted by the UI, never faked, and never crashes the dashboard
 from __future__ import annotations
 
 import logging
+import math
+from collections.abc import Mapping
 
 import httpx
 
@@ -14,8 +16,19 @@ from app.config import get_settings
 logger = logging.getLogger(__name__)
 
 
+def _score_value(value: object) -> float | None:
+    """Parse one canonical RQ Score value; malformed values are absent."""
+    if isinstance(value, bool) or value is None:
+        return None
+    try:
+        score = float(value)
+    except (TypeError, ValueError):
+        return None
+    return score if math.isfinite(score) else None
+
+
 async def fetch_qx(user_uuid: str | None) -> dict | None:
-    """QScore for the user → {qx, quotients}. GET /v1/qscore/{uuid}?org_id (qscore-service)."""
+    """RQ Score for the user → {qx, quotients}. GET /v1/qscore/{uuid}?org_id (qscore-service)."""
     if not user_uuid:
         return None
     s = get_settings()
@@ -25,16 +38,21 @@ async def fetch_qx(user_uuid: str | None) -> dict | None:
         if r.status_code == 404:        # no score computed yet — honest absence
             return None
         r.raise_for_status()
-        d = r.json()
-        qx = d.get("q_score")
-        return {"qx": round(float(qx), 1), "quotients": d.get("quotients") or {}} if qx is not None else None
-    except (httpx.HTTPError, ValueError) as e:
+        payload = r.json()
+        if not isinstance(payload, Mapping):
+            return None
+        qx = _score_value(payload.get("rq_score"))
+        if qx is None:
+            return None
+        quotients = payload.get("quotients")
+        return {"qx": round(qx, 1), "quotients": quotients if isinstance(quotients, Mapping) else {}}
+    except (httpx.HTTPError, ValueError, TypeError) as e:
         logger.warning("fetch_qx failed (non-fatal): %s", e)
         return None
 
 
 async def fetch_qx_trend(user_uuid: str | None) -> list[float] | None:
-    """Q-Score trend (oldest→newest) ← qscore-service GET /v1/qscore/{uuid}/history. Same source as
+    """RQ Score trend (oldest→newest) ← qscore-service GET /v1/qscore/{uuid}/history. Same source as
     the live QX, so the trend's latest point matches the Momentum value (consistent, not stale).
     None on failure → the UI keeps the Readiness card locked."""
     if not user_uuid:
@@ -48,10 +66,21 @@ async def fetch_qx_trend(user_uuid: str | None) -> list[float] | None:
             )
         if r.status_code == 404:
             return None
-        r.raise_for_status()
-        pts = [round(float(p["q_score"]), 1) for p in (r.json().get("points") or []) if p.get("q_score") is not None]
+        payload = r.json()
+        if not isinstance(payload, Mapping):
+            return None
+        points = payload.get("points")
+        if not isinstance(points, list):
+            return None
+        pts = []
+        for point in points:
+            if not isinstance(point, Mapping):
+                continue
+            score = _score_value(point.get("rq_score"))
+            if score is not None:
+                pts.append(round(score, 1))
         return pts if len(pts) >= 2 else None     # need ≥2 points for a trend
-    except (httpx.HTTPError, ValueError, TypeError, KeyError) as e:
+    except (httpx.HTTPError, ValueError, TypeError) as e:
         logger.warning("fetch_qx_trend failed (non-fatal): %s", e)
         return None
 

docker-compose.staging.yml

@@ -35,7 +35,7 @@ services:
         condition: service_healthy
       redis:
         condition: service_started
-    command: uvicorn app.main:app --host 0.0.0.0 --port 8000
+    command: sh -c "alembic upgrade head && exec uvicorn app.main:app --host 0.0.0.0 --port 8000"
     restart: unless-stopped
 
   postgres:

docker-compose.yml

@@ -10,6 +10,7 @@ services:
     environment:
       # inside the compose network the DB host is `postgres` (overrides the host-dev .env value)
       DATABASE_URL: postgresql+asyncpg://matchmaking:matchmaking@postgres:5432/matchmaking
+    command: sh -c "alembic upgrade head && exec uvicorn app.main:app --host 0.0.0.0 --port 8000"
     depends_on:
       postgres:
         condition: service_healthy

tests/test_pool.py

@@ -1,5 +1,13 @@
 """Warm-pool gate — the reorder/floor decision that keeps the shelf stocked without over-fetching."""
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from app.engine import search
 from app.engine.search import pool_decision
+from app.db.repo import search_signature
 
 FLOOR, REORDER = 90, 100   # the configured thresholds (POOL_SAFETY_FLOOR / POOL_REORDER_AT)
 
@@ -39,3 +47,218 @@ def test_floor_never_above_reorder_invariant():
     from app.config import get_settings
     s = get_settings()
     assert s.POOL_SAFETY_FLOOR <= s.POOL_REORDER_AT
+
+
+def test_target_title_changes_search_signature():
+    base = {
+        "title": "Product Manager",
+        "role": ["Product"],
+        "targetTitles": ["Backend Engineer"],
+        "location": ["Bangalore · India"],
+        "workMode": ["Hybrid"],
+    }
+    changed = {**base, "targetTitles": ["Data Analyst"]}
+
+    assert search_signature(base) != search_signature(changed)
+
+
+def test_work_mode_changes_search_signature_using_dashboard_contract():
+    base = {"title": "Data Analyst", "location": ["Bangalore · India"], "workMode": ["Hybrid"]}
+    changed = {**base, "workMode": ["Remote"]}
+
+    assert search_signature(base) != search_signature(changed)
+
+
+def test_legacy_work_mode_does_not_change_adapter_cache_identity():
+    common = {"title": "Data Analyst", "location": ["Bangalore · India"]}
+
+    assert search_signature(common) == search_signature({**common, "work_mode": ["Hybrid"]})
+
+
+def test_retrieval_inputs_change_search_signature():
+    base = {
+        "title": "Data Analyst",
+        "location": ["Bangalore · India"],
+        "workMode": ["Hybrid"],
+        "industry": ["Fintech"],
+        "experience": ["Mid-level"],
+        "years": 5,
+        "user_context": {"skills": ["Python", "SQL"]},
+    }
+    changes = [
+        {"title": "Product Manager"},
+        {"location": ["Mumbai · India"]},
+        {"workMode": ["Remote"]},
+        {"industry": ["Healthcare"]},
+        {"experience": ["Senior"]},
+        {"years": 8},
+        {"user_context": {"skills": ["Rust", "SQL"]}},
+    ]
+
+    for change in changes:
+        assert search_signature(base) != search_signature({**base, **change})
+
+
+def test_rank_only_preferences_do_not_cold_start_raw_pool():
+    base = {"title": "Data Analyst", "location": ["Bangalore · India"]}
+    changed = {**base, "targetComp": "₹22–30L", "dealBreakers": ["Travel"], "stretch": "reach"}
+
+    assert search_signature(base) == search_signature(changed)
+
+
+def test_search_signature_ignores_unrelated_user_context():
+    base = {"title": "Data Analyst", "location": ["Bangalore · India"], "user_context": {"skills": ["Python"]}}
+    changed = {**base, "user_context": {"skills": ["Python"], "education": ["MBA"]}}
+
+    assert search_signature(base) == search_signature(changed)
+
+
+def test_sweep_marks_total_board_exception_failure(monkeypatch):
+    async def fail_board(*_args, **_kwargs):
+        raise RuntimeError("provider blocked")
+
+    monkeypatch.setattr(search, "_fetch_board", fail_board)
+    monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace(
+        BOARDS_ENABLED="naukri,foundit",
+        BOARD_TIMEOUT_S=1,
+    ))
+
+    result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}))
+
+    assert result["all_boards_failed"] is True
+    assert result["failed_boards"] == 2
+    assert result["sources"] == {}
+
+
+def test_sweep_marks_total_board_timeout_failure(monkeypatch):
+    async def timeout_board(*_args, **_kwargs):
+        await asyncio.sleep(0.05)
+        return "naukri", []
+
+    monkeypatch.setattr(search, "_fetch_board", timeout_board)
+    monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace(
+        BOARDS_ENABLED="naukri,foundit",
+        BOARD_TIMEOUT_S=0.001,
+    ))
+    result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}))
+
+    assert result["all_boards_failed"] is True
+    assert result["failed_boards"] == 2
+
+
+def test_sweep_marks_empty_or_unknown_board_config_as_failure(monkeypatch):
+    monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace(
+        BOARDS_ENABLED="unknown-board",
+        BOARD_TIMEOUT_S=1,
+    ))
+
+    result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}))
+
+    assert result["all_boards_failed"] is True
+    assert result["enabled_boards"] == 0
+    assert result["configured_boards"] == ["unknown-board"]
+
+
+def test_total_outage_emits_frontend_readable_agent_error(monkeypatch):
+    from app.agent.session import MatchmakingAgentSession
+
+    ws = MagicMock()
+    ws.send_json = AsyncMock()
+    session = MatchmakingAgentSession(ws, "user-1")
+    async def outage_heartbeat(awaitable, _messages):
+        awaitable.close()
+        return {
+            "opportunities": [],
+            "sources": {},
+            "all_boards_failed": True,
+            "failed_boards": 2,
+        }
+
+    session._with_heartbeat = outage_heartbeat
+    monkeypatch.setattr("app.agent.session.get_settings", lambda: SimpleNamespace(
+        POOL_ENABLED=False,
+        POOL_SAFETY_FLOOR=90,
+        POOL_REORDER_AT=100,
+    ), raising=False)
+    monkeypatch.setattr("app.config.get_settings", lambda: SimpleNamespace(
+        POOL_ENABLED=False,
+        POOL_SAFETY_FLOOR=90,
+        POOL_REORDER_AT=100,
+    ))
+    monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=set()))
+
+    asyncio.run(session.handle_run_search({"title": "Data Analyst", "location": ["Bangalore · India"]}))
+
+    message = ws.send_json.await_args.args[0]
+    assert message["type"] == "agent_error"
+    assert message["message"] == "All enabled job boards are temporarily unavailable."
+    assert message["data"]["code"] == "job_boards_unavailable"
+
+
+def test_background_refill_does_not_mark_outage_as_exhausted(monkeypatch):
+    from app.agent.session import MatchmakingAgentSession
+
+    session = MatchmakingAgentSession(MagicMock(), "user-1")
+    monkeypatch.setattr("app.engine.search.run_sweep", AsyncMock(return_value={
+        "opportunities": [],
+        "sources": {},
+        "all_boards_failed": True,
+    }))
+    save = AsyncMock()
+    exhausted = AsyncMock()
+    monkeypatch.setattr("app.db.repo.pool_save", save)
+    monkeypatch.setattr("app.db.repo.mark_pool_exhausted", exhausted)
+
+    asyncio.run(session._background_refill({"title": "Data Analyst"}, "signature"))
+
+    save.assert_not_awaited()
+    exhausted.assert_not_awaited()
+
+
+def test_background_refill_preserves_signature_retrieval_skill(monkeypatch):
+    from app.agent.session import MatchmakingAgentSession
+    ws = MagicMock()
+    ws.send_json = AsyncMock()
+    session = MatchmakingAgentSession(ws, "user-1")
+    prefs = {
+        "title": "Data Analyst",
+        "location": ["Bangalore · India"],
+        "user_context": {"skills": ["Python", "SQL"], "education": ["MBA"]},
+    }
+    sig = search_signature(prefs)
+    captured = {}
+
+    monkeypatch.setattr("app.config.get_settings", lambda: SimpleNamespace(
+        POOL_ENABLED=True,
+        POOL_SAFETY_FLOOR=90,
+        POOL_REORDER_AT=100,
+    ))
+    monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=set()))
+    monkeypatch.setattr("app.db.repo.pool_load_fresh_unseen", AsyncMock(return_value=[{"id": f"job-{i}"} for i in range(95)]))
+    monkeypatch.setattr("app.db.repo.is_pool_exhausted", AsyncMock(return_value=False))
+    monkeypatch.setattr("app.db.repo.save_feed", AsyncMock())
+    monkeypatch.setattr("app.db.repo.update_salary_band", AsyncMock())
+    monkeypatch.setattr("app.db.repo.mark_seen", AsyncMock())
+    monkeypatch.setattr("app.db.repo.get_opportunity_states", AsyncMock(return_value={}))
+    monkeypatch.setattr("app.agent.session.emit_grow_event", AsyncMock())
+    monkeypatch.setattr("app.engine.sift.sift", lambda *_args: ([{"id": "job-1"}], {"mode": "test", "scored": 1}))
+    monkeypatch.setattr("app.engine.curate.curate", lambda *_args: [{"id": "job-1"}])
+
+    async def capture_refill(refill_prefs, refill_sig):
+        captured["prefs"] = refill_prefs
+        captured["sig"] = refill_sig
+
+    async def direct_heartbeat(awaitable, _messages):
+        return await awaitable
+
+    session._background_refill = capture_refill
+    session._with_heartbeat = direct_heartbeat
+    session._grow_activity_fields = AsyncMock(return_value={})
+
+    asyncio.run(session.handle_run_search(prefs))
+    asyncio.run(asyncio.sleep(0))
+
+    assert captured["sig"] == sig
+    assert captured["prefs"]["_skills"] == ["Python", "SQL"]
+    assert "user_context" not in captured["prefs"]
+    assert search_signature(captured["prefs"]) == sig

tests/test_stats.py

@@ -1,5 +1,10 @@
-"""Dashboard stats — pure-helper contract tests (no DB / no network)."""
+"""Dashboard stats — pure-helper and RQ Score client contract tests."""
+import asyncio
+
+import httpx
+
 from app.engine import stats as S
+from app.engine.stats import clients
 from app.engine.normalize import _posted_date
 from app.db.repo import _activity_score
 
@@ -20,6 +25,25 @@ def test_match_scores_and_histogram():
     assert sum(h) == 2 and len(h) == 11
 
 
+def test_scout_stats_accepts_numeric_applicant_strings(monkeypatch):
+    feed = {"opportunities": [{"applicants": "10"}, {"applicants": 30}, {"applicants": "unknown"}]}
+    activity = {
+        "matches": 3, "viewed": 0, "saved": 0, "applied": 0, "tailored": 0,
+        "searches": 1, "percentile": None, "cohort_size": 1,
+    }
+    monkeypatch.setattr(S._repo, "get_feed", lambda _user_id: asyncio.sleep(0, result=feed))
+    monkeypatch.setattr(S._repo, "get_activity_stats", lambda _user_id: asyncio.sleep(0, result=activity))
+    monkeypatch.setattr(S._repo, "get_active_window", lambda _user_id: asyncio.sleep(0, result={}))
+    monkeypatch.setattr(S._repo, "get_salary_band", lambda _user_id: asyncio.sleep(0, result=None))
+    monkeypatch.setattr(S._clients, "fetch_qx", lambda _user_uuid: asyncio.sleep(0, result=None))
+    monkeypatch.setattr(S._clients, "fetch_qx_trend", lambda _user_uuid: asyncio.sleep(0, result=None))
+    monkeypatch.setattr(S._clients, "fetch_streak", lambda _user_id: asyncio.sleep(0, result=None))
+
+    result = asyncio.run(S.build_scout_stats("user-1"))
+
+    assert result["avgApplicants"] == 20
+
+
 def test_apply_window_freshness():
     # all None when no posting dates (honest — never fabricated)
     assert S._apply_window([{}, {"posted_date": None}])["urgency"] is None
@@ -31,3 +55,70 @@ def test_activity_score_weighting():
     # applies weighted highest, then saves, views, searches
     assert _activity_score(0, 0, 1, 0) > _activity_score(0, 1, 0, 0) > _activity_score(1, 0, 0, 0)
     assert _activity_score(0, 0, 0, 0) == 0
+
+
+class _Response:
+    status_code = 200
+
+    def __init__(self, payload):
+        self._payload = payload
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self._payload
+
+
+def test_fetch_qx_reads_strict_rq_score_contract(monkeypatch):
+    async def _get(_self, _url, **_kwargs):
+        return _Response({"rq_score": 68.25, "quotients": {"technical": 71}})
+
+    monkeypatch.setattr(httpx.AsyncClient, "get", _get)
+
+    assert asyncio.run(clients.fetch_qx("user-uuid")) == {
+        "qx": 68.2,
+        "quotients": {"technical": 71},
+    }
+
+
+def test_fetch_qx_does_not_alias_legacy_q_score(monkeypatch):
+    async def _get(_self, _url, **_kwargs):
+        return _Response({"q_score": 99, "quotients": {"technical": 99}})
+
+    monkeypatch.setattr(httpx.AsyncClient, "get", _get)
+
+    assert asyncio.run(clients.fetch_qx("user-uuid")) is None
+
+
+def test_fetch_qx_treats_malformed_payload_as_absent(monkeypatch):
+    async def _get(_self, _url, **_kwargs):
+        return _Response([{"rq_score": 88}])
+
+    monkeypatch.setattr(httpx.AsyncClient, "get", _get)
+
+    assert asyncio.run(clients.fetch_qx("user-uuid")) is None
+
+
+def test_fetch_qx_trend_ignores_malformed_and_absent_points(monkeypatch):
+    async def _get(_self, _url, **_kwargs):
+        return _Response({"points": [None, {"rq_score": "bad"}, {"rq_score": 35}, {"rq_score": 40}]})
+
+    monkeypatch.setattr(httpx.AsyncClient, "get", _get)
+
+    assert asyncio.run(clients.fetch_qx_trend("user-uuid")) == [35.0, 40.0]
+
+
+def test_fetch_qx_trend_reads_strict_rq_score_points(monkeypatch):
+    async def _get(_self, _url, **_kwargs):
+        return _Response({
+            "points": [
+                {"rq_score": 35, "q_score": 91},
+                {"rq_score": 44.16, "q_score": 92},
+                {"q_score": 93},
+            ]
+        })
+
+    monkeypatch.setattr(httpx.AsyncClient, "get", _get)
+
+    assert asyncio.run(clients.fetch_qx_trend("user-uuid")) == [35.0, 44.2]