keep applicant count fix surgical

Root cause: the VPS ran two Compose projects from the same file — api under
-p matchmaking-v2-staging, postgres/redis under -p matchmaking-v2 — creating
two disjoint <project>_default bridges. Every dependency hostname (postgres,
redis) was NXDOMAIN from the api container, so feed store init, pool_load,
save_feed, and RedisStreamWorker all failed with [Errno -2].

Fix: standardize on a single project (matchmaking-v2) for all three services
with an external shared bridge (matchmaking-v2_default) and inline DB/Redis
host pinning. The deploy script (sync-staging.sh) now brings up all three
services, pre-creates the external network, and has a public_health_url case.

Apify: moved the token from query params (4 sites) to an Authorization header
so no request URL can leak it. All raise_for_status() paths now raise
token-free _ApifyHTTPError. The live 403 (platform-feature-disabled: monthly
usage cap exceeded) is an external blocker — no code fix.

Tests: 15 focused regression tests covering network topology, volume config,
deploy-script alignment, and Apify token redaction (with call-counters).

.env.example

@@ -14,6 +14,12 @@ REDIS_URL=redis://localhost:6379/0
 # Inputs / engine (wired in later slices)
 APIFY_TOKEN=
 OPENAI_API_KEY=
+OPENAI_API_BASE=
+DSPY_API_BASE=
+DSPY_API_KEY=
+CURATE_MODEL=gpt-4o-mini
+SUGGEST_MODEL=gpt-4o-mini
+EMBED_MODEL=text-embedding-3-small
 
 # QScore (consumed)
 QSCORE_BASE_URL=http://localhost:8004

app/agent/session.py

@@ -12,6 +12,8 @@ from __future__ import annotations
 import asyncio
 import logging
 
+from app.grow_events import emit_grow_event
+
 logger = logging.getLogger(__name__)
 
 
@@ -20,6 +22,34 @@ class MatchmakingAgentSession:
         self.ws = websocket
         self.user_id = user_id
         self.context: dict = {}
+    async def _grow_activity_fields(self) -> dict:
+        """Cheap cumulative activity context shared across Grow payloads — the funnel counts the
+        backend extractor maps to qscore signals (applications_submitted, opportunities_viewed).
+        Degrades to empty dict if the DB is off."""
+        from app.db.repo import get_activity_stats
+        try:
+            act = await get_activity_stats(self.user_id)
+            return {
+                "applications_submitted": act.get("applied", 0),   # ← projector reads applications_submitted
+                "opportunities_viewed": act.get("viewed", 0),
+                "jobs_viewed": act.get("viewed", 0),               # ← projector reads jobs_viewed
+                "opportunities_saved": act.get("saved", 0),
+                "searches": act.get("searches", 0),
+                "matches_generated": act.get("matches", 0),
+                "matchCount": act.get("matches", 0),               # ← projector reads matchCount
+            }
+        except Exception:  # noqa: BLE001 — never fatal
+            return {}
+
+    @staticmethod
+    def _match_scores(opps: list[dict]) -> list[int]:
+        """Extract integer match scores from opportunities for source score fields."""
+        out = []
+        for o in opps:
+            s = (o.get("match") or {}).get("score") or o.get("matchScore")
+            if isinstance(s, (int, float)) and s > 0:
+                out.append(int(s))
+        return out
 
     async def push(self, msg_type: str, **kwargs):
         await self.ws.send_json({"type": msg_type, **kwargs})
@@ -94,6 +124,25 @@ class MatchmakingAgentSession:
         from app.db.repo import get_feed
         feed = await get_feed(self.user_id)
         await self.push("agent_data", action="feed_loaded", data=feed or {"opportunities": [], "cached": False})
+        feed_opps = (feed or {}).get("opportunities") or []
+        feed_scores = self._match_scores(feed_opps)
+        await emit_grow_event(
+            "matchmaking.feed.viewed",
+            self.user_id,
+            payload={
+                "action": "get_feed",
+                "match_count": len(feed_opps),
+                "cached": bool((feed or {}).get("cached", True)),
+                "source_scores": feed_scores,
+                "matchMin": min(feed_scores) if feed_scores else None,
+                "matchMax": max(feed_scores) if feed_scores else None,
+                "match_min": min(feed_scores) if feed_scores else None,
+                "match_max": max(feed_scores) if feed_scores else None,
+                **await self._grow_activity_fields(),
+            },
+            subject={"serviceId": "matchmaking-service", "externalId": "feed"},
+            dedupe_key=f"matchmaking:feed:{self.user_id}:{(feed or {}).get('updated_at') or 'none'}",
+        )
 
     async def handle_record_feedback(self, params: dict):
         await self._stub("feedback_recorded", "store SAVE/DISMISS/APPLY label")
@@ -135,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:
@@ -184,11 +245,40 @@ class MatchmakingAgentSession:
         result["opportunities"] = _repo.annotate_and_filter(result["opportunities"], states)
         result["shortlisted"] = len(result["opportunities"])
         await self.push("agent_data", action="search_complete", data=result)
+        # Grow event: matches were generated for this user → backend extracts qscore signals
+        gen_opps = result["opportunities"] or []
+        gen_scores = self._match_scores(gen_opps)
+        await emit_grow_event(
+            "matchmaking.matches.generated",
+            self.user_id,
+            payload={
+                "action": "run_search",
+                "matchCount": len(gen_opps),          # ← projector reads matchCount/matches/shortlisted
+                "match_count": len(gen_opps),          # snake_case alias
+                "scanned": result.get("scanned"),
+                "total_candidates": result.get("scanned"),  # ← projector reads total_candidates for match_rate
+                "shortlisted": result.get("shortlisted"),
+                "sources": result.get("sources"),
+                "engine": result.get("engine"),
+                "source_scores": gen_scores,
+                "matchMin": min(gen_scores) if gen_scores else None,   # ← scout_stats camelCase
+                "matchMax": max(gen_scores) if gen_scores else None,
+                "match_min": min(gen_scores) if gen_scores else None,  # snake_case alias
+                "match_max": max(gen_scores) if gen_scores else None,
+                "top_match_score": max(gen_scores) if gen_scores else None,
+                **await self._grow_activity_fields(),
+            },
+            subject={"serviceId": "matchmaking-service", "externalId": sig},
+            correlation={"taskId": sig},
+            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
@@ -199,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)
@@ -246,6 +339,17 @@ class MatchmakingAgentSession:
         await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), status="applied")
         await self.push("agent_data", action="opportunity_state_saved",
                         data={"opportunity_id": params.get("opportunity_id"), "status": "applied"})
+        await emit_grow_event(
+            "matchmaking.match.applied",
+            self.user_id,
+            payload={
+                "opportunity_id": params.get("opportunity_id"),
+                "status": "applied",
+                **await self._grow_activity_fields(),
+            },
+            subject={"serviceId": "matchmaking-service", "externalId": params.get("opportunity_id", "")},
+            dedupe_key=f"matchmaking:applied:{self.user_id}:{params.get('opportunity_id', '')}",
+        )
 
     async def handle_save_apply_docs(self, params: dict):
         """Remember the resume-builder document ids generated for a role so we never re-pay to
@@ -262,6 +366,17 @@ class MatchmakingAgentSession:
         """The user opened a job → funnel 'Viewed'. Fire-and-forget (no response needed)."""
         from app.db.repo import set_opportunity_state
         await set_opportunity_state(self.user_id, params.get("opportunity_id", ""), viewed=True)
+        await emit_grow_event(
+            "matchmaking.match.viewed",
+            self.user_id,
+            payload={
+                "opportunity_id": params.get("opportunity_id"),
+                "action": "viewed",
+                **await self._grow_activity_fields(),
+            },
+            subject={"serviceId": "matchmaking-service", "externalId": params.get("opportunity_id", "")},
+            dedupe_key=f"matchmaking:viewed:{self.user_id}:{params.get('opportunity_id', '')}",
+        )
 
     async def handle_mark_saved(self, params: dict):
         """The user saved/swiped-right a job → funnel 'Shortlisted'."""

app/config.py

@@ -31,14 +31,14 @@ class Settings(BaseSettings):
     # Inputs / engine
     APIFY_TOKEN: str | None = None
     OPENAI_API_KEY: str | None = None
+    OPENAI_API_BASE: str | None = None
 
-    # LLM gateway (opencode.ai/zen — OpenAI-compatible; reuse interview-service's DSPY key).
-    # One `openai` client (base_url override) serves the Opus curator + embeddings.
+    # LLM gateway override for chat curation. Leave blank to use OPENAI_API_KEY directly.
     DSPY_API_BASE: str | None = None
     DSPY_API_KEY: str | None = None
-    CURATE_MODEL: str = "claude-opus-4-8"          # the senior recruiter (Opus, latest) — Opus's call is final
-    SUGGEST_MODEL: str = "claude-haiku-4-5"        # the bubbling engine — fast + cheap (latency-sensitive UI)
-    EMBED_MODEL: str = "text-embedding-3-small"    # the vibe-engineer (direct OpenAI — gateway has no embeddings route)
+    CURATE_MODEL: str = "gpt-4o-mini"             # the senior recruiter
+    SUGGEST_MODEL: str = "gpt-4o-mini"            # the bubbling engine
+    EMBED_MODEL: str = "text-embedding-3-small"   # direct OpenAI embeddings
     ENGINE_LLM_ENABLED: bool = True                # off / key absent → white-box sift + templated cards (safety net)
     SIFT_TOP_K: int = 22                           # candidate pool the curator reads — widened from 18 for
                                                    # recall, trimmed from 28 to keep the Opus payload lean
@@ -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/apify_client.py

@@ -51,13 +51,31 @@ _RUNS_BASE = "https://api.apify.com/v2"
 _POLL = 3.0  # async-run poll interval (s)
 
 
+def _auth_headers(token: str) -> dict:
+    """Bearer auth — keeps the token OUT of request URLs (and thus out of error/log strings).
+
+    Passing the token as a query param puts it in resp.request.url, which httpx embeds in
+    HTTPStatusError.__str__ — so any logged error would leak the credential. The Apify v2 API
+    accepts a Bearer header for every endpoint, so this is the single-point fix.
+    """
+    return {"Authorization": f"Bearer {token}"}
+
+
+class _ApifyHTTPError(RuntimeError):
+    """Token-free HTTP error raised in place of httpx.HTTPStatusError (whose str includes the URL)."""
+
+
 async def _run_actor_async(actor_id: str, run_input: dict, *, timeout: float, token: str) -> list[dict]:
     """Start an ASYNC run, poll to completion, return its dataset items. Needed for incremental /
     stateful actors (state persisted across runs via `stateKey`) whose crawl exceeds the run-sync
     window — that window aborts the run mid-crawl. No cache: the actor itself is the dedup."""
-    async with httpx.AsyncClient(timeout=60.0) as client:
-        r = await client.post(f"{_BASE}/{actor_id}/runs", params={"token": token}, json=run_input)
-        r.raise_for_status()
+    headers = _auth_headers(token)
+    async with httpx.AsyncClient(timeout=60.0, headers=headers) as client:
+        r = await client.post(f"{_BASE}/{actor_id}/runs", json=run_input)
+        try:
+            r.raise_for_status()
+        except httpx.HTTPStatusError:
+            raise _ApifyHTTPError(f"apify async run-start {actor_id}: HTTP {r.status_code}") from None
         run = r.json().get("data", {})
         run_id = run.get("id")
         ds_id = run.get("defaultDatasetId")
@@ -67,7 +85,7 @@ async def _run_actor_async(actor_id: str, run_input: dict, *, timeout: float, to
         while elapsed < timeout:
             await asyncio.sleep(_POLL)
             elapsed += _POLL
-            s = await client.get(f"{_RUNS_BASE}/actor-runs/{run_id}", params={"token": token})
+            s = await client.get(f"{_RUNS_BASE}/actor-runs/{run_id}")
             data = s.json().get("data", {})
             status = data.get("status")
             ds_id = data.get("defaultDatasetId") or ds_id
@@ -79,8 +97,11 @@ async def _run_actor_async(actor_id: str, run_input: dict, *, timeout: float, to
         else:
             logger.warning("apify async run %s did not finish within %ss", actor_id, timeout)
             return []
-        d = await client.get(f"{_RUNS_BASE}/datasets/{ds_id}/items", params={"token": token, "format": "json"})
-        d.raise_for_status()
+        d = await client.get(f"{_RUNS_BASE}/datasets/{ds_id}/items", params={"format": "json"})
+        try:
+            d.raise_for_status()
+        except httpx.HTTPStatusError:
+            raise _ApifyHTTPError(f"apify async dataset {actor_id}: HTTP {d.status_code}") from None
         items = d.json()
         n = len(items) if isinstance(items, list) else 0
         logger.info("apify async run %s SUCCEEDED — %d items (incremental)", actor_id, n)
@@ -114,15 +135,18 @@ async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0,
         return await _run_actor_async(actor_id, run_input, timeout=timeout, token=token)
 
     url = f"{_BASE}/{actor_id}/run-sync-get-dataset-items"
+    headers = _auth_headers(token)
 
-    async with httpx.AsyncClient(timeout=timeout) as client:
+    async with httpx.AsyncClient(timeout=timeout, headers=headers) as client:
         for attempt in range(1, _MAX_ATTEMPTS + 1):
             try:
-                resp = await client.post(url, params={"token": token}, json=run_input)
+                resp = await client.post(url, json=run_input)
                 if resp.status_code >= 500:
-                    raise httpx.HTTPStatusError(
-                        f"{resp.status_code} from Apify", request=resp.request, response=resp)
-                resp.raise_for_status()
+                    raise _ApifyHTTPError(f"{resp.status_code} from Apify")
+                try:
+                    resp.raise_for_status()
+                except httpx.HTTPStatusError:
+                    raise _ApifyHTTPError(f"apify {actor_id}: HTTP {resp.status_code}") from None
                 items = resp.json()
                 items = items if isinstance(items, list) else []
                 if path and mode in ("readwrite", "refresh"):
@@ -135,4 +159,10 @@ async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0,
                     await asyncio.sleep(_BACKOFF)
                     continue
                 raise
+            except _ApifyHTTPError as e:
+                if attempt < _MAX_ATTEMPTS and "from Apify" in str(e):
+                    logger.warning("apify %s: %s — retrying once", actor_id, e)
+                    await asyncio.sleep(_BACKOFF)
+                    continue
+                raise
     return []  # pragma: no cover

app/engine/curate.py

@@ -121,10 +121,12 @@ def _parse_json(text: str) -> dict:
     return json.loads(t)
 
 
-def _salvage_objects(text: str, *, after: str | None = None) -> list[dict]:
+def _salvage_objects(text: str | None, *, after: str | None = None) -> list[dict]:
     """Recover valid objects from malformed/truncated model JSON: brace-scan each top-level {...} and
     json.loads it on its own, skipping the one broken/cut-off object. `after` anchors the scan past a key
     (e.g. '"kept"'); omit it to scan the whole text (the cards stage uses '"cards"', not '"kept"')."""
+    if not text:
+        return []
     s = text
     if after:
         i = text.find(after)
@@ -145,9 +147,7 @@ def _salvage_objects(text: str, *, after: str | None = None) -> list[dict]:
                     pass
                 start = None
     return out
-
-
-def _salvage_kept(text: str) -> list[dict]:
+def _salvage_kept(text: str | None) -> list[dict]:
     """Score-stage salvage — recover kept-job objects (anchored after the '"kept"' key)."""
     return _salvage_objects(text, after='"kept"')
 
@@ -216,12 +216,15 @@ def _score_stage(prefs: dict, ctx: dict | None, sifted_jobs: list[dict], s) -> l
     except Exception as e:  # noqa: BLE001
         logger.warning("score stage (Opus) failed, falling back to sift: %s", e)
         return None
+    if not content or not content.strip():
+        logger.warning("score stage LLM returned empty content — falling back to sift")
+        return None
     try:
         kept = _parse_json(content).get("kept") or []
     except Exception:  # noqa: BLE001
         kept = _salvage_kept(content)
         if not kept:
-            logger.warning("score stage unparseable — falling back to sift")
+            logger.warning("score stage unparseable — falling back to sift (content[:200]=%r)", content[:200])
             return None
         logger.warning("score stage JSON malformed; salvaged %d items", len(kept))
     # Floor check on the RAW computed score (weak never shown), THEN the transparent presentation
@@ -256,6 +259,9 @@ def _cards_stage(prefs: dict, ctx: dict | None, scored: list[dict], by_id: dict,
     except Exception as e:  # noqa: BLE001 — prose is non-critical; degrade to templated cards
         logger.warning("cards stage (Haiku) call failed, using evidence-based fallback prose: %s", e)
         return {}
+    if not content or not content.strip():
+        logger.warning("cards stage LLM returned empty content — using evidence-based fallback prose")
+        return {}
     try:
         cards = _parse_json(content).get("cards") or []
     except Exception:  # noqa: BLE001

app/engine/embed.py

@@ -10,6 +10,37 @@ from app.config import get_settings
 from app.engine.llm import embed_client
 
 
+def _embedding_vectors(s, texts: list[str]) -> list[list[float]] | None:
+    if s.OPENAI_API_BASE:
+        import httpx
+
+        try:
+            resp = httpx.post(
+                s.OPENAI_API_BASE.rstrip("/") + "/embeddings",
+                headers={
+                    "Authorization": f"Bearer {s.OPENAI_API_KEY}",
+                    "Content-Type": "application/json",
+                    "HTTP-Referer": "https://sai-onchain.me",
+                    "X-Title": "Sai Onchain Matchmaking",
+                },
+                json={"model": s.EMBED_MODEL, "input": texts},
+                timeout=30,
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            return [item["embedding"] for item in data.get("data", [])]
+        except Exception:
+            return None
+    client = embed_client()
+    if not client:
+        return None
+    try:
+        resp = client.embeddings.create(model=s.EMBED_MODEL, input=texts)
+        return [d.embedding for d in resp.data]
+    except Exception:
+        return None
+
+
 def _job_text(job: dict) -> str:
     # The embedding is the best semantic matcher — feed it the FULL signal, not a 600-char stub.
     # role_category + industry anchor non-tech JDs (narrative prose, few repeated title tokens).
@@ -51,18 +82,15 @@ def _cos(a, b) -> float:
 
 def vibe_scores(prefs: dict, ctx: dict | None, jobs: list[dict]) -> dict[str, float] | None:
     """id → cosine sim (0..1). None if embeddings can't run (caller falls back to white-box only)."""
-    client = embed_client()
-    if not client or not jobs:
+    if not jobs:
         return None
     ptext = _profile_text(prefs, ctx)
     if not ptext.strip():
         return None
     s = get_settings()
     texts = [ptext] + [_job_text(j) for j in jobs]
-    try:
-        resp = client.embeddings.create(model=s.EMBED_MODEL, input=texts)
-    except Exception:
+    vecs = _embedding_vectors(s, texts)
+    if not vecs or len(vecs) != len(texts):
         return None
-    vecs = [d.embedding for d in resp.data]
     pvec, jvecs = vecs[0], vecs[1:]
     return {job["id"]: max(0.0, min(1.0, _cos(pvec, jv))) for job, jv in zip(jobs, jvecs)}

app/engine/llm.py

@@ -1,8 +1,8 @@
 """LLM clients — ONE place to build them.
 
 Two backends, by capability:
-- **Gateway** (opencode.ai/zen, OpenAI-compatible) → chat for the Opus curator. Has no embeddings route.
-- **Direct OpenAI** (poc key) → embeddings (`text-embedding-3-small`).
+- **Gateway** → chat for the curator, using DSPY_* when configured or direct OpenAI otherwise.
+- **Embeddings** → direct OpenAI or an OpenAI-compatible embeddings API via OPENAI_API_BASE.
 
 Both keys come from config/.env (git-ignored). Each builder returns `None` when unconfigured so the
 engine degrades gracefully (white-box sift + templated cards — the safety net).
@@ -16,25 +16,32 @@ from app.config import get_settings
 
 @lru_cache
 def gateway_client():
-    """OpenAI-compatible client for the opencode.ai/zen gateway — chat (Opus). None if unconfigured."""
+    """Chat client for curation/suggestions. Prefer DSPY_* override, otherwise use OPENAI_API_KEY."""
     s = get_settings()
-    if not (s.DSPY_API_BASE and s.DSPY_API_KEY):
+    api_key = s.DSPY_API_KEY or s.OPENAI_API_KEY
+    if not api_key:
         return None
     from openai import OpenAI
-    return OpenAI(base_url=s.DSPY_API_BASE, api_key=s.DSPY_API_KEY, timeout=60, max_retries=2)
+    kwargs = {"api_key": api_key, "timeout": 60, "max_retries": 2}
+    if s.DSPY_API_BASE:
+        kwargs["base_url"] = s.DSPY_API_BASE
+    return OpenAI(**kwargs)
 
 
 @lru_cache
 def embed_client():
-    """Direct OpenAI client for embeddings (the gateway has no embeddings route). None if no key."""
+    """OpenAI-compatible client for embeddings. None if no key."""
     s = get_settings()
     if not s.OPENAI_API_KEY:
         return None
     from openai import OpenAI
-    return OpenAI(api_key=s.OPENAI_API_KEY, timeout=30, max_retries=2)
+    kwargs = {"api_key": s.OPENAI_API_KEY, "timeout": 30, "max_retries": 2}
+    if s.OPENAI_API_BASE:
+        kwargs["base_url"] = s.OPENAI_API_BASE
+    return OpenAI(**kwargs)
 
 
 def curate_enabled() -> bool:
-    """Is the Opus curator usable? (flag on + gateway key present)."""
+    """Is the curator usable? (flag on + any chat key present)."""
     s = get_settings()
-    return bool(s.ENGINE_LLM_ENABLED and s.DSPY_API_BASE and s.DSPY_API_KEY)
+    return bool(s.ENGINE_LLM_ENABLED and (s.DSPY_API_KEY or s.OPENAI_API_KEY))

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
 

app/grow_events.py

@@ -0,0 +1,73 @@
+"""Grow event emitter — best-effort POST to the backend /events/ingest/service endpoint.
+
+Every call degrades to a no-op when GROW_EVENTS_URL is unset or the POST fails, so
+a Grow outage can never break the search/apply flows. The emitter runs in a
+background thread (via asyncio) so it never blocks the agent session loop.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+from datetime import datetime, timezone
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+SERVICE_NAME = "matchmaking-service"
+
+
+def _post(event: dict) -> None:
+    """Synchronous POST — run off the event loop via ``asyncio.to_thread``."""
+    grow_url = os.getenv("GROW_EVENTS_URL", "").rstrip("/")
+    if not grow_url:
+        return
+    url = grow_url + "/events/ingest/service"
+    token = os.getenv("GROW_EVENTS_TOKEN", "") or os.getenv("A2A_SERVICE_KEY", "")
+    headers = {"Content-Type": "application/json", "x-growqr-source": SERVICE_NAME}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    try:
+        resp = httpx.post(url, json=event, headers=headers, timeout=5.0)
+        resp.raise_for_status()
+    except Exception as e:  # noqa: BLE001 — never fatal
+        logger.warning("emit_grow_event failed (%s): %s", event.get("type"), e)
+
+
+async def emit_grow_event(
+    event_type: str,
+    user_id: str,
+    *,
+    org_id: str | None = None,
+    payload: dict | None = None,
+    correlation: dict | None = None,
+    subject: dict | None = None,
+    dedupe_key: str | None = None,
+) -> None:
+    """Emit a Grow event to the backend ingest endpoint.
+
+    Non-blocking and non-fatal: if GROW_EVENTS_URL is unset the call is a no-op,
+    and any HTTP error is logged and swallowed. The POST runs in a worker thread
+    so the agent session loop is never blocked.
+    """
+    if not user_id:
+        return
+
+    event: dict = {
+        "source": SERVICE_NAME,
+        "type": event_type,
+        "category": "service",
+        "userId": user_id,
+        "orgId": org_id or "growqr",
+        "occurredAt": datetime.now(timezone.utc).isoformat(),
+        "payload": payload or {},
+    }
+    if correlation:
+        event["correlation"] = correlation
+    if subject:
+        event["subject"] = subject
+    if dedupe_key:
+        event["dedupeKey"] = dedupe_key
+
+    await asyncio.to_thread(_post, event)

docker-compose.staging.yml

@@ -0,0 +1,74 @@
+# ── Matchmaking v2 staging stack ────────────────────────────────────────────
+# Project name is `matchmaking-v2` (matches scripts/sync-staging.sh compose_project
+# and the live volume/network names). All three services deploy together so they
+# share one Compose default bridge and the depends_on graph holds:
+#
+#   docker compose -p matchmaking-v2 -f docker-compose.staging.yml up -d
+#
+# The prior VPS split (api under `-p matchmaking-v2-staging`, postgres/redis under
+# `-p matchmaking-v2`) created two disjoint `<project>_default` bridges → every
+# dependency hostname was NXDOMAIN from the api. Standardizing on `matchmaking-v2`
+# for all services fixes that at the root and matches the deploy automation.
+#
+# The Postgres data volume stays project-scoped (matchmaking-v2_matchmaking_pgdata) —
+# the existing live volume with its PG cluster is reused as-is, no migration or rename.
+#
+# External network — pre-created by the deploy guard (sync-staging.sh compose_networks)
+# or manually (guard is a no-op if it already exists):
+#   docker network create matchmaking-v2_default 2>/dev/null || true
+services:
+  api:
+    build: .
+    container_name: growqr-matchmaking-api
+    ports:
+      - "127.0.0.1:18006:8000"
+    env_file:
+      - .env.staging
+    # Pin DB/Redis hosts to the compose service names so connectivity doesn't
+    # hinge on .env.staging (git-ignored). Credentials mirror the postgres
+    # service below — no new secret introduced.
+    environment:
+      DATABASE_URL: postgresql+asyncpg://matchmaking:matchmaking@postgres:5432/matchmaking
+      REDIS_URL: redis://redis:6379/0
+    depends_on:
+      postgres:
+        condition: service_healthy
+      redis:
+        condition: service_started
+    command: sh -c "alembic upgrade head && exec uvicorn app.main:app --host 0.0.0.0 --port 8000"
+    restart: unless-stopped
+
+  postgres:
+    image: postgres:16-alpine
+    container_name: growqr-matchmaking-postgres
+    environment:
+      POSTGRES_USER: matchmaking
+      POSTGRES_PASSWORD: matchmaking
+      POSTGRES_DB: matchmaking
+    ports:
+      - "127.0.0.1:15446:5432"
+    volumes:
+      - matchmaking_pgdata:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U matchmaking -d matchmaking"]
+      interval: 5s
+      timeout: 3s
+      retries: 20
+    restart: unless-stopped
+
+  redis:
+    image: redis:7-alpine
+    container_name: growqr-matchmaking-redis
+    restart: unless-stopped
+
+# External bridge — pre-created and shared so a cold redeploy works even if the
+# network was last created under a different project name.
+networks:
+  default:
+    name: matchmaking-v2_default
+    external: true
+
+# Project-scoped volume — the live `matchmaking-v2_matchmaking_pgdata` (47.7M PG
+# cluster) is reused as-is. No rename, no migration, no data risk.
+volumes:
+  matchmaking_pgdata:

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_grow_events.py

@@ -0,0 +1,355 @@
+"""Grow event emitter + handler contract tests (red→green).
+
+Covers:
+  1. The POST URL normalizes to ``/events/ingest/service`` (not ``/events/ingest``,
+     which was the wrong ingest route — a wrong Clerk route cannot recur).
+  2. The ``x-growqr-source`` header identifies the emitting service.
+  3. ``raise_for_status`` is invoked: a 401 response surfaces as a logged warning
+     (401 is observable) rather than being silently swallowed.
+  4. Emit-after-mutation ordering: handlers fire the Grow event *after* the
+     persistent mutation, never before / never without it.
+  5. Action-specific dedupe keys: stable for the same evidence (retries collapse),
+     distinct for different evidence (a different viewed job / a different search
+     sig / a later feed snapshot is NOT dropped).
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import httpx
+import pytest
+
+from app.grow_events import _post, emit_grow_event, SERVICE_NAME
+from app.agent.session import MatchmakingAgentSession
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+
+def _capture_post(monkeypatch):
+    """Patch httpx.post to capture the request kwargs and return a fake Response."""
+    captured: dict = {}
+    fake_resp = MagicMock(spec=httpx.Response)
+    fake_resp.status_code = 202
+    fake_resp.raise_for_status = MagicMock()
+    def _fake_post(url, **kw):
+        captured["url"] = url
+        captured["headers"] = kw.get("headers", {})
+        captured["json"] = kw.get("json", {})
+        return fake_resp
+    monkeypatch.setattr(httpx, "post", _fake_post)
+    return captured
+
+
+def _capture_post_401(monkeypatch):
+    """Patch httpx.post to return a 401 Response whose raise_for_status raises."""
+    captured: dict = {}
+    fake_resp = MagicMock(spec=httpx.Response)
+    fake_resp.status_code = 401
+    fake_resp.raise_for_status.side_effect = httpx.HTTPStatusError(
+        "401 Unauthorized", request=httpx.Request("POST", "http://x"), response=fake_resp,
+    )
+    def _fake_post(url, **kw):
+        captured["url"] = url
+        captured["headers"] = kw.get("headers", {})
+        captured["json"] = kw.get("json", {})
+        return fake_resp
+    monkeypatch.setattr(httpx, "post", _fake_post)
+    return captured
+
+
+async def _run(coro):
+    return await coro
+
+
+# ── 1. URL normalize → /events/ingest/service ────────────────────────────────
+
+def test_post_url_is_events_ingest_service(monkeypatch):
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    _post({"type": "X", "userId": "u1"})
+    assert captured["url"].rstrip("/").endswith("/events/ingest/service"), (
+        f"URL must be /events/ingest/service, got {captured['url']!r}"
+    )
+
+
+def test_post_url_trailing_slash_normalized(monkeypatch):
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com/")
+    _post({"type": "X", "userId": "u1"})
+    assert captured["url"].rstrip("/").endswith("/events/ingest/service")
+
+
+# ── 2. x-growqr-source header ────────────────────────────────────────────────
+
+def test_post_has_growqr_source_header(monkeypatch):
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    _post({"type": "X", "userId": "u1"})
+    assert captured["headers"].get("x-growqr-source") == SERVICE_NAME, (
+        f"x-growqr-source header must be {SERVICE_NAME!r}, got {captured['headers'].get('x-growqr-source')!r}"
+    )
+
+
+# ── 3. raise_for_status observability ────────────────────────────────────────
+
+def test_post_raises_for_status_on_401_and_logs(monkeypatch, caplog):
+    captured = _capture_post_401(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    with caplog.at_level("WARNING", logger="app.grow_events"):
+        _post({"type": "X", "userId": "u1"})
+    assert any("401" in r.message or "Unauthorized" in r.message for r in caplog.records if r.levelname == "WARNING"), (
+        f"a 401 must surface as a logged warning: {[r.message for r in caplog.records]!r}"
+    )
+
+
+def test_post_401_does_not_raise_out_of_post(monkeypatch):
+    """The 401 is caught + logged; it never escapes _post (non-fatal contract)."""
+    _capture_post_401(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    # Must not raise.
+    _post({"type": "X", "userId": "u1"})
+
+
+# ── 4. emit-after-mutation ordering ───────────────────────────────────────────
+
+class _StubSession:
+    """Minimal stand-in matching MatchmakingAgentSession interface for ordering tests."""
+    pass
+
+
+def _make_session(user_id="u1"):
+    ws = MagicMock()
+    ws.send_json = AsyncMock()
+    # Build a real session so we exercise the actual handler code path.
+    from app.agent.session import MatchmakingAgentSession
+    s = MatchmakingAgentSession(ws, user_id)
+    return s, ws
+
+
+def test_emit_mark_applied_after_set_opportunity_state(monkeypatch):
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+
+    called = []
+
+    async def fake_set_state(uid, oid, **kw):
+        called.append("mutate")
+        return None
+
+    emitted = []
+    async def fake_event(etype, uid, **kw):
+        emitted.append(("emit", etype))
+        return None
+
+    s, ws = _make_session("u1")
+    monkeypatch.setattr("app.agent.session.set_opportunity_state", fake_set_state, raising=False)
+    # Need the actual repo patch path the handler uses (from app.db.repo import set_opportunity_state)
+    monkeypatch.setattr("app.db.repo.set_opportunity_state", fake_set_state)
+    monkeypatch.setattr("app.agent.session.emit_grow_event", fake_event)
+
+    asyncio.run(s.handle_mark_applied({"opportunity_id": "job-A"}))
+    assert called == ["mutate"], "set_opportunity_state must run before emit"
+    assert len(emitted) == 1 and emitted[0][1] == "matchmaking.match.applied"
+    assert called[0] == "mutate"  # mutation precedes emit
+
+
+def test_emit_no_event_without_mutation(monkeypatch):
+    """If the mutation does not run, the handler must NOT emit a Grow event."""
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    async def fake_set_state_raise(uid, oid, **kw):
+        raise RuntimeError("db down")
+    emitted = []
+    async def fake_event(*a, **kw):
+        emitted.append("emit")
+        return None
+    s, ws = _make_session("u1")
+    monkeypatch.setattr("app.db.repo.set_opportunity_state", fake_set_state_raise)
+    monkeypatch.setattr("app.agent.session.emit_grow_event", fake_event)
+    # handler logs+swallows via the on_user_action catch; call directly → raises
+    try:
+        asyncio.run(s.handle_mark_applied({"opportunity_id": "job-A"}))
+    except RuntimeError:
+        pass
+    assert emitted == [], "no event may be emitted when the mutation fails"
+
+
+# ── 5. dedupe keys: stable for same evidence, distinct for different ──────────
+
+def test_dedupe_key_mark_applied_stable_and_distinct(monkeypatch):
+    """retry same (user,opp) → same key; different opp → different key (not dropped)."""
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+
+    async def fake_set_state(uid, oid, **kw):
+        return None
+    monkeypatch.setattr("app.db.repo.set_opportunity_state", fake_set_state)
+
+    s, ws = _make_session("u1")
+    asyncio.run(s.handle_mark_applied({"opportunity_id": "job-A"}))
+    asyncio.run(s.handle_mark_applied({"opportunity_id": "job-A"}))  # retry
+    asyncio.run(s.handle_mark_applied({"opportunity_id": "job-B"}))  # distinct later action
+
+    events = captured["json"].__class__ if False else None  # placeholder
+    # We patch emit_grow_event to capture the dedupe_key instead
+    keys = []
+    async def capture_event(etype, uid, *, dedupe_key=None, **kw):
+        keys.append((etype, dedupe_key))
+        return None
+    monkeypatch.setattr("app.agent.session.emit_grow_event", capture_event)
+    s2, _ = _make_session("u1")
+    asyncio.run(s2.handle_mark_applied({"opportunity_id": "job-A"}))
+    asyncio.run(s2.handle_mark_applied({"opportunity_id": "job-A"}))
+    asyncio.run(s2.handle_mark_applied({"opportunity_id": "job-B"}))
+    assert keys[0] == keys[1], " retries of the same (user,opp) must collapse (same dedupe_key)"
+    assert keys[2][1] != keys[0][1], "a different opp must get a distinct dedupe_key so it is NOT dropped"
+    assert keys[0][1] is not None and "job-A" in keys[0][1]
+    assert keys[2][1] is not None and "job-B" in keys[2][1]
+
+
+def test_dedupe_key_mark_viewed_stable_and_distinct(monkeypatch):
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    async def fake_set_state(uid, oid, **kw):
+        return None
+    monkeypatch.setattr("app.db.repo.set_opportunity_state", fake_set_state)
+
+    keys = []
+    async def capture_event(etype, uid, *, dedupe_key=None, **kw):
+        keys.append((etype, dedupe_key))
+        return None
+    monkeypatch.setattr("app.agent.session.emit_grow_event", capture_event)
+
+    s, _ = _make_session("u1")
+    asyncio.run(s.handle_mark_viewed({"opportunity_id": "job-A"}))
+    asyncio.run(s.handle_mark_viewed({"opportunity_id": "job-A"}))  # retry same
+    asyncio.run(s.handle_mark_viewed({"opportunity_id": "job-B"}))  # distinct
+    assert keys[0][1] is not None, "mark_viewed must have a dedupe_key"
+    assert keys[0][1] == keys[1][1], "retry of same viewed opp collapses"
+    assert keys[2][1] != keys[0][1], "different viewed opp must be distinct (not dropped)"
+    assert "job-A" in keys[0][1] and "job-B" in keys[2][1]
+
+
+def test_dedupe_key_matches_generated_stable_and_distinct(monkeypatch):
+    """generated with sig1 vs sig2 -> different keys; retry same sig -> same key."""
+    # We test the dedupe_key construction by calling the handler path with mocked
+    # engine/repo so only the emit-grow-event key shape is exercised.
+    import app.engine.sift, app.engine.curate, app.engine.rank, app.engine.search  # noqa: F401 -- pre-import for monkeypatch
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+
+    # Mock the repo bits the handler imports inline.
+    async def fake_save_feed(uid, prefs, result, **kw):
+        return None
+    async def fake_update_salary(uid, sals):
+        return None
+    async def fake_mark_seen(uid, ids):
+        return None
+    async def fake_get_states(uid):
+        return {}
+    async def fake_activity_fields_hook(self):
+        return {}
+    def fake_annotate(opps, states):
+        return opps
+    monkeypatch.setattr("app.db.repo.save_feed", fake_save_feed)
+    monkeypatch.setattr("app.db.repo.update_salary_band", fake_update_salary)
+    monkeypatch.setattr("app.db.repo.mark_seen", fake_mark_seen)
+    monkeypatch.setattr("app.db.repo.get_opportunity_states", fake_get_states)
+    monkeypatch.setattr("app.db.repo.annotate_and_filter", fake_annotate)
+    # pool helpers
+    async def fake_pool_load(*a, **kw):
+        return []
+    async def fake_is_pool_exhausted(*a, **kw):
+        return False
+    def fake_search_sig(prefs):
+        import hashlib
+        return hashlib.md5(str(prefs).encode()).hexdigest()
+    monkeypatch.setattr("app.db.repo.pool_load_fresh_unseen", fake_pool_load)
+    monkeypatch.setattr("app.db.repo.is_pool_exhausted", fake_is_pool_exhausted)
+    monkeypatch.setattr("app.db.repo.search_signature", fake_search_sig)
+    monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=[]))
+
+    keys = []
+    async def capture_event(etype, uid, *, dedupe_key=None, **kw):
+        keys.append((etype, dedupe_key))
+        return None
+    monkeypatch.setattr("app.agent.session.emit_grow_event", capture_event)
+
+    # We directly call the event-emit portion logic via the full handler with
+    # mocked engine so the expensive LLM path is skipped. To keep the test
+    # deterministic we mock the engine modules the handler imports inline.
+    class FakeSift:
+        @staticmethod
+        def sift(prefs, ctx, cands):
+            return [], {"mode": "test", "scored": 0}
+    class FakeCurate:
+        @staticmethod
+        def curate(prefs, ctx, top):
+            return [{"id": "j1", "match": {"score": 90}}]
+    class FakeRank:
+        @staticmethod
+        def select(top, stretch):
+            return top
+    monkeypatch.setattr("app.engine.sift", FakeSift)
+    monkeypatch.setattr("app.engine.curate", FakeCurate)
+    monkeypatch.setattr("app.engine.rank", FakeRank)
+    monkeypatch.setattr("app.engine.search", MagicMock(has_mvq=lambda p: True, BOARDS=["naukri"]))
+    # patch _grow_activity_fields to avoid DB
+    monkeypatch.setattr(MatchmakingAgentSession, "_grow_activity_fields", fake_activity_fields_hook)
+
+    s, _ = _make_session("u1")
+
+    # search A
+    asyncio.run(s.handle_run_search({"title": "PM", "location": "BLR"}))
+    # retry (same prefs -> same sig) should collapse
+    asyncio.run(s.handle_run_search({"title": "PM", "location": "BLR"}))
+    # search B: different prefs -> different sig
+    asyncio.run(s.handle_run_search({"title": "SWE", "location": "BLR"}))
+
+    assert all(k[1] is not None for k in keys), "matches.generated must carry a dedupe_key"
+    assert keys[0][1] == keys[1][1], "retry same sig must collapse"
+    assert keys[2][1] != keys[0][1], "different sig must be distinct so a later search is not dropped"
+
+
+def test_dedupe_key_feed_viewed_stable_and_distinct(monkeypatch):
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+
+    async def fake_get_feed(uid):
+        return {"opportunities": [{"id": "j1"}], "sources": {}, "updated_at": "2026-07-10T00:00:00"}
+    async def fake_activity(self):
+        return {}
+    monkeypatch.setattr("app.db.repo.get_feed", fake_get_feed)
+    monkeypatch.setattr(MatchmakingAgentSession, "_grow_activity_fields", fake_activity)
+
+    keys = []
+    async def capture_event(etype, uid, *, dedupe_key=None, **kw):
+        keys.append((etype, dedupe_key))
+        return None
+    monkeypatch.setattr("app.agent.session.emit_grow_event", capture_event)
+
+    s, _ = _make_session("u1")
+    asyncio.run(s.handle_get_feed({}))
+    asyncio.run(s.handle_get_feed({}))  # same updated_at → same key
+
+    # Now simulate a re-search (feed updated_at changes)
+    async def fake_get_feed2(uid):
+        return {"opportunities": [{"id": "j2"}], "sources": {}, "updated_at": "2026-07-11T00:00:00"}
+    monkeypatch.setattr("app.db.repo.get_feed", fake_get_feed2)
+    asyncio.run(s.handle_get_feed({}))  # new updated_at → distinct key
+
+    assert keys[0][1] is not None, "feed.viewed must carry a dedupe_key"
+    assert keys[0][1] == keys[1][1], "same feed snapshot retries must collapse"
+    assert keys[2][1] != keys[0][1], "a later feed snapshot must be distinct so it is NOT dropped"
+
+
+def test_dedupe_key_idempotent_retries_match_applied(monkeypatch):
+    """Direct emitter-level check: the same dedupe_key produces the same posted dedupeKey."""
+    captured = _capture_post(monkeypatch)
+    monkeypatch.setenv("GROW_EVENTS_URL", "https://backend.example.com")
+    asyncio.run(emit_grow_event("matchmaking.match.applied", "u1", dedupe_key="matchmaking:applied:u1:job-A"))
+    k1 = captured["json"].get("dedupeKey")
+    captured2 = _capture_post(monkeypatch)
+    asyncio.run(emit_grow_event("matchmaking.match.applied", "u1", dedupe_key="matchmaking:applied:u1:job-A"))
+    k2 = captured2["json"].get("dedupeKey")
+    assert k1 == k2 == "matchmaking:applied:u1:job-A"

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_staging_infra.py

@@ -0,0 +1,224 @@
+"""Staging infra regression — the durable fixes this slice landed:
+
+1. Single-project topology: docker-compose.staging.yml deploys all three services
+   under ONE project name (`matchmaking-v2` — matches scripts/sync-staging.sh
+   compose_project and the live volume/network names) with fixed container names and
+   an external shared bridge (matchmaking-v2_default). The prior split (api under
+   `-p matchmaking-v2-staging`, postgres/redis under `-p matchmaking-v2`) created two
+   disjoint `<project>_default` bridges → every dependency hostname was NXDOMAIN from
+   the api → feed store init failed with [Errno -2].
+
+2. Project-scoped volume: the Postgres data volume stays project-scoped
+   (matchmaking-v2_matchmaking_pgdata) — the existing live volume with its PG cluster
+   is reused as-is, no migration or rename.
+
+3. Apify token redaction: the client authenticates via an Authorization header (not a
+   query param), so no request URL — and therefore no httpx.HTTPStatusError string —
+   can ever contain the token. A 403 surfaces as a token-free RuntimeError.
+
+These tests don't need a live stack — they lock the CONFIG shape and the ERROR SHAPE so
+the regressions they guard (network split, token-in-logs) can't return.
+"""
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+import httpx
+import pytest
+import yaml
+
+from app.engine import apify_client
+
+
+_COMPOSE = Path(__file__).resolve().parents[1] / "docker-compose.staging.yml"
+
+
+def _load_compose() -> dict:
+    return yaml.safe_load(_COMPOSE.read_text())
+
+
+# ── 1. Single-project topology + external shared network ─────────────────────
+
+def test_staging_compose_uses_external_shared_network():
+    """The default network must be external + named matchmaking-v2_default so a cold
+    redeploy works even if the bridge was last created under a different project."""
+    d = _load_compose()
+    net = d["networks"]["default"]
+    assert net.get("external") is True, "network must be external (no project owns/removes it)"
+    assert net.get("name") == "matchmaking-v2_default", (
+        "name must be the literal matchmaking-v2_default — the bridge postgres/redis already use"
+    )
+
+
+def test_staging_api_depends_on_postgres_and_redis():
+    """The api service must depend on BOTH postgres and redis — this is the depends_on
+    graph that guarantees they land on the same Compose network when deployed as one
+    project. Breaking this (e.g. --no-deps deploy) reintroduces the network split."""
+    d = _load_compose()
+    deps = d["services"]["api"].get("depends_on", {})
+    assert "postgres" in deps, "api must depend_on postgres (same-network guarantee)"
+    assert "redis" in deps, "api must depend_on redis (same-network guarantee)"
+    assert deps["postgres"].get("condition") == "service_healthy", "postgres dep must wait for healthcheck"
+    assert deps["redis"].get("condition") == "service_started", "redis dep must wait for start"
+
+
+def test_staging_api_pins_db_and_redis_hosts():
+    """Inline environment overrides guarantee the api reaches postgres/redis by service
+    name even if the git-ignored .env.staging drifts to localhost."""
+    d = _load_compose()
+    env = d["services"]["api"]["environment"]
+    assert env["DATABASE_URL"].endswith("@postgres:5432/matchmaking"), "DB host must be the compose service name"
+    assert env["REDIS_URL"].startswith("redis://redis:"), "Redis host must be the compose service name"
+
+
+def test_staging_service_names_are_unique():
+    """Unique container names prevent cross-project DNS ambiguity on the shared bridge."""
+    d = _load_compose()
+    names = {s["container_name"] for s in d["services"].values()}
+    assert names == {"growqr-matchmaking-api", "growqr-matchmaking-postgres", "growqr-matchmaking-redis"}
+
+
+# ── 2. Project-scoped volume ─────────────────────────────────────────────────
+
+def test_staging_postgres_mounts_volume():
+    """The postgres service must mount the data volume at the PG data dir."""
+    d = _load_compose()
+    mounts = d["services"]["postgres"]["volumes"]
+    assert any("matchmaking_pgdata:/var/lib/postgresql/data" in m for m in mounts), (
+        "postgres must mount matchmaking_pgdata at the PG data directory"
+    )
+
+
+def test_staging_volume_is_project_scoped():
+    """The volume must be project-scoped (NOT external) so the live
+    matchmaking-v2_matchmaking_pgdata is reused as-is — no rename, no data risk."""
+    d = _load_compose()
+    vol = d["volumes"]["matchmaking_pgdata"]
+    assert vol is None or (isinstance(vol, dict) and not vol.get("external")), (
+        "volume must NOT be external — the live project-scoped volume is reused as-is"
+    )
+
+
+# ── 2b. Deploy script alignment ─────────────────────────────────────────────
+
+_SYNC_STAGING = Path(__file__).resolve().parents[2] / "scripts" / "sync-staging.sh"
+
+
+def test_sync_staging_compose_services_includes_all_three():
+    """The deploy script must bring up all three services (api, postgres, redis) under
+    the single project — not just api. Without this, postgres/redis stay on whatever
+    project they were last started under, and the network split can recur."""
+    if not _SYNC_STAGING.exists():
+        pytest.skip("sync-staging.sh not in this checkout (run from growqr monorepo)")
+    text = _SYNC_STAGING.read_text()
+    # compose_services must have an explicit matchmaking-v2 case listing all three
+    assert "matchmaking-v2)     printf 'api postgres redis'" in text, (
+        "sync-staging.sh compose_services must list 'api postgres redis' for matchmaking-v2"
+    )
+
+
+def test_sync_staging_compose_networks_includes_shared_bridge():
+    """The deploy guard must pre-create the external matchmaking-v2_default network
+    so `external: true` in the compose never fails on ordering."""
+    if not _SYNC_STAGING.exists():
+        pytest.skip("sync-staging.sh not in this checkout (run from growqr monorepo)")
+    text = _SYNC_STAGING.read_text()
+    assert "matchmaking-v2_default" in text, (
+        "sync-staging.sh compose_networks must pre-create matchmaking-v2_default"
+    )
+
+
+def test_sync_staging_public_health_url_has_matchmaking_case():
+    """public_health_url must have a matchmaking-v2 case — without it, the function
+    returns 1 (the `*)` fallthrough), and under `set -e` the command substitution at
+    the call site exits the script before the deploy runs."""
+    if not _SYNC_STAGING.exists():
+        pytest.skip("sync-staging.sh not in this checkout (run from growqr monorepo)")
+    text = _SYNC_STAGING.read_text()
+    assert "matchmaking-v2)     printf 'https://matchmaking-staging.gqr.puter.wtf/api/v1/health'" in text, (
+        "sync-staging.sh public_health_url must have a matchmaking-v2 case to avoid set -e abort"
+    )
+
+
+# ── 3. Apify token redaction ─────────────────────────────────────────────────
+
+_TOKEN = "apify_api_secret_TOKEN_value_DO_NOT_LEAK"
+
+
+def test_auth_header_not_query_param():
+    """Token must travel in a header, never in the URL."""
+    h = apify_client._auth_headers(_TOKEN)
+    assert h == {"Authorization": f"Bearer {_TOKEN}"}
+
+
+def test_run_actor_4xx_raises_token_free_error(monkeypatch):
+    """A provider 403 (the live failure mode) must surface WITHOUT the token in the message.
+
+    The live Apify account returns 403 platform-feature-disabled (monthly cap exceeded).
+    That error must be explicit and safe — no credential in the exception or any log line.
+    """
+    calls = {"post": 0}
+
+    class _FakeResponse:
+        status_code = 403
+        request = httpx.Request("POST", "https://api.apify.com/v2/acts/x/run-sync-get-dataset-items")
+
+        def raise_for_status(self):
+            raise httpx.HTTPStatusError("403 Forbidden", request=self.request, response=self)
+
+        def json(self):
+            return {}
+
+    async def _fake_post(self, url, **kwargs):
+        calls["post"] += 1
+        # CRITICAL: assert no token leaked into the URL or query params
+        assert "token" not in str(url), "token must not be in the request URL"
+        assert "token" not in str(kwargs.get("params", {})), "token must not be in query params"
+        return _FakeResponse()
+
+    _FakeSettings = type("S", (), {
+        "APIFY_TOKEN": _TOKEN, "APIFY_CACHE_MODE": "off", "APIFY_CACHE_DIR": ".apify_cache",
+    })
+    monkeypatch.setattr(apify_client, "get_settings", lambda: _FakeSettings())
+    monkeypatch.setattr(httpx.AsyncClient, "post", _fake_post)
+
+    with pytest.raises(RuntimeError) as exc_info:
+        asyncio.run(apify_client.run_actor("test~actor", {"q": "dev"}))
+
+    assert calls["post"] >= 1, "the fake POST must be reached (else the test passed for the wrong reason)"
+    msg = str(exc_info.value)
+    assert _TOKEN not in msg, f"token leaked into error message: {msg!r}"
+    assert "403" in msg, f"the HTTP status must be in the message: {msg!r}"
+
+
+def test_run_actor_async_4xx_raises_token_free_error(monkeypatch):
+    """The async-run path (incremental actors) must likewise not leak the token."""
+    calls = {"post": 0}
+
+    class _FakeResponse:
+        status_code = 403
+        request = httpx.Request("POST", "https://api.apify.com/v2/acts/x/runs")
+
+        def raise_for_status(self):
+            raise httpx.HTTPStatusError("403 Forbidden", request=self.request, response=self)
+
+        def json(self):
+            return {}
+
+    async def _fake_post(self, url, **kwargs):
+        calls["post"] += 1
+        assert "token" not in str(url), "token must not be in the request URL"
+        return _FakeResponse()
+
+    _FakeSettings = type("S", (), {
+        "APIFY_TOKEN": _TOKEN, "APIFY_CACHE_MODE": "off", "APIFY_CACHE_DIR": ".apify_cache",
+    })
+    monkeypatch.setattr(apify_client, "get_settings", lambda: _FakeSettings())
+    monkeypatch.setattr(httpx.AsyncClient, "post", _fake_post)
+
+    with pytest.raises(RuntimeError) as exc_info:
+        asyncio.run(apify_client.run_actor("test~actor", {"q": "dev", "incremental": True}))
+
+    assert calls["post"] >= 1, "the fake POST must be reached (else the test passed for the wrong reason)"
+    assert _TOKEN not in str(exc_info.value), "token leaked into async error message"

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]