surface total job board outages

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

Dockerfile

@@ -8,6 +8,9 @@ COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 
 COPY app ./app
+COPY alembic.ini .
+COPY alembic/ ./alembic/
 
 EXPOSE 8000
+# migrations run from the compose command (alembic upgrade head && uvicorn …) — the house pattern
 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

alembic.ini

@@ -0,0 +1,39 @@
+[alembic]
+script_location = alembic
+prepend_sys_path = .
+
+# Default (overridden in alembic/env.py from DATABASE_URL)
+sqlalchemy.url = postgresql+asyncpg://matchmaking:matchmaking@localhost:5446/matchmaking
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

alembic/env.py

@@ -0,0 +1,67 @@
+"""Alembic async env — matches the sibling-service house pattern (user-service et al.).
+
+Reads the live DB URL from app settings (overriding the alembic.ini placeholder), targets
+Base.metadata (all models live in app.db.models), and runs migrations on the async engine.
+"""
+import asyncio
+import sys
+from logging.config import fileConfig
+from pathlib import Path
+
+# Ensure the project root is importable when alembic runs via the console script.
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from alembic import context
+from sqlalchemy import pool
+from sqlalchemy.ext.asyncio import async_engine_from_config
+
+from app.config import get_settings
+from app.db.models import Base  # importing Base registers every model on Base.metadata
+
+settings = get_settings()
+config = context.config
+
+# Live URL from settings (DATABASE_URL) overrides the ini placeholder; fall back to the ini value.
+config.set_main_option("sqlalchemy.url", settings.DATABASE_URL or config.get_main_option("sqlalchemy.url"))
+
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+target_metadata = Base.metadata
+
+
+def run_migrations_offline() -> None:
+    url = config.get_main_option("sqlalchemy.url")
+    context.configure(url=url, target_metadata=target_metadata, literal_binds=True,
+                      dialect_opts={"paramstyle": "named"})
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def do_run_migrations(connection):
+    context.configure(connection=connection, target_metadata=target_metadata)
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+async def run_async_migrations() -> None:
+    connectable = async_engine_from_config(
+        config.get_section(config.config_ini_section, {}),
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+    async with connectable.connect() as connection:
+        await connection.run_sync(do_run_migrations)
+    await connectable.dispose()
+
+
+def run_migrations_online() -> None:
+    asyncio.run(run_async_migrations())
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

alembic/script.py.mako

@@ -0,0 +1,24 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+revision: str = ${repr(up_revision)}
+down_revision: Union[str, None] = ${repr(down_revision)}
+branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
+depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

alembic/versions/0001_baseline.py

@@ -0,0 +1,51 @@
+"""baseline — the schema as it exists in prod today (from the cutover's create_all):
+user_feed (WITHOUT exhausted_at) + opportunity_state. On prod this is STAMPED (the tables already
+exist); on a fresh DB `upgrade` creates them. The exhausted_at column + user_job_pool are added in 0002.
+
+Revision ID: 0001_baseline
+Revises:
+Create Date: 2026-06-26
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0001_baseline"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "user_feed",
+        sa.Column("user_id", sa.String(), primary_key=True),
+        sa.Column("prefs", sa.JSON(), nullable=True),
+        sa.Column("opportunities", sa.JSON(), nullable=True),
+        sa.Column("sources", sa.JSON(), nullable=True),
+        sa.Column("engine", sa.String(), nullable=True),
+        sa.Column("scanned", sa.Integer(), nullable=True),
+        sa.Column("query_sig", sa.String(), nullable=True),
+        sa.Column("cursors", sa.JSON(), nullable=True),
+        sa.Column("search_count", sa.Integer(), server_default="0", nullable=True),
+        sa.Column("salary_peak_sum", sa.Float(), server_default="0", nullable=True),
+        sa.Column("salary_peak_count", sa.Integer(), server_default="0", nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True),
+    )
+    op.create_table(
+        "opportunity_state",
+        sa.Column("user_id", sa.String(), primary_key=True),
+        sa.Column("opportunity_id", sa.String(), primary_key=True),
+        sa.Column("status", sa.String(), nullable=True),
+        sa.Column("tailored_resume_id", sa.String(), nullable=True),
+        sa.Column("tailored_version_id", sa.String(), nullable=True),
+        sa.Column("cover_letter_id", sa.String(), nullable=True),
+        sa.Column("seen", sa.Boolean(), server_default="false", nullable=True),
+        sa.Column("viewed", sa.Boolean(), server_default="false", nullable=True),
+        sa.Column("saved", sa.Boolean(), server_default="false", nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("opportunity_state")
+    op.drop_table("user_feed")

alembic/versions/0002_pool_and_exhausted.py

@@ -0,0 +1,33 @@
+"""warm pool — create user_job_pool + add user_feed.exhausted_at.
+
+The actual change this deploy ships: the warm-pool table and the exhaustion-guard column. Additive +
+non-breaking (old code ignores them). On prod: stamp 0001 (existing schema), then this runs.
+
+Revision ID: 0002_pool_and_exhausted
+Revises: 0001_baseline
+Create Date: 2026-06-26
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0002_pool_and_exhausted"
+down_revision = "0001_baseline"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "user_job_pool",
+        sa.Column("user_id", sa.String(), primary_key=True),
+        sa.Column("query_sig", sa.String(), primary_key=True),
+        sa.Column("job_id", sa.String(), primary_key=True),
+        sa.Column("job", sa.JSON(), nullable=True),
+        sa.Column("pooled_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True),
+    )
+    op.add_column("user_feed", sa.Column("exhausted_at", sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("user_feed", "exhausted_at")
+    op.drop_table("user_job_pool")

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")
@@ -102,36 +151,64 @@ class MatchmakingAgentSession:
         await self._stub("opportunity_detail", "return match-score breakdown")
 
     async def handle_run_search(self, params: dict):
-        """Slice-1a: ScoutPrefs → live multi-board fetch → cards. No engine, no user-arm."""
+        """ScoutPrefs → warm-pool gate (serve banked jobs; sweep Apify only when the pool dips) →
+        sift → Opus curate → cards."""
         from app.engine import search
+        from app.config import get_settings
 
         prefs = params  # the frontend's ScoutPrefs arrive as params (user_context, if present, is ignored)
-        fresh = bool(prefs.get("_fresh"))  # dev toggle: force a live sweep instead of replaying cache
+        fresh = bool(prefs.get("_fresh"))  # dev toggle: force a live sweep instead of serving the pool
         if not search.has_mvq(prefs):
             await self.push("agent_data", action="search_complete",
                             data={"opportunities": [], "needs": ["title", "location"]})
             return
-        await self.push("agent_thinking",
-                        message="Fetching fresh roles across job boards…" if fresh
-                        else "Scanning live roles across job boards…")
-        # Per-board search cursors: re-running the SAME query advances EACH board's own page so every
-        # run fetches GENUINELY NEW jobs. Resets when the query changes OR >24h (boards refresh ~daily).
-        # Dedup is the per-user seen-net below — never cross-user, no opaque actor-side state.
         from app.db import repo as _repo
+        s = get_settings()
         sig = _repo.search_signature(prefs)
-        cursors = await _repo.get_search_cursors(self.user_id, sig)
-        prefs["_cursors"] = cursors
-        # Heartbeat through the long Apify scrape so the relay channel stays alive (see _with_heartbeat).
-        sweep = await self._with_heartbeat(
-            search.run_sweep(prefs, fresh=fresh),
-            ["Scanning live roles across the boards…", "Pulling the latest Naukri + LinkedIn + Foundit roles…",
-             "Gathering fresh postings for you…"])
-        # Safety net: drop anything already shown / applied / dismissed (covers boards that can't
-        # paginate, so the same posting never reappears even if a board re-returns it).
         seen = await _repo.get_seen_ids(self.user_id)
-        candidates = [o for o in sweep["opportunities"] if o.get("id") not in seen]
-        # Engine: assemble FULL context (profile + prefs). Cheap rankers SIFT 90 → top ~18 (white-box ‖
-        # embedding vibe); then Opus reads only those and curates the honest shortlist + report cards.
+
+        # ── Warm-pool gate ──────────────────────────────────────────────────────────────────────────
+        # Bank surplus fetched jobs per (user, query); serve from the pool and re-pay Apify only when the
+        # fresh-unseen count dips below the floor. Reorder EARLY (background top-up) so it never runs thin.
+        pooled = await _repo.pool_load_fresh_unseen(self.user_id, sig, seen) if s.POOL_ENABLED else []
+        exhausted = await _repo.is_pool_exhausted(self.user_id, sig) if s.POOL_ENABLED else False
+        decision = search.pool_decision(len(pooled), fresh, s.POOL_SAFETY_FLOOR, s.POOL_REORDER_AT, exhausted)
+
+        if decision == "sweep":
+            # Pool below the hard floor (or forced): blocking sweep → bank ALL of it. No cursors — every
+            # fetch pulls the freshest page; the seen-net dedups + daily board refresh brings new jobs.
+            await self.push("agent_thinking",
+                            message="Fetching fresh roles across job boards…" if fresh
+                            else "Scanning live roles across job boards…")
+            sweep = await self._with_heartbeat(
+                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",
+                    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:
+                await _repo.mark_pool_exhausted(self.user_id, sig)
+            pooled = await _repo.pool_load_fresh_unseen(self.user_id, sig, seen)
+            sources = sweep["sources"]
+        else:
+            # Pool healthy → serve from the shelf, zero Apify. (Cursors unchanged; refill, if any, is bg.)
+            await self.push("agent_thinking", message="Pulling your latest matches…")
+            sources = {"pool": len(pooled)}
+
+        candidates = pooled
+        # Engine: cheap rankers SIFT the pool → top ~22 (white-box ‖ embedding vibe); then Opus reads only
+        # those and curates the honest shortlist + report cards.
         from app.engine import curate as _curate
         from app.engine import rank as _rank
         from app.engine import sift as _sift
@@ -150,19 +227,13 @@ class MatchmakingAgentSession:
             engine = f"fallback:{dbg['mode']}"
         else:
             engine = "opus"
-        result = {"opportunities": curated, "sources": sweep["sources"],
+        result = {"opportunities": curated, "sources": sources,
                   "shortlisted": len(curated), "scanned": dbg["scored"], "engine": engine,
                   "has_profile": bool(user_context and user_context.get("skills"))}
-        # Persist the RAW deck + ADVANCE every board that ran (its page +1) so the next run of this
-        # query fetches new jobs. Per-opportunity state (applied/dismissed/docs) lives separately.
+        # Persist the deck (no cursors — pagination removed; freshness is date-sort + the seen-net).
         clean_prefs = {k: v for k, v in prefs.items()
                        if k not in ("user_context", "_fresh", "_cursors")}
-        # Store the LAST page fetched per paginating board (last + 1 = the page we just pulled). So a
-        # NEW query's first search stores 1 (page 1), not 2. Naukri/Foundit have no page param → no
-        # cursor (they dedup via the per-user seen-net).
-        next_cursors = {b: int(cursors.get(b, 0)) + 1
-                        for b in (sweep.get("sources") or {}) if b in search.PAGINATING_BOARDS}
-        await _repo.save_feed(self.user_id, clean_prefs, result, query_sig=sig, cursors=next_cursors)
+        await _repo.save_feed(self.user_id, clean_prefs, result, query_sig=sig)
         # Accumulate the salary band — each deck contributes its peak salary, averaged over decks
         # (survives decks that disclose nothing, like Naukri "Not disclosed").
         await _repo.update_salary_band(self.user_id, [o.get("salary_lpa") for o in (result["opportunities"] or [])])
@@ -173,6 +244,54 @@ 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.)
+        if decision == "pool+refill":
+            asyncio.create_task(self._background_refill(clean_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
+        fresh-unseen count climbs back above the reorder mark before the user's next search. No cursors —
+        the seen-net dedups and daily board refresh supplies new jobs. Best-effort, non-fatal."""
+        from app.engine import search
+        from app.db import repo as _repo
+        try:
+            from app.config import get_settings
+            sweep = await search.run_sweep(prefs, fresh=True)
+            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)
+            logger.info("pool refill (bg): +%d NEW jobs, sig=%s", new_count, sig[:8])
+        except Exception as e:  # noqa: BLE001
+            logger.warning("background refill failed (non-fatal): %s", e)
 
     async def handle_suggest_bubbles(self, params: dict):
         """Fine-tune bubbling — stage-aware (broad→narrow→role) bubbles from the picks so far."""
@@ -214,6 +333,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
@@ -230,6 +360,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,16 +31,22 @@ 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 = 18                           # candidate pool the curator reads
+    SIFT_TOP_K: int = 22                           # candidate pool the curator reads — widened from 18 for
+                                                   # recall, trimmed from 28 to keep the Opus payload lean
+    MATCH_FLOOR: int = 50                          # HARD filter — never surface a match scored below this
+    # Presentation calibration of the HEADLINE score (breakdown stays raw evidence). Transparent +
+    # monotonic + floor-anchored: a genuinely-strong match presents in the 90s; weak NEVER becomes strong.
+    CALIBRATION_ENABLED: bool = True
+    CALIBRATION_GAMMA: float = 1.3                  # 50→50, 70→74, 90→94, 95→97 (see rubric.calibrate)
 
     # Result cache (dev cost-saver). Keyed by (actor, exact input) → saved job set on disk.
     #   off       = always live, never save  (PRODUCTION default)
@@ -52,24 +58,44 @@ class Settings(BaseSettings):
     # Active boards — "Balanced" India stack: all three filter to the exact city at the board
     # (hyper-relevant) and surface offsite apply links. Naukri=blackfalcondata feed,
     # Foundit=Monster India, LinkedIn=geoId/city. Others (ats/indeed/naukri_v1) stay registered, off.
-    BOARDS_ENABLED: str = "naukri,foundit,linkedin"
+    BOARDS_ENABLED: str = "naukri,foundit,linkedin,timesjobs,workindia"   # signal-weighted; indeed dropped (0% skills, redundant)
+    BOARD_TIMEOUT_S: float = 55.0      # per-board cap — a slow/hung board is skipped, the search stays fast
+
+    # Warm pool ("the shelf") — bank surplus fetched jobs per (user, query) so we don't re-pay Apify for
+    # a full sweep on every search. Serve from the pool; refill only when it dips. Reorder EARLY (top up
+    # in the background) and hard-floor a blocking fetch so the pool never runs thin.
+    POOL_ENABLED: bool = True
+    POOL_REORDER_AT: int = 100         # fresh-unseen < this → background top-up (while still serving)
+    POOL_SAFETY_FLOOR: int = 90        # fresh-unseen < this → BLOCK + fetch before serving (hard floor)
+                                       # 100/90 keeps a DEEP pool (richer decks) at the cost of more
+                                       # frequent refills — we pay for volume, deliberately.
+    POOL_TTL_HOURS: int = 72           # banked jobs older than this are stale (likely filled) → dropped
+    POOL_MAX_PER_QUERY: int = 400      # cap the bank per (user, query) — STORAGE cap (kills DB swell) + load cap
+    # Exhaustion guard — a refill that adds fewer than this many NEW jobs means the query is tapped out;
+    # we then RELAX the floor (serve what's pooled, stop block-fetching duplicates) until it expires.
+    POOL_MIN_NEW_PER_REFILL: int = 8
+    POOL_EXHAUSTION_TTL_HOURS: int = 12   # re-try a tapped query after this (boards post new jobs ~daily)
 
     # Per-board fetch budget (the #1 Apify cost lever). Demo-sized to conserve credits.
-    NAUKRI_MAX_JOBS: int = 15       # blackfalcondata maxResults (no floor); fetchDetails=True for offsite apply links
-    FOUNDIT_MAX_JOBS: int = 50
-    LINKEDIN_MAX_JOBS: int = 25
+    # Signal-weighted budgets → ~150 jobs/sweep, tilted to the high-signal boards.
+    NAUKRI_MAX_JOBS: int = 30       # blackfalcondata maxResults; fetchDetails=True for offsite apply links
+    FOUNDIT_MAX_JOBS: int = 40
+    LINKEDIN_MAX_JOBS: int = 30
+    TIMESJOBS_MAX_JOBS: int = 35    # shahidirfan — the richest board (skills+salary+experience)
     # ATS lane (off by default) — company-seeded global/remote "dream companies".
     ATS_MAX_PER_COMPANY: int = 4
     ATS_COMPANIES: str = "stripe,databricks,gitlab,figma,ramp,notion,razorpay,zerodha"
     INDEED_MAX_JOBS: int = 30
+    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/models.py

@@ -27,6 +27,9 @@ class UserFeed(Base):
     # Dedup is purely per-user (the seen-net below) — never cross-user, never opaque actor-side state.
     query_sig: Mapped[str] = mapped_column(String, default="")
     cursors: Mapped[dict] = mapped_column(JSON, default=dict)
+    # Set when a refill for `query_sig` returned almost no NEW jobs (the query is tapped out). While set
+    # (within POOL_EXHAUSTION_TTL), the pool gate relaxes the floor instead of block-fetching duplicates.
+    exhausted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     search_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")  # total searches run (engagement)
     # Salary band = AVERAGE of each deck's HIGHEST disclosed salary (₹L). Each deck contributes one
     # peak; the band is the running mean of those peaks. Survives decks that disclose nothing.
@@ -37,6 +40,23 @@ class UserFeed(Base):
     )
 
 
+class UserJobPool(Base):
+    """The WARM POOL (the 'shelf'). Surplus fetched-but-not-yet-shown jobs banked per (user, query) so
+    subsequent searches serve from here instead of re-paying Apify for a fresh sweep every time. One row
+    per (user, query_sig, job). TTL'd by `pooled_at`; already-shown jobs are filtered on read via the
+    seen-net. The gate reorders (background top-up) at REORDER fresh-unseen and hard-floors a blocking
+    fetch below SAFETY_FLOOR, so the pool never runs thin."""
+    __tablename__ = "user_job_pool"
+
+    user_id: Mapped[str] = mapped_column(String, primary_key=True)
+    query_sig: Mapped[str] = mapped_column(String, primary_key=True)
+    job_id: Mapped[str] = mapped_column(String, primary_key=True)
+    job: Mapped[dict] = mapped_column(JSON, default=dict)          # the normalized ScoutJob (pre-sift/curate)
+    pooled_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+
+
 class OpportunityState(Base):
     """Per-(user, opportunity) state that SURVIVES deck refreshes. Persists: `applied`/`dismissed`
     status, the resume-builder document ids already generated (never re-pay to re-tailor), and the

app/db/repo.py

@@ -9,7 +9,7 @@ import logging
 from sqlalchemy import func, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
-from app.db.models import Base, OpportunityState, UserFeed
+from app.db.models import Base, OpportunityState, UserFeed, UserJobPool
 from app.db.session import engine, session_factory
 
 logger = logging.getLogger(__name__)
@@ -17,14 +17,35 @@ 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}
+    # Canonicalize the fields exactly as the board adapters consume them. The
+    # dashboard contract is camelCase; work_mode remains a legacy fallback.
+    work_mode = prefs.get("workMode") if "workMode" in prefs else prefs.get("work_mode")
+    payload = {
+        "effective_title": _effective_title(prefs),
+        "role": prefs.get("role"),
+        "location": prefs.get("location"),
+        "industry": prefs.get("industry"),
+        "experience": prefs.get("experience"),
+        "years": prefs.get("years"),
+        "workMode": work_mode,
+        "remote": prefs.get("remote"),
+    }
     blob = json.dumps(payload, sort_keys=True, default=str)
     return hashlib.sha1(blob.encode()).hexdigest()[:16]
 
@@ -346,7 +367,10 @@ async def mark_seen(user_id: str, opportunity_ids: list[str]) -> None:
     factory = session_factory()
     if factory is None or not user_id or not opportunity_ids:
         return
-    rows = [{"user_id": user_id, "opportunity_id": oid, "seen": True} for oid in opportunity_ids if oid]
+    # DEDUP the ids — a duplicate (user, oppid) in ONE insert trips PG's "ON CONFLICT cannot affect row
+    # a second time" (CardinalityViolationError). A deck can carry the same id twice (cross-board dupes).
+    seen_ids = list(dict.fromkeys(o for o in opportunity_ids if o))
+    rows = [{"user_id": user_id, "opportunity_id": oid, "seen": True} for oid in seen_ids]
     if not rows:
         return
     try:
@@ -360,3 +384,141 @@ async def mark_seen(user_id: str, opportunity_ids: list[str]) -> None:
             await s.commit()
     except Exception as e:  # noqa: BLE001
         logger.warning("mark_seen failed (non-fatal): %s", e)
+
+
+# ── Warm pool (the shelf) ────────────────────────────────────────────────────
+async def pool_save(user_id: str, query_sig: str, jobs: list[dict]) -> int:
+    """Bank fetched jobs into the per-(user, query) pool (upsert by job id; refresh pooled_at). Returns
+    the count of GENUINELY NEW job ids (not already pooled) — the exhaustion signal. Then TTL-sweeps stale
+    rows AND caps storage to POOL_MAX_PER_QUERY freshest (kills DB swell). 0 if the DB is off."""
+    from datetime import datetime, timedelta, timezone
+    from app.config import get_settings
+    factory = session_factory()
+    if factory is None or not user_id or not jobs:
+        return 0
+    rows = [{"user_id": user_id, "query_sig": query_sig, "job_id": j["id"], "job": j}
+            for j in jobs if j.get("id")]
+    if not rows:
+        return 0
+    s = get_settings()
+    cutoff = datetime.now(timezone.utc) - timedelta(hours=s.POOL_TTL_HOURS)
+    incoming = {r["job_id"] for r in rows}
+    try:
+        async with factory() as sess:
+            # how many of these are NEW (not already pooled for this query)?
+            existing = set((await sess.execute(
+                select(UserJobPool.job_id).where(
+                    (UserJobPool.user_id == user_id) & (UserJobPool.query_sig == query_sig)
+                    & (UserJobPool.job_id.in_(incoming)))
+            )).scalars().all())
+            new_count = len(incoming - existing)
+
+            stmt = pg_insert(UserJobPool).values(rows)
+            stmt = stmt.on_conflict_do_update(
+                index_elements=[UserJobPool.user_id, UserJobPool.query_sig, UserJobPool.job_id],
+                set_={"job": stmt.excluded.job, "pooled_at": func.now()},
+            )
+            await sess.execute(stmt)
+            # TTL sweep (stale postings, likely filled).
+            await sess.execute(UserJobPool.__table__.delete().where(
+                (UserJobPool.user_id == user_id) & (UserJobPool.query_sig == query_sig)
+                & (UserJobPool.pooled_at < cutoff)))
+            # STORAGE cap: keep only the POOL_MAX_PER_QUERY freshest for this (user, query).
+            keep = (await sess.execute(
+                select(UserJobPool.job_id)
+                .where((UserJobPool.user_id == user_id) & (UserJobPool.query_sig == query_sig))
+                .order_by(UserJobPool.pooled_at.desc()).limit(s.POOL_MAX_PER_QUERY)
+            )).scalars().all()
+            if len(keep) >= s.POOL_MAX_PER_QUERY:
+                await sess.execute(UserJobPool.__table__.delete().where(
+                    (UserJobPool.user_id == user_id) & (UserJobPool.query_sig == query_sig)
+                    & (UserJobPool.job_id.notin_(keep))))
+            await sess.commit()
+            return new_count
+    except Exception as e:  # noqa: BLE001
+        logger.warning("pool_save failed (non-fatal): %s", e)
+        return 0
+
+
+async def mark_pool_exhausted(user_id: str, query_sig: str) -> None:
+    """Flag this (user, query) as tapped out (a refill added almost no new jobs) → the gate relaxes the
+    floor until POOL_EXHAUSTION_TTL passes. Upsert on the one-row-per-user feed."""
+    factory = session_factory()
+    if factory is None or not user_id:
+        return
+    try:
+        async with factory() as s:
+            stmt = pg_insert(UserFeed).values(user_id=user_id, query_sig=query_sig, exhausted_at=func.now())
+            stmt = stmt.on_conflict_do_update(
+                index_elements=[UserFeed.user_id],
+                set_={"exhausted_at": func.now(), "query_sig": query_sig},
+            )
+            await s.execute(stmt)
+            await s.commit()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("mark_pool_exhausted failed (non-fatal): %s", e)
+
+
+async def is_pool_exhausted(user_id: str, query_sig: str) -> bool:
+    """True if this exact query was marked exhausted within POOL_EXHAUSTION_TTL (boards refresh ~daily,
+    so we re-try after it expires). False on any other query / DB off / expired."""
+    from datetime import datetime, timedelta, timezone
+    from app.config import get_settings
+    factory = session_factory()
+    if factory is None or not user_id:
+        return False
+    try:
+        async with factory() as s:
+            row = (await s.execute(select(UserFeed).where(UserFeed.user_id == user_id))).scalar_one_or_none()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("is_pool_exhausted failed (non-fatal): %s", e)
+        return False
+    if not row or row.query_sig != query_sig or not row.exhausted_at:
+        return False
+    ttl = get_settings().POOL_EXHAUSTION_TTL_HOURS
+    return (datetime.now(timezone.utc) - row.exhausted_at).total_seconds() < ttl * 3600
+
+
+async def pool_load_fresh_unseen(user_id: str, query_sig: str, seen: set[str]) -> list[dict]:
+    """The banked jobs for this query that are still FRESH (within TTL) and UNSEEN (not already shown),
+    freshest first. This is the count the reorder/floor gate reads. Empty list if the DB is off."""
+    from datetime import datetime, timedelta, timezone
+    from app.config import get_settings
+    factory = session_factory()
+    if factory is None or not user_id:
+        return []
+    s = get_settings()
+    cutoff = datetime.now(timezone.utc) - timedelta(hours=s.POOL_TTL_HOURS)
+    try:
+        async with factory() as sess:
+            rows = (await sess.execute(
+                select(UserJobPool.job_id, UserJobPool.job)
+                .where((UserJobPool.user_id == user_id)
+                       & (UserJobPool.query_sig == query_sig)
+                       & (UserJobPool.pooled_at >= cutoff))
+                .order_by(UserJobPool.pooled_at.desc())
+                .limit(s.POOL_MAX_PER_QUERY)
+            )).all()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("pool_load failed (non-fatal): %s", e)
+        return []
+    return [r.job for r in rows if r.job_id not in seen]
+
+
+async def save_cursors(user_id: str, query_sig: str, cursors: dict) -> None:
+    """Persist advanced per-board cursors WITHOUT touching the saved feed — used by the background
+    refill (which sweeps + banks but has no new deck to show). Upsert on the one-row-per-user feed."""
+    factory = session_factory()
+    if factory is None or not user_id:
+        return
+    try:
+        async with factory() as s:
+            stmt = pg_insert(UserFeed).values(user_id=user_id, query_sig=query_sig, cursors=cursors or {})
+            stmt = stmt.on_conflict_do_update(
+                index_elements=[UserFeed.user_id],
+                set_={"cursors": cursors or {}, "query_sig": query_sig},
+            )
+            await s.execute(stmt)
+            await s.commit()
+    except Exception as e:  # noqa: BLE001
+        logger.warning("save_cursors failed (non-fatal): %s", e)

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/board_adapters/adapters.py

@@ -7,6 +7,7 @@ System knobs (budget / freshness) are arguments with sane defaults, not user inp
 """
 from __future__ import annotations
 from typing import Any
+from urllib.parse import urlencode
 
 from . import coerce as C
 
@@ -75,7 +76,9 @@ def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details:
     # Apify. Cross-deck dedup is the PER-USER seen-net in our DB. (The feed actor has no page/offset.)
     city = next((lp["city"] for lp in C.parsed_locations(prefs) if lp["city"]), "")
     return {
-        "keyword": C.title_of(prefs) or C.build_keyword(prefs),
+        # Targeted keyword (seniority + industry + a top skill) → fetch RIGHT-level/industry/skill jobs,
+        # not bare-title breadth. Text-targeting degrades gracefully (never zeroes); the pool gives breadth.
+        "keyword": C.build_keyword(prefs, seniority=True, industry=True, skills=True),
         "location": city,                # precise city, e.g. "New Delhi"
         "maxResults": max_jobs,
         "fetchDetails": fetch_details,
@@ -84,20 +87,22 @@ def build_naukri_feed_input(prefs: Prefs, *, max_jobs: int = 50, fetch_details:
     }
 
 
-# ── Foundit / Monster India (city-precise; apply_url is the offsite redirect) ──
-# Foundit uses legacy city names — it returns 0 for "New Delhi"/"Bengaluru".
-_FOUNDIT_CITY = {
+# Legacy India job boards (Foundit/Monster, TimesJobs) use OLD city names — they return 0 for the
+# modern "New Delhi"/"Bengaluru"/"Gurugram". Map to their canonical names. (Verified live: TimesJobs
+# returns 0 for "New Delhi", 15 for "Delhi".)
+_LEGACY_IN_CITY = {
     "new delhi": "Delhi", "delhi ncr": "Delhi", "bengaluru": "Bangalore",
     "gurugram": "Gurgaon", "mumbai": "Mumbai", "pune": "Pune", "hyderabad": "Hyderabad",
     "chennai": "Chennai", "kolkata": "Kolkata", "noida": "Noida",
 }
+_FOUNDIT_CITY = _LEGACY_IN_CITY   # back-compat alias
 
 
 def build_foundit_input(prefs: Prefs, *, results_wanted: int = 50) -> dict:
     city = next((lp["city"] for lp in C.parsed_locations(prefs) if lp["city"]), "")
     city = _FOUNDIT_CITY.get(city.lower(), city)  # alias to Foundit's canonical name
     return {
-        "keyword": C.title_of(prefs) or C.build_keyword(prefs),
+        "keyword": C.build_keyword(prefs, seniority=True, industry=True, skills=True),  # targeted
         "location": city,
         "results_wanted": results_wanted,
     }
@@ -109,8 +114,13 @@ def build_linkedin_input(prefs: Prefs, *, max_items: int = 80, posted: str = "we
     # recall=True (retrieve stage): title + COUNTRY-level location only — drop the narrowing
     # filters (easyApply / workplaceType / experienceLevel / industryIds / freshness), which
     # become ENGINE ranking later. City-level location + a wrong industryId zero the actor out.
+    # jobTitles is a TITLE field — prepend the seniority band (real titles, e.g. "Senior Sales Manager")
+    # to target right-level roles. Industry stays OUT of the title field (it's not part of a job title).
+    _title = C.title_of(prefs)
+    if C.seniority_band(prefs) in ("Senior", "Lead") and _title:
+        _title = f"{C.seniority_band(prefs)} {_title}"
     out: dict[str, Any] = {
-        "jobTitles": [C.title_of(prefs)],
+        "jobTitles": [_title],
         "maxItems": max_items,
         "sortBy": "date",
         "page": page,                    # Start Page Number — advance per re-run for NEW jobs
@@ -156,16 +166,15 @@ def build_linkedin_input(prefs: Prefs, *, max_items: int = 80, posted: str = "we
 
 # ── Indeed (blunt: what + where + country only; everything else folds in) ────
 def build_indeed_input(prefs: Prefs, *, max_items: int = 80) -> dict:
+    # valig~indeed-jobs-scraper: title + location + country(lowercase ISO2) + limit. ~7s/run (vs misceres 52s).
     loc = next((l for l in C.parsed_locations(prefs) if l["city"] or l["country"]), {})
     country_name = loc.get("country") or "India"
     return {
-        # Indeed has no work-mode/seniority/industry params → fold them into the query.
-        "position": C.build_keyword(prefs, seniority=True, industry=True),
+        # Indeed has no work-mode/seniority/industry params → fold them into the title query.
+        "title": C.build_keyword(prefs, seniority=True, industry=True),
         "location": loc.get("city") or "",
-        "country": C.COUNTRY_ISO2.get(country_name, "IN"),
-        "maxItemsPerSearch": max_items,
-        "followApplyRedirects": True,   # → real offsite apply URL
-        "saveOnlyUniqueItems": True,
+        "country": C.COUNTRY_ISO2.get(country_name, "IN").lower(),   # valig wants lowercase ("in")
+        "limit": max_items,
     }
 
 
@@ -208,6 +217,68 @@ def build_ats_input(prefs: Prefs, *, companies: list[dict], recent_days: int = 3
     return out
 
 
+# ── WorkIndia (India blue/grey-collar — the non-tech workhorse: sales, BPO, retail, field, ops) ──
+def build_linkedin_cc_input(prefs: Prefs, *, count: int = 30) -> dict:
+    # curious_coder~linkedin-jobs-scraper takes LinkedIn SEARCH URLs (structured industry+jobFunction+
+    # applicants). NO pagination — the actor only honors start=0 (start>0 returns empty). Every fetch
+    # pulls the freshest page-1 by date; the per-user seen-net dedups, and daily board refresh feeds new jobs.
+    kw = C.build_keyword(prefs, seniority=True, industry=True)   # targeted search string
+    loc = next((f"{lp['city']}, {lp['country']}" if lp["country"] else lp["city"]
+                for lp in C.parsed_locations(prefs) if lp["city"]), None) \
+        or next((lp["country"] for lp in C.parsed_locations(prefs) if lp["country"]), "India")
+    q = urlencode({"keywords": kw, "location": loc, "sortBy": "DD"})  # DD = newest first
+    return {"urls": [f"https://www.linkedin.com/jobs/search/?{q}"],
+            "count": max(10, count), "scrapeCompany": False}   # count min 10 (actor constraint)
+
+
+def build_timesjobs_input(prefs: Prefs, *, results_wanted: int = 35) -> dict:
+    # shahidirfan/Timesjobs-Scraper — FULL signal (skills+salary+experience+offsite external_job_url).
+    # keyword + city + an experience BAND ("5-10") for right-level targeting. Clean keyword input, no page.
+    city = next((lp["city"] for lp in C.parsed_locations(prefs) if lp["city"]), "")
+    city = _LEGACY_IN_CITY.get(city.lower(), city)   # TimesJobs needs "Delhi" not "New Delhi" (verified)
+    out: dict[str, Any] = {
+        "keyword": C.build_keyword(prefs, seniority=True, industry=True, skills=True),
+        "results_wanted": results_wanted,
+    }
+    if city:
+        out["location"] = city
+    yrs = C.years_of(prefs)
+    if yrs is not None:
+        out["experience"] = f"{max(0, yrs - 2)}-{yrs + 3}"   # a band around the candidate's years
+    return out
+
+
+def build_workindia_input(prefs: Prefs, *, results_wanted: int = 20) -> dict:
+    # keyword + city; includeDetails=True pulls the real skills + full description (verified live Phase 0).
+    # No page param → per-user seen-net dedups. City is a lowercased name slug ("mumbai", "pune").
+    city = next((lp["city"] for lp in C.parsed_locations(prefs) if lp["city"]), "")
+    out: dict[str, Any] = {
+        "keyword": C.title_of(prefs) or C.build_keyword(prefs),
+        "results_wanted": results_wanted,
+        "includeDetails": True,
+    }
+    if city:
+        out["city"] = city.lower()
+    return out
+
+
+# ── Wellfound / AngelList (startup roles — equity + skills; remote-friendly for India) ──
+def build_wellfound_input(prefs: Prefs, *, max_results: int = 15) -> dict:
+    # query=title; city → location slug (lowercase-hyphen); remote honored. enrichDetail=True is
+    # REQUIRED — it pulls skills[] + the full description the curator scores on (the core fields
+    # alone are thin). No page param → dedup via the per-user seen-net (like Naukri/Foundit).
+    city = next((lp["city"] for lp in C.parsed_locations(prefs) if lp["city"]), "")
+    out: dict[str, Any] = {
+        "query": C.title_of(prefs) or C.build_keyword(prefs),
+        "maxResults": max_results,
+        "enrichDetail": True,                 # → skills + description (the curator's signal)
+        "remote": C.is_remote(prefs),
+    }
+    if city:
+        out["location"] = city.lower().replace(" ", "-")   # "New Delhi" → "new-delhi"
+    return out
+
+
 if __name__ == "__main__":
     import json
     demo: Prefs = {

app/engine/board_adapters/coerce.py

@@ -102,12 +102,23 @@ def years_of(prefs: Prefs) -> Optional[int]:
     b = seniority_band(prefs)
     return SENIORITY_YEARS.get(b) if b else None
 
-def build_keyword(prefs: Prefs, *, seniority: bool = False, industry: bool = False) -> str:
-    """Composite keyword for boards lacking department/industry filters."""
+def _top_skill(prefs: Prefs) -> Optional[str]:
+    """The candidate's single most relevant skill (from the resume context), for keyword targeting.
+    One skill only — more terms over-narrow the board's text search."""
+    sk = (prefs.get("user_context") or {}).get("skills") or prefs.get("_skills") or []
+    return sk[0] if sk else None
+
+
+def build_keyword(prefs: Prefs, *, seniority: bool = False, industry: bool = False,
+                  skills: bool = False) -> str:
+    """Composite keyword for boards lacking department/industry filters. Targets RIGHTER jobs (seniority
+    + industry + a top skill folded into the text) so the deck aligns on more rubric dimensions."""
     parts: list[str] = []
     if seniority and seniority_band(prefs) in ("Senior", "Lead"):
         parts.append(seniority_band(prefs))  # prefix only the high bands
     parts.append(title_of(prefs))
+    if skills and (sk := _top_skill(prefs)):
+        parts.append(sk)                     # ONE top skill — surfaces jobs needing it (skills dim ↑)
     if industry:
         inds = prefs.get("industry") or []
         if inds and inds[0] not in ("Any", None):

app/engine/curate.py

@@ -11,50 +11,86 @@ import json
 import logging
 
 from app.config import get_settings
+from app.engine import rubric as _rubric
 from app.engine.llm import curate_enabled, gateway_client
 from app.engine.schema import Growth, MatchDim, MatchResult
 
 logger = logging.getLogger(__name__)
 
-_SYSTEM = """You are Scout's senior recruiter. You receive a candidate's PROFILE and a pre-sifted list of \
-JOBS (already narrowed to the right city and title by cheaper tools). Two jobs:
+# Two-stage curation, by capability + cost:
+#   STAGE 1 (Opus, _SYSTEM_SCORE): the JUDGMENT — rate the rubric dimensions on evidence. Integrity-
+#     critical, so it stays on Opus. Lean output (scores + terse notes, no prose) → tighter calibration
+#     AND ~75% less Opus output cost.
+#   STAGE 2 (Haiku, _SYSTEM_CARDS): the PROSE — write the report cards FROM Opus's scores+notes. Cheap
+#     output ($5/M vs $75/M); Haiku never judges fit or picks a score, only writes.
+# The overall score is COMPUTED by the engine (rubric.aggregate) from Stage-1 ratings — never model-emitted.
 
-1. CURATE — keep ONLY the jobs genuinely worth this candidate's time. Be honest: if only 8 of the jobs \
-are real matches, return 8. NEVER pad to a number. Drop weak, duplicate, or off-target roles.
+_SYSTEM_SCORE = (
+    "You are Scout's senior recruiter scoring job fit. You receive a candidate PROFILE and a pre-sifted "
+    "list of JOBS (already narrowed to the right city + title by cheaper tools). Do two things:\n\n"
+    "1. CURATE — keep ONLY jobs genuinely worth this candidate's time. Be honest: if only 8 are real "
+    "matches, return 8. NEVER pad, drop weak/duplicate/off-target roles. The OVERALL score is COMPUTED "
+    "from your ratings — you do NOT choose it — so a job earns its place on merit, not a number you set.\n\n"
+    "2. For each KEPT job, RATE these fixed rubric dimensions 0-100, EACH against its anchor, each with a "
+    "SHORT evidence note (a few words) grounded in THIS job's real content. Rate every dimension that "
+    'applies (omit "industry" only if the candidate states no industry preference). Never omit a hard '
+    "dimension to flatter the score — you rate EVIDENCE, not an outcome.\n"
+    + _rubric.prompt_block() + "\n\n"
+    "Rate against the ANCHORS honestly and use the FULL range: a same-function, same-city, same-level job "
+    "where the candidate genuinely meets most requirements earns dimension scores in the 90s (so the "
+    "computed overall is a confident 90s match — that hyper-relevance is the goal); a partial match earns "
+    "less; a weak dimension MUST score low and drag the overall down. Don't inflate, don't undersell.\n\n"
+    "NON-TECH roles (sales, finance, HR, ops, support, …): rate \"skills\" as the share of the job's stated "
+    "RESPONSIBILITIES the candidate can do — NOT a tag checklist; a missing skill-tag list is not a negative.\n\n"
+    'Also emit fit: "fit" (current-state match) or "stretch" (a genuine reach).\n\n'
+    "Respond with ONLY a JSON object (no prose, no markdown fences):\n"
+    '{"kept":[{"id","fit","dimensions":{"role":{"score","note"},"skills":{"score","note"},'
+    '"seniority":{"score","note"},"location":{"score","note"},"industry":{"score","note"},'
+    '"experience":{"score","note"}}}]}\n'
+    'Use each job\'s exact "id".'
+)
 
-2. For each KEPT job, write a REPORT CARD in this voice (modeled on a supportive video-coaching analysis):
-   - archetype: a short, characterful label — e.g. "The Stretch Worth Taking", "The Safe Powerhouse", \
-"The Skill-Adjacent Pivot". Memorable, never generic.
-   - one_line: ONE honest, balanced sentence — name the strength AND the gap. Never pure hype.
-   - breakdown: 3-4 dimensions (e.g. Skill match, Experience, Location, Industry), each with an honest \
-0-100 score, a level ("Strong" >=72 / "Solid" 52-71 / "Light" <52), and a specific one-line note grounded \
-in THIS job's actual skills/title.
-   - coach_note: one warm, actionable line — what to lead with, or what to shore up before applying.
-   - growth (optional): {"text","from","to"} — the gap to close and the score lift closing it buys.
-   - score: honest 0-100 overall fit. fit: "fit" (a current-state match) or "stretch" (a genuine reach).
+_SYSTEM_CARDS = (
+    "You are Scout's encouraging career coach. For each scored job below you get its title, company, the "
+    "candidate's situation, the overall fit score, and the recruiter's per-dimension EVIDENCE notes. Write "
+    "a short REPORT CARD per job, grounded ONLY in those notes — invent nothing, never contradict a score.\n\n"
+    "Voice: warm, specific, honest (supportive video-coaching). For each id, write:\n"
+    '  - archetype: a short characterful label — e.g. "The Stretch Worth Taking", "The Safe Powerhouse".\n'
+    "  - one_line: ONE balanced sentence — name the real strength AND the honest gap. Never pure hype.\n"
+    "  - coach_note: one actionable line — what to lead with, or what to shore up before applying.\n"
+    '  - growth (optional): {"text","from","to"} — text = the gap to close; from/to = NUMBERS, the score '
+    "now and the realistic score after closing it (e.g. from 82 to 90).\n\n"
+    "Respond with ONLY a JSON object (no prose/fences):\n"
+    '{"cards":[{"id","archetype","one_line","coach_note","growth":{"text","from","to"}}]}\n'
+    'Use each job\'s exact "id".'
+)
 
-Honesty rules: scores reflect reality — a partial match is in the 60s-70s, not the 90s. Ground every note \
-in the job's real content; invent nothing. Frame as helpful guidance, not a hiring verdict.
 
-Respond with ONLY a JSON object (no prose, no markdown fences):
-{"kept":[{"id","score","fit","archetype","one_line","breakdown":[{"name","score","level","note"}],\
-"coach_note","growth":{"text","from","to"}}]}
-Order "kept" best-first. Use each job's exact "id"."""
+def _compact(d: dict) -> dict:
+    """Drop None / empty values so the LLM payload stays lean (no null noise)."""
+    return {k: v for k, v in d.items() if v not in (None, "", [], {})}
 
 
 def _profile_brief(prefs: dict, ctx: dict | None) -> dict:
     ctx = ctx or {}
-    return {
+    # Richer profile = the rubric can SEE that skills/experience are genuinely met (so those dimensions
+    # score what they deserve, not a guess). All fields are free (already in the resume context).
+    return _compact({
         "target_title": prefs.get("title"),
         "target_roles": prefs.get("role"),
         "target_location": prefs.get("location"),
         "seniority": prefs.get("experience"),
-        "years": prefs.get("years"),
+        "years": prefs.get("years") or ctx.get("years_experience"),
         "target_industry": prefs.get("industry"),
-        "skills": ctx.get("skills"),
+        "skills": (ctx.get("skills") or [])[:20],
         "current_role": ctx.get("current_role"),
+        "current_company": ctx.get("current_company"),
+        "past_titles": (ctx.get("past_titles") or ctx.get("titles") or [])[:4],
+        "experience_summary": str(ctx.get("experience_summary") or ctx.get("summary") or "")[:600] or None,
+        "education": ctx.get("education"),
+        "industries_worked": ctx.get("industries"),
         "stretch_appetite": prefs.get("stretch", "balanced"),
-    }
+    })
 
 
 def _job_brief(j: dict) -> dict:
@@ -64,8 +100,12 @@ def _job_brief(j: dict) -> dict:
         "title": j.get("title"),
         "company": j.get("organization"),
         "location": f"{j.get('location_city') or ''} {j.get('location_country') or ''}".strip(),
-        "skills": (d.get("skills") or [])[:12],
-        "snippet": (d.get("description") or "")[:500],
+        "role_category": d.get("role_category"),
+        "industry": d.get("industry"),
+        "seniority": j.get("seniority_level"),
+        "pay": j.get("payLabel"),
+        "skills": (d.get("skills") or j.get("required_skills") or [])[:10],
+        "responsibilities": (d.get("description") or "")[:600],   # enough for Opus to judge role-fit; keeps the payload lean
         "prelim_score": j.get("matchScore"),
     }
 
@@ -81,60 +121,181 @@ def _parse_json(text: str) -> dict:
     return json.loads(t)
 
 
-def _to_match(k: dict) -> MatchResult:
-    dims = [
-        MatchDim(name=str(d["name"]), score=int(d["score"]), level=d["level"], note=d.get("note"))
-        for d in (k.get("breakdown") or [])
-    ]
-    g = k.get("growth")
-    growth = Growth(text=g["text"], **{"from": int(g["from"]), "to": int(g["to"])}) if g else None
+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)
+        s = text[i:] if i != -1 else text
+    out: list[dict] = []
+    depth, start = 0, None
+    for idx, ch in enumerate(s):
+        if ch == "{":
+            if depth == 0:
+                start = idx
+            depth += 1
+        elif ch == "}" and depth:
+            depth -= 1
+            if depth == 0 and start is not None:
+                try:
+                    out.append(json.loads(s[start : idx + 1]))
+                except Exception:  # noqa: BLE001 — skip the broken object, keep the rest
+                    pass
+                start = None
+    return out
+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"')
+
+
+def _scored_card(k: dict) -> dict | None:
+    """Stage-1 result → a partial card carrying the COMPUTED overall + the rubric breakdown (no prose yet).
+    Returns None for a malformed entry (no dimensions). The engine computes the score — never the model."""
+    raw = k.get("dimensions") or {}
+    dim_scores = {key: float(raw[key]["score"]) for key in _rubric.KEYS
+                  if isinstance(raw.get(key), dict) and raw[key].get("score") is not None}
+    if not dim_scores:
+        return None
+    overall = _rubric.aggregate(dim_scores)
+    dims = [{"key": key, "name": _rubric.LABELS[key], "score": round(dim_scores[key]),
+             "level": _rubric.level_for(dim_scores[key]),
+             "note": (raw[key].get("note") if isinstance(raw.get(key), dict) else None)}
+            for key in _rubric.KEYS if key in dim_scores]
+    return {"id": k.get("id"), "overall": overall,
+            "fit": k.get("fit") or ("fit" if overall >= 75 else "stretch"), "dims": dims}
+
+
+def _to_match(sc: dict, prose: dict | None) -> MatchResult:
+    """Combine a Stage-1 scored card (sc) with Stage-2 prose (or templated fallback) → the MatchResult."""
+    prose = prose or {}
+    dims = [MatchDim(name=d["name"], score=d["score"], level=d["level"], note=d.get("note")) for d in sc["dims"]]
+    growth = None
+    g = prose.get("growth")
+    if g:
+        try:
+            growth = Growth(text=g["text"], **{"from": int(g["from"]), "to": int(g["to"])})
+        except (KeyError, ValueError, TypeError):
+            growth = None
+    # Fallback prose (when Haiku gave nothing) is built from Opus's REAL evidence notes — never a generic
+    # "Strong on X". A graceful card with real content, even if the writer stage failed.
+    strong = [d for d in sc["dims"] if d.get("note") and d["score"] >= 72]
+    weak = min((d for d in sc["dims"] if d.get("note")), key=lambda d: d["score"], default=None)
+    fb_line = "; ".join(d["note"] for d in strong[:2]) if strong else (
+        (max(sc["dims"], key=lambda d: d["score"])["note"] if sc["dims"] and sc["dims"][0].get("note")
+         else "A solid match worth a look."))
+    fb_coach = (f"Shore up {weak['name'].lower()}: {weak['note']}" if weak and weak["score"] < 72 else
+                "Lead with your strongest, most relevant experience.")
+    one_line = prose.get("one_line") or fb_line
     return MatchResult(
-        score=int(k["score"]),
-        fit=k["fit"],
-        archetype=k.get("archetype"),
-        one_line=k.get("one_line"),
-        reason=k.get("one_line") or k.get("archetype") or "Curated by Scout",
-        breakdown=dims,
-        growth=growth,
-        coach_note=k.get("coach_note"),
-        proofReady=any(d.level == "Strong" for d in dims),
-        factors={},
+        score=sc["overall"], fit=sc["fit"],
+        archetype=prose.get("archetype") or ("The Strong Match" if sc["overall"] >= 80 else "Worth a Look"),
+        one_line=one_line, reason=one_line,
+        breakdown=dims, growth=growth, coach_note=prose.get("coach_note") or fb_coach,
+        proofReady=any(d.level == "Strong" for d in dims), factors={},
     )
 
 
-def curate(prefs: dict, ctx: dict | None, sifted_jobs: list[dict]) -> list[dict] | None:
-    """Opus reads the sifted pool → the curated jobs (kept, best-first, with real report cards).
-    `None` on any failure so the caller degrades to the sift + templated cards."""
+def _score_stage(prefs: dict, ctx: dict | None, sifted_jobs: list[dict], s) -> list[dict] | None:
+    """STAGE 1 (Opus): rate the rubric dimensions → scored cards (computed overall, ≥ floor, best-first).
+    None on a hard failure (caller falls back to the sift). Empty list = Opus kept nothing strong."""
     client = gateway_client()
-    if not (curate_enabled() and client and sifted_jobs):
-        return None
-    s = get_settings()
     payload = {"profile": _profile_brief(prefs, ctx), "jobs": [_job_brief(j) for j in sifted_jobs]}
     try:
         resp = client.chat.completions.create(
             model=s.CURATE_MODEL,
-            messages=[
-                {"role": "system", "content": _SYSTEM},
-                {"role": "user", "content": json.dumps(payload, ensure_ascii=False)},
-            ],
-            max_tokens=4000,
+            messages=[{"role": "system", "content": _SYSTEM_SCORE},
+                      {"role": "user", "content": json.dumps(payload, ensure_ascii=False)}],
+            max_tokens=4000,   # scores + terse notes only (no prose) → far smaller than the old combined call
+            timeout=120,
         )
-        data = _parse_json(resp.choices[0].message.content)
-    except Exception as e:  # noqa: BLE001 — any failure → fall back to the sift
-        logger.warning("curate (Opus) failed, falling back to sift: %s", e)
+        content = resp.choices[0].message.content
+    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 (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
+    # calibration (monotonic + floor-anchored, so ordering holds and weak never becomes strong).
+    cards = [c for c in (_scored_card(k) for k in kept) if c and c["overall"] >= s.MATCH_FLOOR]
+    if s.CALIBRATION_ENABLED:
+        for c in cards:
+            c["overall"] = _rubric.calibrate(c["overall"], floor=s.MATCH_FLOOR, gamma=s.CALIBRATION_GAMMA)
+    cards.sort(key=lambda c: -c["overall"])       # best-first by the (calibrated) score
+    return cards
 
+
+def _cards_stage(prefs: dict, ctx: dict | None, scored: list[dict], by_id: dict, s) -> dict:
+    """STAGE 2 (Haiku): write report-card prose FROM Opus's scores+notes. Cheap output, never judges.
+    Returns {id: prose}. On any failure → {} so _to_match uses templated prose over the real scores."""
+    client = gateway_client()
+    if not client or not scored:
+        return {}
+    briefs = [{"id": c["id"], "title": (by_id.get(c["id"]) or {}).get("title"),
+               "company": (by_id.get(c["id"]) or {}).get("organization"),
+               "overall": c["overall"], "fit": c["fit"],
+               "evidence": {d["key"]: d.get("note") for d in c["dims"]}} for c in scored]
+    payload = {"profile": _profile_brief(prefs, ctx), "scored_jobs": briefs}
+    try:
+        resp = client.chat.completions.create(
+            model=s.SUGGEST_MODEL,    # Haiku — fast + ~15x cheaper output than Opus
+            messages=[{"role": "system", "content": _SYSTEM_CARDS},
+                      {"role": "user", "content": json.dumps(payload, ensure_ascii=False)}],
+            max_tokens=4000, timeout=60,
+        )
+        content = resp.choices[0].message.content
+    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
+        cards = _salvage_objects(content)   # brace-scan any [{…}] (works regardless of the wrapper key)
+    prose = {c.get("id"): c for c in cards if c.get("id")}
+    # Coverage telemetry — a silent partial/empty cards stage is exactly what shipped generic reads.
+    if len(prose) < len(scored):
+        logger.warning("cards stage covered %d/%d jobs — %d will use evidence-based fallback prose",
+                       len(prose), len(scored), len(scored) - len(prose))
+    return prose
+
+
+def curate(prefs: dict, ctx: dict | None, sifted_jobs: list[dict]) -> list[dict] | None:
+    """Two-stage curation: Opus SCORES the rubric (integrity) → Haiku WRITES the cards (cheap prose).
+    `None` only on a Stage-1 failure (caller falls back to the sift + templated cards)."""
+    s = get_settings()
+    if not (curate_enabled() and gateway_client() and sifted_jobs):
+        return None
+    scored = _score_stage(prefs, ctx, sifted_jobs, s)
+    if scored is None:                 # hard Opus failure → sift fallback
+        return None
     by_id = {j["id"]: j for j in sifted_jobs}
+    prose = _cards_stage(prefs, ctx, scored, by_id, s)   # {} on failure → templated prose, scores intact
     out: list[dict] = []
-    for k in data.get("kept") or []:
-        job = by_id.get(k.get("id"))
+    for c in scored:
+        job = by_id.get(c["id"])
         if not job:
             continue
-        try:
-            m = _to_match(k)
-        except Exception:  # noqa: BLE001 — skip a malformed card, keep the rest
+        m = _to_match(c, prose.get(c["id"]))
+        if m.score < s.MATCH_FLOOR:    # belt-and-suspenders (Stage 1 already floored)
             continue
         job["match"] = m.as_dict()
         job["matchScore"] = m.score
         out.append(job)
-    return out or None
+    # Opus succeeded (None only on hard failure) → honor its curation even if empty ("no strong matches"),
+    # never fall back to UNFILTERED white-box cards that could surface a sub-floor match.
+    return out

app/engine/embed.py

@@ -10,25 +10,64 @@ 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).
     d = job.get("details") or {}
     parts = [
         job.get("title", ""),
         job.get("organization", ""),
-        " ".join(d.get("skills") or []),
-        (d.get("description") or "")[:600],
+        d.get("role_category") or "",
+        d.get("industry") or "",
+        " ".join(d.get("skills") or job.get("required_skills") or []),
+        (d.get("description") or "")[:1000],   # JD signal is front-loaded; 1k keeps embeds lean + cheap
     ]
     return " · ".join(p for p in parts if p)
 
 
 def _profile_text(prefs: dict, ctx: dict | None) -> str:
     ctx = ctx or {}
+    # Embed the candidate richly too — current_role + experience summary + seniority, not just title+skills.
+    exp = ctx.get("experience_summary") or ctx.get("summary") or ""
     parts = [
         prefs.get("title", ""),
         " ".join(prefs.get("role") or []),
-        " ".join(ctx.get("skills") or []),
         ctx.get("current_role", "") or "",
+        " ".join(ctx.get("skills") or []),
         " ".join(prefs.get("industry") or []),
+        " ".join(prefs.get("experience") or []),
+        str(exp)[:1200],
     ]
     return " · ".join(p for p in parts if p)
 
@@ -43,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/normalize.py

@@ -106,6 +106,25 @@ def _salary_from_amount(amount: int | None, per_year: bool) -> tuple[float | Non
     return lpa, f"₹{lpa:g}L/yr"
 
 
+_TITLE_SENIORITY = [
+    (r"\b(intern|trainee|fresher|graduate|entry[- ]?level|jr|junior)\b", "junior"),
+    (r"\b(associate)\b", "mid"),
+    (r"\b(senior|sr|staff|manager|specialist)\b", "senior"),
+    (r"\b(lead|principal|head|director|vp|chief|cxo|founding)\b", "lead"),
+]
+
+
+def _seniority_from_title(title: str | None):
+    """Infer seniority from a job title when the board omits it — so the experience factor + Opus brief
+    aren't blind. Conservative: only clear signal words; ambiguous titles stay None."""
+    t = (title or "").lower()
+    hit = None
+    for pat, sen in _TITLE_SENIORITY:   # last match wins → "Lead"/"Director" beats "Senior"
+        if re.search(pat, t):
+            hit = sen
+    return hit
+
+
 def _seniority_from_years(min_yrs: int | None):
     if min_yrs is None:
         return None
@@ -180,43 +199,79 @@ def naukri_to_scoutjob(it: dict, board: str = "Naukri") -> dict | None:
     }
 
 
-def indeed_to_scoutjob(it: dict, board: str = "Indeed") -> dict | None:
-    if str(it.get("isExpired", "False")).lower() == "true":
+# Indeed (valig actor) leaves location.city empty — only lat/long. Map to the nearest Indian metro so
+# the card + the location factor have a city (the board already filtered to the searched city anyway).
+_INDIA_METROS = {
+    "Mumbai": (19.076, 72.877), "Delhi": (28.61, 77.21), "Bengaluru": (12.97, 77.59),
+    "Hyderabad": (17.385, 78.486), "Chennai": (13.083, 80.27), "Kolkata": (22.57, 88.36),
+    "Pune": (18.52, 73.856), "Ahmedabad": (23.03, 72.58), "Gurugram": (28.46, 77.03),
+    "Noida": (28.535, 77.39), "Jaipur": (26.91, 75.79), "Chandigarh": (30.73, 76.78),
+    "Kochi": (9.93, 76.27), "Indore": (22.72, 75.86), "Coimbatore": (11.02, 76.96),
+}
+
+
+def _city_from_latlon(lat, lon) -> str | None:
+    try:
+        lat, lon = float(lat), float(lon)
+    except (TypeError, ValueError):
         return None
-    title = (it.get("positionName") or "").strip()
-    org = (it.get("company") or "").strip()
+    best, bestd = None, 9e9
+    for name, (clat, clon) in _INDIA_METROS.items():
+        d = (lat - clat) ** 2 + (lon - clon) ** 2
+        if d < bestd:
+            bestd, best = d, name
+    return best if bestd < 1.5 else None   # ~130km radius; farther → leave unknown, don't mislabel
+
+
+def indeed_to_scoutjob(it: dict, board: str = "Indeed") -> dict | None:
+    # valig~indeed-jobs-scraper schema (7s vs misceres's 52s): title, employer.name, description.text,
+    # location{city/lat/long}, jobUrl (offsite employer site), url (Indeed listing), key, baseSalary.
+    if it.get("expired"):
+        return None
+    title = (it.get("title") or "").strip()
+    org = ((it.get("employer") or {}).get("name") or "").strip()
     if not title or not org:
         return None
-    # Indeed location is "Area, City, State" (or "City, State") — the city is the 2nd-to-last part.
-    parts = [p.strip() for p in (it.get("location") or "").split(",") if p.strip()]
-    city = (parts[-2] if len(parts) >= 2 else (parts[0] if parts else None))
-    types = []
-    try:
-        types = [t.strip() for t in eval(it.get("jobType") or "[]")] if it.get("jobType") else []
-    except Exception:  # noqa: BLE001
-        pass
-    sal = it.get("salary") or ""
-    lpa, pay = _salary_from_amount(_digits(sal), per_year="year" in sal.lower()) if sal else (None, "Not disclosed")
-    ext = it.get("externalApplyLink")
-    offsite = bool(ext and ext != "None")          # externalApplyLink = real offsite employer link
-    apply_url = ext if offsite else it.get("url")  # offsite first, Indeed listing fallback
+    loc = it.get("location") or {}
+    city = (loc.get("city") or "").strip() or _city_from_latlon(loc.get("latitude"), loc.get("longitude"))
+    country = loc.get("countryName") or "India"
+    job_url = it.get("jobUrl") or ""
+    offsite = bool(job_url and "indeed.com" not in job_url)   # jobUrl = real offsite employer link
+    apply_url = job_url if offsite else it.get("url")          # offsite first, Indeed listing fallback
+    bs = it.get("baseSalary") or {}
+    lpa, pay = None, "Not disclosed"
+    amt = bs.get("max") or bs.get("min")
+    if amt and bs.get("currencyCode") in (None, "INR"):
+        mult = {"YEAR": 1, "MONTH": 12, "WEEK": 52, "DAY": 260, "HOUR": 2080}.get(bs.get("unitOfWork"), 1)
+        lpa = round(amt * mult / 100000, 1)
+        pay = f"₹{lpa:g} LPA"
+    etypes = [str(v) for v in (it.get("jobTypes") or {}).values()]
+    occ = [str(v) for v in (it.get("occupations") or {}).values()]
+    mode = "remote" if any("remote" in t.lower() for t in etypes) else None
+    details = _compact({
+        "description": _plain((it.get("description") or {}).get("text")),
+        "employment_type": ", ".join(etypes) or None,
+        "role_category": occ[0] if occ else None,
+    })
     return {
-        "id": f"indeed-{it.get('id')}",
+        "id": f"indeed-{it.get('key')}",
         "title": title,
         "organization": org,
-        "logo": _logo(org, _logo_url(it)),
-        "posted_date": _posted_date(it),
+        "logo": _logo(org),
+        "posted_date": (it.get("datePublished") or it.get("dateOnIndeed") or None),
         "location_city": city,
-        "location_country": "India",
+        "location_country": country,
+        "location_mode": mode,
         "matchScore": 0,
         "salary_lpa": lpa,
         "payLabel": pay,
-        "required_skills": types,
-        "tags": types[:3],
+        "required_skills": [],          # Indeed exposes no skill list; description carries the signal
+        "tags": etypes[:3],
         "applicants": None,
         "note": f"Live from {board}",
         "apply_url": apply_url,
         "offsite_apply": offsite,
+        "details": details,
         "qx": 10,
     }
 
@@ -415,12 +470,236 @@ def foundit_to_scoutjob(it: dict, board: str = "Foundit") -> dict | None:
     }
 
 
+# LinkedIn seniorityLevel → our band (often "Not Applicable" for India → None, then title-inferred centrally).
+_LI_SENIORITY = {"internship": "junior", "entry level": "junior", "associate": "mid",
+                 "mid-senior level": "senior", "director": "lead", "executive": "lead"}
+
+
+def linkedin_cc_to_scoutjob(it: dict, board: str = "LinkedIn") -> dict | None:
+    """curious_coder~linkedin-jobs-scraper — structured INDUSTRY + jobFunction (role) + applicantsCount.
+    (seniorityLevel is present but often 'Not Applicable' for India; we map it when real, else infer.)"""
+    title = (it.get("title") or "").strip()
+    org = (it.get("companyName") or "").strip()
+    if not title or not org:
+        return None
+    # location is "City, State, Country" — first part is the city, last is the country.
+    parts = [p.strip() for p in str(it.get("location") or "").split(",") if p.strip()]
+    city = parts[0] if parts else None
+    country = parts[-1] if len(parts) >= 2 else "India"
+    sen = _LI_SENIORITY.get(str(it.get("seniorityLevel") or "").strip().lower())   # None if "Not Applicable"
+    inds = [s.strip() for s in str(it.get("industries") or "").split(",") if s.strip()]
+    fn = (it.get("jobFunction") or "").strip() or None
+    etype = (it.get("employmentType") or "").strip()
+    mode = "remote" if "remote" in etype.lower() else None
+    ext = it.get("applyUrl")
+    offsite = bool(ext and "linkedin.com" not in ext)        # applyUrl (when present) = offsite employer
+    apply_url = ext if offsite else it.get("link")           # else the LinkedIn job page
+    details = _compact({
+        "description": _plain(it.get("descriptionText") or it.get("descriptionHtml")),
+        "industry": inds[0] if inds else None,
+        "role_category": fn,
+        "employment_type": etype or None,
+    })
+    return {
+        "id": f"linkedin-{it.get('id')}",
+        "title": title,
+        "organization": org,
+        "logo": _logo(org, it.get("companyLogo")),
+        "posted_date": it.get("postedAt"),
+        "location_city": city,
+        "location_country": country,
+        "location_mode": mode,
+        "matchScore": 0,
+        "salary_lpa": None,
+        "payLabel": "Not disclosed",
+        "seniority_level": sen,
+        "required_skills": [],          # LinkedIn exposes no skill tags; description + jobFunction carry it
+        "industry": inds[0] if inds else None,
+        "tags": [fn] if fn else [],
+        "applicants": it.get("applicantsCount"),
+        "note": f"Live from {board}",
+        "apply_url": apply_url,
+        "offsite_apply": offsite,
+        "details": details,
+        "qx": 10,
+    }
+
+
+def timesjobs_to_scoutjob(it: dict, board: str = "TimesJobs") -> dict | None:
+    """shahidirfan/Timesjobs-Scraper — the richest India board: skills + salary + experience +
+    description + an OFFSITE employer link (external_job_url). Clean structured fields."""
+    title = (it.get("title") or "").strip()
+    title = re.sub(r"^title:\s*", "", title, flags=re.I).strip()   # some rows carry a stray "Title:" prefix
+    org = (it.get("company") or "").strip()
+    if not title or not org:
+        return None
+    if org.islower():
+        org = org.title()
+    # location is "City" or "City, City2" — take the first concrete city.
+    city = next((p.strip() for p in str(it.get("location") or "").split(",") if p.strip()), None)
+    skills = it.get("skills") or []
+    if isinstance(skills, str):
+        skills = [s.strip() for s in skills.split(",") if s.strip()]
+    skills = [str(s) for s in skills if s]
+    # experience "6 - 10 Yrs" → seniority from the LOW end of the band.
+    seniority = None
+    m = re.search(r"\d+", str(it.get("experience") or ""))
+    if m:
+        seniority = _seniority_from_years(int(m.group()))
+    # salary is usually the text "As per industry standard"; surface a ₹L band only if a real number is there.
+    sal_raw = str(it.get("salary") or "")
+    lpa, pay = (None, sal_raw or "Not disclosed")
+    if re.search(r"\d", sal_raw):
+        lpa, pay = _salary_from_amount(_digits(sal_raw), per_year=True)
+    types = [t.strip() for t in str(it.get("job_type") or "").split(",") if t.strip()]
+    mode = "remote" if any("remote" in t.lower() for t in types) else None
+    ext = it.get("external_job_url")
+    offsite = bool(ext and "timesjobs.com" not in (ext or ""))   # external_job_url = real employer site
+    apply_url = ext if offsite else it.get("url")
+    details = _compact({
+        "description": _plain(it.get("description_text") or it.get("description_html")),
+        "experience": it.get("experience"),
+        "employment_type": ", ".join(types) or None,
+        "company_about": _plain(it.get("company_description"), limit=400),
+        "skills": skills,
+    })
+    return {
+        "id": f"timesjobs-{it.get('job_id')}",
+        "title": title,
+        "organization": org,
+        "logo": _logo(org),
+        "posted_date": it.get("date_posted"),
+        "location_city": city,
+        "location_country": "India",
+        "location_mode": mode,
+        "matchScore": 0,
+        "salary_lpa": lpa,
+        "payLabel": pay,
+        "seniority_level": seniority,
+        "required_skills": skills[:8],
+        "tags": skills[:3],
+        "applicants": None,
+        "note": f"Live from {board}",
+        "apply_url": apply_url,
+        "offsite_apply": offsite,
+        "details": details,
+        "qx": 10,
+    }
+
+
+def workindia_to_scoutjob(it: dict, board: str = "WorkIndia") -> dict | None:
+    """shahidirfan/workindia-jobs-scraper (includeDetails=True) — India blue/grey-collar non-tech.
+    Verified live (Phase 0): returns real `skills` + `profile_job_description`. Apply is on-platform."""
+    title = (it.get("profile_job_title") or "").strip()
+    org = (it.get("branch_company_name") or "").strip()
+    if not title or not org:
+        return None
+    city = (str(it.get("branch_location_city_name") or "").strip() or None)
+    skills = it.get("skills") or []
+    if isinstance(skills, str):
+        skills = [s.strip() for s in skills.split(",") if s.strip()]
+    skills = [str(s) for s in skills if s]
+    seniority = None
+    m = re.search(r"\d+", str(it.get("job_experience") or it.get("experience") or ""))
+    if m:
+        seniority = _seniority_from_years(int(m.group()))
+    # Blue/grey-collar pay is usually monthly ₹ and noisy — show the structure text, don't risk the ₹L band.
+    details = _compact({
+        "description": _plain(it.get("profile_job_description") or it.get("profile_html_description")
+                              or it.get("profile_short_description")),
+        "experience": it.get("job_experience") or it.get("experience"),
+        "employment_type": it.get("employment_type"),
+        "education": it.get("degree") or it.get("profile_qualification_required"),
+        "industry": it.get("profile_industry_display_name"),
+        "skills": skills,
+    })
+    return {
+        "id": f"workindia-{it.get('job_id') or it.get('id')}",
+        "title": title.title() if title.isupper() else title,
+        "organization": org,
+        "logo": _logo(org, _logo_url(it)),
+        "posted_date": it.get("job_published_on_platform") or it.get("created_at"),
+        "location_city": city,
+        "location_country": "India",
+        "matchScore": 0,
+        "salary_lpa": None,
+        "payLabel": it.get("profile_salary_structure") or "Not disclosed",
+        "seniority_level": seniority,
+        "required_skills": skills[:6],
+        "tags": skills[:3],
+        "applicants": None,
+        "note": f"Live from {board}",
+        # source_url is a GENERIC listing page (same for every job) → would dedup-collapse the deck.
+        # Build a per-job URL from job_id so each is unique AND clickable.
+        "apply_url": (f"https://www.workindia.in/jobs/{it.get('job_id')}/"
+                      if it.get("job_id") else it.get("source_url")),
+        "offsite_apply": False,   # WorkIndia is a platform — apply on WorkIndia
+        "details": details,
+        "qx": 10,
+    }
+
+
+def wellfound_to_scoutjob(it: dict, board: str = "Wellfound") -> dict | None:
+    """blackfalcondata/wellfound-scraper (enrichDetail=True) — startup roles with skills, equity,
+    funding stage. Apply is on-platform (Wellfound), so apply_url = the listing (offsite_apply=False)."""
+    title = (it.get("title") or "").strip()
+    org = (it.get("companyName") or "").strip()
+    if not title or not org:
+        return None
+    locs = it.get("locationNames") or []
+    city = (str(locs[0]).split(",")[0].strip() or None) if isinstance(locs, list) and locs else None
+    skills = [str(s) for s in (it.get("skills") or []) if s]
+    # Salary → LPA only when the listing is in INR (don't pollute the India band with USD); else the
+    # raw compensation string is the honest label.
+    lpa, pay = None, "Not disclosed"
+    cur = (it.get("salaryCurrency") or "").upper()
+    if it.get("salaryMax") and cur == "INR":
+        lpa, pay = _salary_from_amount(int(it["salaryMax"]), per_year=True)
+    elif it.get("compensation"):
+        pay = str(it["compensation"])
+    elif it.get("salaryMin") or it.get("salaryMax"):
+        pay = f"{cur or '$'} {it.get('salaryMin') or ''}–{it.get('salaryMax') or ''}".strip()
+    details = _compact({
+        "description": _plain(it.get("description")),
+        "employment_type": (it.get("jobType") or "").replace("_", " ").title() or None,
+        "skills": skills,
+        "funding_stage": it.get("companyFundingStage"),
+        "equity": it.get("equityMin") or it.get("equity"),
+        "visa": it.get("visaSponsorship"),
+    })
+    return {
+        "id": f"wellfound-{it.get('id')}",
+        "title": title,
+        "organization": org,
+        "logo": _logo(org, _logo_url(it)),
+        "posted_date": _posted_date(it),
+        "location_city": city,
+        "location_country": None,
+        "location_mode": "remote" if it.get("remote") else None,
+        "matchScore": 0,
+        "salary_lpa": lpa,
+        "payLabel": pay,
+        "required_skills": skills[:6],
+        "tags": skills[:3],
+        "applicants": None,
+        "note": f"Live from {board}",
+        "apply_url": it.get("portalUrl"),
+        "offsite_apply": False,   # apply happens on Wellfound (on-platform), not an offsite employer redirect
+        "details": details,
+        "qx": 10,
+    }
+
+
 # board key → normalizer
 NORMALIZERS = {
     "naukri": naukri_feed_to_scoutjob,   # blackfalcondata (primary)
     "foundit": foundit_to_scoutjob,
-    "linkedin": linkedin_to_scoutjob,
+    "linkedin": linkedin_cc_to_scoutjob,   # curious_coder (structured industry+jobFunction+applicants)
+    "linkedin_v1": linkedin_to_scoutjob,   # harvestapi (fallback)
     "naukri_v1": naukri_to_scoutjob,     # muhammetakkurtt (fallback)
     "indeed": indeed_to_scoutjob,
     "ats": ats_to_scoutjob,
+    "wellfound": wellfound_to_scoutjob,    # blackfalcondata (startup roles)
+    "workindia": workindia_to_scoutjob,    # shahidirfan (India blue/grey-collar non-tech)
+    "timesjobs": timesjobs_to_scoutjob,    # shahidirfan (India — full signal: skills+salary+experience)
 }

app/engine/rank.py

@@ -53,7 +53,7 @@ def profile_from(prefs: dict, user_context: dict | None = None) -> UserProfile:
     countries = [lp["country"].lower() for lp in locs if lp.get("country")]
     wm = [w.lower() for w in (prefs.get("workMode") or [])]
     title = C.title_of(prefs) or None
-    skills = S.canon((user_context or {}).get("skills"))  # empty in Phase 1 (prefs-only)
+    skills = S.canon((user_context or {}).get("skills"))  # absent for non-tech → semantic (cosine) carries it
     toks = {t for t in re.findall(r"[a-z0-9+#]+", f"{title or ''} {' '.join(skills)}".lower()) if len(t) > 2}
     return UserProfile(
         title=title, skills=skills, years=C.years_of(prefs),
@@ -130,6 +130,10 @@ def f_role(p, job): return _cat_match(p.roles, _jrole(job))
 
 
 def f_semantic(p, job):
+    # Real embedding cosine when the sift threaded it on (the true semantic signal); else token overlap.
+    c = job.get("_vibe_cosine")
+    if c is not None:
+        return float(c)
     if not p.profile_tokens:
         return None
     blob = f"{job.get('title') or ''} {(job.get('details') or {}).get('description') or ''}".lower()
@@ -142,11 +146,29 @@ def f_semantic(p, job):
 FACTORS = {"skill": f_skill, "experience": f_experience, "location": f_location, "salary": f_salary,
            "industry": f_industry, "role": f_role, "semantic": f_semantic}
 
+# Non-tech identity = role + industry + meaning, NOT a skill checklist (which boards rarely list).
+# When the profile reads non-tech, tilt the weights there; skill drops, semantic/role/industry rise.
+NON_TECH_FAMILIES = {"sales", "marketing", "finance", "accounting", "hr", "human resource", "operations",
+                     "ops", "customer support", "customer service", "bpo", "legal", "supply chain", "admin",
+                     "business development"}
+NONTECH_WEIGHTS = {"skill": 0.18, "experience": 0.22, "location": 0.14, "salary": 0.06,
+                   "industry": 0.12, "role": 0.16, "semantic": 0.12}
+
+
+def _is_nontech(p: UserProfile) -> bool:
+    blob = " ".join([*(r.lower() for r in p.roles), (p.title or "").lower()])
+    return any(fam in blob for fam in NON_TECH_FAMILIES)
+
+
+def get_weights(p: UserProfile) -> dict:
+    return NONTECH_WEIGHTS if _is_nontech(p) else WEIGHTS
+
 
 def utility(p: UserProfile, job: dict):
+    w = get_weights(p)
     factors = {k: fn(p, job) for k, fn in FACTORS.items()}
-    num = sum(WEIGHTS[k] * v for k, v in factors.items() if v is not None)
-    den = sum(WEIGHTS[k] for k, v in factors.items() if v is not None)
+    num = sum(w[k] * v for k, v in factors.items() if v is not None)
+    den = sum(w[k] for k, v in factors.items() if v is not None)
     U = (num / den) if den else None           # NO present signals → None (filtered out)
     return U, factors
 

app/engine/rubric.py

@@ -0,0 +1,66 @@
+"""The match rubric — the ONE published source of truth for how a fit score is computed.
+
+Integrity rule: the displayed score is NEVER emitted by a model. The curator (Opus) assesses each
+DIMENSION 0-100 against a fixed ANCHOR (observable facts, not vibes); `aggregate()` combines them with
+FIXED weights into the overall score. So a 94 is *earned* — role 100, location 100, skills 88, … — and
+every score decomposes + audits. There is no prompt knob that lifts the number without the evidence.
+"""
+from __future__ import annotations
+
+# key · label (shown in the breakdown) · weight · the ANCHOR the curator scores against. Weights sum to 1.
+DIMENSIONS = [
+    {"key": "role", "label": "Role fit", "weight": 0.25,
+     "anchor": "100 = same function · 75 = adjacent function · 45 = different function · 15 = unrelated"},
+    {"key": "skills", "label": "Skills fit", "weight": 0.25,
+     "anchor": "share of the job's stated requirements the candidate genuinely meets (100 = all · 50 = half · 0 = none)"},
+    {"key": "seniority", "label": "Seniority fit", "weight": 0.15,
+     "anchor": "100 = same level · 70 = one band off · 35 = two+ bands off"},
+    {"key": "location", "label": "Location fit", "weight": 0.15,
+     "anchor": "100 = target city or remote-ok · 60 = same metro/region · 25 = different city"},
+    {"key": "industry", "label": "Industry fit", "weight": 0.10,
+     "anchor": "100 = same domain · 65 = adjacent domain · 30 = unrelated domain (omit if no industry preference)"},
+    {"key": "experience", "label": "Experience fit", "weight": 0.10,
+     "anchor": "100 = years + trajectory align · 60 = slightly under/over · 30 = large gap"},
+]
+
+KEYS = [d["key"] for d in DIMENSIONS]
+WEIGHTS = {d["key"]: d["weight"] for d in DIMENSIONS}
+LABELS = {d["key"]: d["label"] for d in DIMENSIONS}
+
+
+def aggregate(scores: dict) -> int:
+    """{dimension_key: 0-100} → the weighted overall (0-100). Only dimensions actually scored count; a
+    genuinely-absent one (e.g. industry with no preference) drops out and its weight redistributes over
+    the rest — floor-free: never a silent zero that tanks the score, never a constant that props it up.
+    The five core dimensions (role/skills/seniority/location/experience) always apply, so this can't be
+    gamed by omitting the hard ones — the curator is required to score every applicable dimension."""
+    present = {k: max(0.0, min(100.0, float(scores[k]))) for k in KEYS if scores.get(k) is not None}
+    if not present:
+        return 0
+    wsum = sum(WEIGHTS[k] for k in present)
+    return round(sum(WEIGHTS[k] * v for k, v in present.items()) / wsum)
+
+
+def calibrate(raw: float, *, floor: float = 50.0, gamma: float = 1.3) -> int:
+    """Transparent presentation calibration of the HEADLINE score (the per-dimension breakdown stays raw
+    evidence). A match% is a calibrated product judgment, not a physical measurement — so we apply ONE
+    documented curve, anchored at the floor and gently expanded at the top, so a genuinely-strong match
+    presents in the 90s. Guardrails that keep it honest:
+      • MONOTONIC — strictly increasing, so a better job ALWAYS scores higher (ordering never lies).
+      • FLOOR-ANCHORED — a weak match at the floor is unchanged; weak NEVER becomes strong.
+      • UNIFORM — the same curve for every job (no cherry-picking).
+    Example (gamma 1.3): 50→50, 60→63, 70→74, 80→85, 90→94, 95→97. Raw stays as-is below the floor."""
+    if raw <= floor:
+        return round(raw)
+    span = 100.0 - floor
+    return round(100.0 - span * ((100.0 - raw) / span) ** gamma)
+
+
+def level_for(score: float) -> str:
+    """Strong / Solid / Light band for a dimension (drives the breakdown chips)."""
+    return "Strong" if score >= 72 else ("Solid" if score >= 52 else "Light")
+
+
+def prompt_block() -> str:
+    """The rubric rendered for the curator prompt: the exact dimensions to score, with their anchors."""
+    return "\n".join(f'    - "{d["key"]}": {d["label"]} — {d["anchor"]}' for d in DIMENSIONS)

app/engine/search.py

@@ -21,10 +21,34 @@ def _ats_companies() -> list[dict]:
     return [{"company": c.strip()} for c in get_settings().ATS_COMPANIES.split(",") if c.strip()]
 
 
+def pool_decision(n_fresh: int, force_fresh: bool, floor: int, reorder: int,
+                  exhausted: bool = False) -> str:
+    """The warm-pool gate. Given how many FRESH, UNSEEN jobs are banked, decide where candidates come from:
+      'sweep'       — blocking Apify fetch (pool below the hard floor, or a forced fresh search)
+      'pool+refill' — serve from the pool NOW + top it up in the BACKGROUND (dipped below the reorder mark)
+      'pool'        — serve from the pool, no fetch at all (pool healthy)
+    Reorder EARLY and floor HIGH so the pool never runs thin (floor ≤ reorder).
+
+    EXHAUSTION guard: when the query is tapped out (a recent refill added almost no NEW jobs), a blocking
+    sweep would just re-buy duplicates — so as long as the pool isn't EMPTY, serve what's there instead of
+    block-fetching. This is what makes a high floor safe on niche queries. A forced-fresh always sweeps."""
+    if force_fresh:
+        return "sweep"
+    if exhausted and n_fresh > 0:
+        return "pool"                       # tapped out but we have something → serve, don't re-buy dupes
+    if n_fresh < floor:
+        return "sweep"
+    if n_fresh < reorder:
+        return "pool+refill"
+    return "pool"
+
+
 # Boards whose actor takes a real page/offset → they get a per-board cursor that advances each run.
 # Everyone else (Naukri/Foundit feeds — no page param) relies purely on the per-user seen-net; we do
 # NOT track a meaningless cursor for them.
-PAGINATING_BOARDS = {"linkedin", "indeed"}
+# NO board paginates: curious_coder only honors start=0 (start>0 → empty), and the rest are date feeds.
+# Freshness comes from the boards' "latest by date" + the per-user seen-net — NOT page cursors.
+PAGINATING_BOARDS: set[str] = set()
 
 # board key → (actor_id, build_input, normalizer). "Balanced" India stack = naukri+foundit+linkedin,
 # all city-filtered at the board. The rest stay registered (code ready) but off unless enabled.
@@ -36,11 +60,13 @@ BOARDS = {
     "foundit": ("shahidirfan~Foundit-Jobs-Scraper",
                 lambda p: A.build_foundit_input(p, results_wanted=get_settings().FOUNDIT_MAX_JOBS),
                 N.foundit_to_scoutjob),
-    "linkedin": ("harvestapi~linkedin-job-search",
-                 # cursor stores the LAST page fetched (0 = none yet) → next page = last + 1.
-                 lambda p: A.build_linkedin_input(p, recall=True, max_items=get_settings().LINKEDIN_MAX_JOBS,
-                                                  page=int((p.get("_cursors") or {}).get("linkedin", 0)) + 1),
-                 N.linkedin_to_scoutjob),
+    "linkedin": ("curious_coder~linkedin-jobs-scraper",   # structured industry+jobFunction+applicants
+                 # NO cursor — actor only honors start=0; freshness via date sort + the seen-net.
+                 lambda p: A.build_linkedin_cc_input(p, count=get_settings().LINKEDIN_MAX_JOBS),
+                 N.linkedin_cc_to_scoutjob),
+    "linkedin_v1": ("harvestapi~linkedin-job-search",   # registered fallback (jobTitles input)
+                    lambda p: A.build_linkedin_input(p, recall=True, max_items=get_settings().LINKEDIN_MAX_JOBS),
+                    N.linkedin_to_scoutjob),
     # ── registered fallbacks / other lanes (off unless added to BOARDS_ENABLED) ──
     "ats": ("bovi~greenhouse-lever-ashby-job-scraper",
             lambda p: A.build_ats_input(p, recall=True, companies=_ats_companies(),
@@ -49,16 +75,40 @@ BOARDS = {
     "naukri_v1": ("muhammetakkurtt~naukri-job-scraper",
                   lambda p: A.build_naukri_input(p, recall=True, max_jobs=get_settings().NAUKRI_MAX_JOBS),
                   N.naukri_to_scoutjob),
-    "indeed": ("misceres~indeed-scraper",
+    "indeed": ("valig~indeed-jobs-scraper",   # ~7s vs misceres ~52s; richer fields + offsite jobUrl
                lambda p: A.build_indeed_input(p, max_items=get_settings().INDEED_MAX_JOBS),
                N.indeed_to_scoutjob),
+    "wellfound": ("blackfalcondata~wellfound-scraper",  # registered, OFF — US-startup-heavy, not the India fit
+                  lambda p: A.build_wellfound_input(p, max_results=get_settings().WELLFOUND_MAX_JOBS),
+                  N.wellfound_to_scoutjob),
+    "workindia": ("shahidirfan~workindia-jobs-scraper",
+                  # India blue/grey-collar non-tech; no page param → per-user seen-net dedups.
+                  lambda p: A.build_workindia_input(p, results_wanted=get_settings().WORKINDIA_MAX_JOBS),
+                  N.workindia_to_scoutjob),
+    "timesjobs": ("shahidirfan~Timesjobs-Scraper",   # India full-signal; no page param → seen-net dedups.
+                  lambda p: A.build_timesjobs_input(p, results_wanted=get_settings().TIMESJOBS_MAX_JOBS),
+                  N.timesjobs_to_scoutjob),
 }
 
 
 async def _fetch_board(key: str, prefs: dict, cache_mode: str | None = None):
     actor, build, norm = BOARDS[key]
-    items = await run_actor(actor, build(prefs), cache_mode=cache_mode)
-    jobs = [sj for it in items if (sj := norm(it))]
+    inp = build(prefs)
+    items = await run_actor(actor, inp, cache_mode=cache_mode)
+    if not items:
+        # Scrape-based boards (Naukri-feed, curious_coder LinkedIn) intermittently return 0 under the
+        # site's anti-scrape rate-limiting, then recover. One fresh retry rescues the transient case
+        # (a genuinely-empty query just returns 0 again — cheap). RCA'd: same input flakes 0 then 15.
+        logger.info("board %s returned 0 — retrying once (transient scrape flake)", key)
+        items = await run_actor(actor, inp, cache_mode="refresh")
+    jobs = []
+    for it in items:
+        sj = norm(it)
+        if not sj:
+            continue
+        if not sj.get("seniority_level"):                 # board omitted it → infer from the title
+            sj["seniority_level"] = N._seniority_from_title(sj.get("title"))
+        jobs.append(sj)
     return key, jobs
 
 
@@ -67,13 +117,25 @@ async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict:
     # 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]
-    results = await asyncio.gather(*(_fetch_board(k, prefs, cache_mode) for k in enabled), return_exceptions=True)
+    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)
+        except asyncio.TimeoutError:
+            logger.warning("board %s exceeded %ss — skipped (search stays fast)", k, cap)
+            return k, []
+
+    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
@@ -86,7 +148,13 @@ 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": bool(enabled) and failed_boards == len(enabled),
+        "failed_boards": failed_boards,
+        "enabled_boards": len(enabled),
+    }
 
 
 def has_mvq(prefs: dict) -> bool:

app/engine/sift.py

@@ -11,32 +11,49 @@ from app.config import get_settings
 from app.engine import embed as _embed
 from app.engine import rank as _rank
 
-W_WHITEBOX, W_VIBE = 0.55, 0.45   # white-box (checklist) leads slightly; vibe (semantics) sharpens
+W_WHITEBOX, W_VIBE = 0.4, 0.6   # embedding (the truer semantic signal) now LEADS the noisy checklist
 
 
-def _ranks(score_by_id: dict[str, float]) -> dict[str, float]:
-    """id → normalized rank in [0,1] (1 = best, 0 = worst). Position-based, so scale-free."""
-    order = sorted(score_by_id, key=lambda i: -score_by_id[i])
-    n = len(order)
-    return {i: (1.0 - pos / (n - 1)) if n > 1 else 1.0 for pos, i in enumerate(order)}
+def _minmax(score_by_id: dict[str, float]) -> dict[str, float]:
+    """id → score min-max-normalized into [0,1]. PRESERVES MAGNITUDE (a much-better job stays much
+    better) — unlike rank-position fusion, which flattened cosine 0.95 and 0.72 to the same gap."""
+    vals = score_by_id.values()
+    lo, hi = min(vals), max(vals)
+    rng = hi - lo
+    return {i: ((s - lo) / rng if rng else 1.0) for i, s in score_by_id.items()}
 
 
 def sift(prefs: dict, ctx: dict | None, jobs: list[dict], k: int | None = None):
-    """Returns (top_k_jobs, debug). top_k_jobs are ranked best-first; each keeps its white-box match."""
+    """Returns (top_k_jobs, debug). top_k_jobs are ranked best-first; each keeps its white-box match.
+
+    Order matters: embed FIRST so the white-box `f_semantic` can reuse the real cosine (via the
+    threaded `_vibe_cosine`) instead of crude token overlap. Then fuse by magnitude, embedding-led."""
     s = get_settings()
     k = k or s.SIFT_TOP_K
-    ranked = _rank.rank(prefs, ctx, jobs)["opportunities"]      # white-box scores ALL (no early cut)
-    if not ranked:
+    if not jobs:
         return [], {"mode": "empty", "scored": 0, "k": 0}
 
-    wb = {j["id"]: j["matchScore"] / 100.0 for j in ranked}
-    vibe = _embed.vibe_scores(prefs, ctx, ranked)              # id → cosine, or None
+    # 1. embedding cosine FIRST — thread onto each job so rank.f_semantic reads the real signal.
+    vibe = _embed.vibe_scores(prefs, ctx, jobs)               # id → cosine, or None
     if vibe:
-        wr, vr = _ranks(wb), _ranks(vibe)
-        fused = {i: W_WHITEBOX * wr[i] + W_VIBE * vr.get(i, 0.0) for i in wb}
-        mode = "whitebox+vibe"
+        for j in jobs:
+            if j["id"] in vibe:
+                j["_vibe_cosine"] = vibe[j["id"]]
+
+    # 2. white-box scores ALL (now semantically-aware where the cosine was threaded).
+    ranked = _rank.rank(prefs, ctx, jobs)["opportunities"]
+    if not ranked:
+        return [], {"mode": "empty", "scored": 0, "k": 0}
+    wb = {j["id"]: j["matchScore"] / 100.0 for j in ranked}
+
+    # 3. magnitude-preserving, embedding-led fusion.
+    if vibe:
+        wbn = _minmax(wb)
+        vbn = _minmax({i: vibe[i] for i in wb if i in vibe})
+        fused = {i: W_WHITEBOX * wbn[i] + W_VIBE * vbn.get(i, 0.0) for i in wb}
+        mode = "vibe-led(magnitude)"
     else:
-        fused = _ranks(wb)
+        fused = wb
         mode = "whitebox-only"
 
     top = sorted(ranked, key=lambda j: -fused[j["id"]])[:k]

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.
@@ -109,6 +109,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

requirements.txt

@@ -11,3 +11,4 @@ openai>=1.40
 numpy
 sqlalchemy[asyncio]>=2.0
 asyncpg
+alembic>=1.13.0

tests/_seed_regression.py

@@ -0,0 +1,79 @@
+"""One-off seeder for the regression fixture. Runs live sweeps (tech + non-tech), freezes ~12 candidate
+jobs per case, and Opus-judges each (profile, job) for a recruiter-fit target. Writes
+tests/fixtures/regression_set.json. Re-run only to refresh the baseline set (costs Apify + Opus)."""
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+
+from app.engine import search
+from app.engine.llm import gateway_client
+from app.config import get_settings
+
+OUT = Path(__file__).parent / "fixtures" / "regression_set.json"
+N_PER_CASE = 40   # MUST exceed SIFT_TOP_K so the sift actually cuts → membership-recall is meaningful
+
+CASES = [
+    {"label": "tech: backend engineer · bangalore",
+     "prefs": {"title": "Backend Engineer", "role": ["Engineering"], "location": ["Bangalore · India"]},
+     "user_context": {"id": "reg-tech", "skills": ["Python", "Django", "PostgreSQL", "AWS", "Docker", "REST APIs"],
+                      "current_role": "Software Engineer"}},
+    {"label": "non-tech: sales manager · mumbai",
+     "prefs": {"title": "Sales Manager", "role": ["Sales"], "location": ["Mumbai · India"]},
+     "user_context": {"id": "reg-sales", "skills": ["B2B Sales", "Account Management", "Negotiation", "CRM", "Lead Generation"],
+                      "current_role": "Business Development Manager"}},
+    {"label": "non-tech sparse: operations manager · pune",
+     "prefs": {"title": "Operations Manager", "role": ["Ops"], "location": ["Pune · India"]},
+     "user_context": {"id": "reg-ops", "skills": [], "current_role": "Operations Lead"}},
+]
+
+_JUDGE = """You are a veteran Indian recruiter. Given a CANDIDATE profile and a list of JOBS, score each job
+0-100 for how good a match it is for THIS candidate (role fit + seniority + location + skills/responsibilities).
+Be realistic and calibrated: a genuinely strong fit is 80-92, a solid fit 70-80, a stretch 55-70, off-target
+<50. Respond ONLY with JSON: {"scores":{"<job_id>": <int>, ...}} using each job's exact id."""
+
+
+async def judge(client, prefs, ctx, jobs):
+    brief = {"candidate": {"target": prefs.get("title"), "role": prefs.get("role"),
+                           "location": prefs.get("location"), "skills": ctx.get("skills"),
+                           "current_role": ctx.get("current_role")},
+             "jobs": [{"id": j["id"], "title": j.get("title"), "company": j.get("organization"),
+                       "location": f"{j.get('location_city') or ''} {j.get('location_country') or ''}".strip(),
+                       "skills": (j.get("details") or {}).get("skills") or j.get("required_skills"),
+                       "snippet": ((j.get("details") or {}).get("description") or "")[:400]} for j in jobs]}
+    resp = client.chat.completions.create(
+        model=get_settings().CURATE_MODEL, max_tokens=2000,
+        messages=[{"role": "system", "content": _JUDGE}, {"role": "user", "content": json.dumps(brief, ensure_ascii=False)}])
+    txt = resp.choices[0].message.content
+    i, j = txt.find("{"), txt.rfind("}")
+    return {str(k): int(v) for k, v in json.loads(txt[i:j + 1])["scores"].items()}
+
+
+async def main():
+    client = gateway_client()
+    assert client, "no gateway client (DSPY_API_BASE/KEY) — cannot judge"
+    out_cases = []
+    for c in CASES:
+        prefs = {**c["prefs"], "_fresh": True}
+        sweep = await search.run_sweep(prefs, fresh=True)
+        import random
+        pool = list(sweep["opportunities"])
+        random.Random(42).shuffle(pool)        # board-diverse sample, deterministic
+        jobs = pool[:N_PER_CASE]
+        if not jobs:
+            print(f"  ⚠️  {c['label']}: 0 jobs swept — skipping"); continue
+        targets = await judge(client, c["prefs"], c["user_context"], jobs)
+        # keep only jobs that got a target
+        jobs = [j for j in jobs if j["id"] in targets]
+        out_cases.append({"label": c["label"], "prefs": c["prefs"], "user_context": c["user_context"],
+                          "jobs": jobs, "targets": {j["id"]: targets[j["id"]] for j in jobs}})
+        print(f"  ✅ {c['label']}: {len(jobs)} jobs judged "
+              f"(sources={sweep['sources']}, target range {min(targets.values())}-{max(targets.values())})")
+    OUT.parent.mkdir(parents=True, exist_ok=True)
+    OUT.write_text(json.dumps({"version": "phase0-baseline", "cases": out_cases}, indent=2, ensure_ascii=False))
+    print(f"  → wrote {len(out_cases)} cases, {sum(len(c['jobs']) for c in out_cases)} pairs to {OUT}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/run_regression.py

@@ -0,0 +1,93 @@
+"""Regression runner — the measurement gate for every engine-improvement phase.
+
+The fixture `tests/fixtures/regression_set.json` freezes a set of CASES, each = {prefs, user_context,
+jobs[], targets{job_id: recruiter_score}}. The jobs are captured from a real sweep so the run is
+reproducible WITHOUT re-hitting Apify (deterministic, cheap, comparable across phases).
+
+For each case this runs the REAL engine (rank → sift) over the frozen jobs and reports two metrics:
+  • SCORE fidelity — MAE / correlation of the engine's matchScore vs the human/judge target.
+  • SIFT TOP-K MEMBERSHIP RECALL — of the genuinely-best jobs (target ≥ 75), how many reach the top-K
+    pool Opus actually sees. This is the "is the funnel feeding Opus the best jobs?" metric.
+
+Usage:  python tests/run_regression.py            # prints the table for the current engine
+        python tests/run_regression.py --json      # machine-readable (to diff phases)
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+from app.engine import rank as _rank
+from app.engine import sift as _sift
+from app.engine import embed as _embed
+from app.config import get_settings
+
+FIXTURE = Path(__file__).parent / "fixtures" / "regression_set.json"
+GREAT = 75   # target ≥ this = a "genuinely good" job that SHOULD reach Opus
+
+
+def _mae(pairs):
+    return round(sum(abs(a - b) for a, b in pairs) / len(pairs), 1) if pairs else None
+
+
+def run(emit_json: bool = False):
+    data = json.loads(FIXTURE.read_text())
+    cases = data["cases"]
+    k = get_settings().SIFT_TOP_K
+    rows, all_score_pairs, recalls = [], [], []
+
+    for c in cases:
+        prefs, ctx, jobs = c["prefs"], c.get("user_context"), c["jobs"]
+        targets = {str(j): int(s) for j, s in c["targets"].items()}
+        # engine score = the funnel's white-box AFTER the embedding cosine is threaded in (what the
+        # fallback shows + feeds the sift) — replicate embed→thread→rank so MAE reflects the real funnel.
+        jc = [dict(j) for j in jobs]
+        vibe = _embed.vibe_scores(prefs, ctx, jc)
+        if vibe:
+            for j in jc:
+                if j["id"] in vibe:
+                    j["_vibe_cosine"] = vibe[j["id"]]
+        ranked = _rank.rank(prefs, ctx, jc)["opportunities"]
+        eng = {j["id"]: j.get("matchScore", 0) for j in ranked}
+        # the sift top-K the curator actually sees
+        top, _ = _sift.sift(prefs, ctx, [dict(j) for j in jobs], k=k)
+        top_ids = {j["id"] for j in top}
+
+        score_pairs = [(eng[jid], targets[jid]) for jid in eng if jid in targets]
+        all_score_pairs += score_pairs
+        # membership recall: of target-great jobs, how many made the sift top-K
+        great = [jid for jid, t in targets.items() if t >= GREAT and jid in eng]
+        in_top = [jid for jid in great if jid in top_ids]
+        recall = round(len(in_top) / len(great), 2) if great else None
+        if recall is not None:
+            recalls.append(recall)
+        rows.append({
+            "case": c["label"], "n_jobs": len(jobs), "n_great": len(great),
+            "great_in_topK": len(in_top), "membership_recall": recall,
+            "score_mae": _mae(score_pairs),
+        })
+
+    agg = {
+        "sift_top_k": k,
+        "overall_score_mae": _mae(all_score_pairs),
+        "overall_membership_recall": round(sum(recalls) / len(recalls), 2) if recalls else None,
+        "n_cases": len(cases), "n_pairs": len(all_score_pairs),
+    }
+    if emit_json:
+        print(json.dumps({"cases": rows, "aggregate": agg}, indent=2))
+        return
+    print(f"\n  Regression — engine scoring vs target (SIFT_TOP_K={k})")
+    print("  " + "-" * 78)
+    print(f"  {'case':<40}{'jobs':>5}{'great':>7}{'in-topK':>9}{'recall':>8}{'MAE':>6}")
+    for r in rows:
+        print(f"  {r['case'][:40]:<40}{r['n_jobs']:>5}{r['n_great']:>7}{r['great_in_topK']:>9}"
+              f"{str(r['membership_recall']):>8}{str(r['score_mae']):>6}")
+    print("  " + "-" * 78)
+    print(f"  AGGREGATE → score MAE {agg['overall_score_mae']} · "
+          f"sift membership recall {agg['overall_membership_recall']} "
+          f"({agg['n_pairs']} pairs, {agg['n_cases']} cases)\n")
+
+
+if __name__ == "__main__":
+    run(emit_json="--json" in sys.argv)

tests/test_contracts.py

@@ -0,0 +1,98 @@
+"""Phase-0 contract checks — the guardrails that catch silent failures BEFORE they ship.
+
+These assert the load-bearing contracts the engine-improvement plan depends on:
+  1. BOARDS ↔ NORMALIZERS alignment (an enabled board with no normalizer = silent drop).
+  2. The funnel shapes (sift fusion weights, embed text fns, SIFT_TOP_K) are where the plan expects.
+  3. overlap() returns None when a side has no skills (the floor-free contract).
+  4. The frontend↔backend EXACT-STRING vocab contract: every ScoutPrefs ROLE/INDUSTRY option the UI
+     offers must resolve to a coerce.py board code, else it silently degrades to a keyword fold.
+     (Cross-repo read of growqr-demo's scout.ts; skipped if the sibling repo isn't present.)
+"""
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import pytest
+
+from app.engine import search, normalize, sift, embed
+from app.engine.board_adapters import coerce as C
+from app.engine import skills as S
+from app.config import get_settings
+
+
+# ── 1. BOARDS ↔ NORMALIZERS alignment ────────────────────────────────────────
+def test_boards_have_normalizers():
+    for board in search.BOARDS:
+        assert board in normalize.NORMALIZERS, f"BOARDS has '{board}' with no NORMALIZERS entry"
+
+
+def test_enabled_boards_are_registered():
+    enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip()]
+    for b in enabled:
+        assert b in search.BOARDS, f"BOARDS_ENABLED lists '{b}' which is not a registered board"
+        assert b in normalize.NORMALIZERS, f"enabled board '{b}' has no normalizer"
+
+
+# ── 2. Funnel shapes (guards the Phase-2 edits) ──────────────────────────────
+def test_sift_fusion_weights_present_and_normalized():
+    assert hasattr(sift, "W_WHITEBOX") and hasattr(sift, "W_VIBE")
+    assert abs((sift.W_WHITEBOX + sift.W_VIBE) - 1.0) < 1e-9, "sift fusion weights must sum to 1"
+
+
+def test_embed_text_builders_exist():
+    # the Phase-2 richer-embedding work edits these; assert they exist + return strings
+    assert callable(embed._job_text) and callable(embed._profile_text)
+    assert isinstance(embed._job_text({"title": "X", "organization": "Y"}), str)
+    assert isinstance(embed._profile_text({"title": "X"}, None), str)
+
+
+def test_sift_top_k_configured():
+    assert isinstance(get_settings().SIFT_TOP_K, int) and get_settings().SIFT_TOP_K > 0
+
+
+# ── 3. Floor-free skill contract ─────────────────────────────────────────────
+def test_overlap_none_when_a_side_is_empty():
+    assert S.overlap([], ["python", "sql"]) is None, "empty user skills must yield None (not a floor)"
+    assert S.overlap(["python"], []) is None, "empty job skills must yield None (not a floor)"
+    # both present → a real dict with a score
+    o = S.overlap(["python", "sql"], ["python", "django"])
+    assert isinstance(o, dict) and "score" in o
+
+
+# ── 4. Frontend↔backend EXACT-STRING vocab contract ──────────────────────────
+# Values the UI sends that are intentionally NOT numeric board codes — they fold into the keyword.
+# The ACTIVE boards (naukri-feed, linkedin-recall, foundit, indeed, workindia) all search by keyword +
+# city NAME, so the non-tech vocab below works WITHOUT a code (the code maps are only for the off,
+# code-based actors). Codes for these are deferred per plan; listed here so drift still fails loudly.
+_ROLE_KEYWORD_ONLY = {"Open to any", "Finance", "HR", "Customer Support", "Supply Chain",
+                      "Business Development", "Legal"}
+_INDUSTRY_KEYWORD_ONLY = {"Any", "Open to any", "", "Banking", "Insurance", "Retail", "FMCG",
+                          "Pharma", "Manufacturing", "Logistics"}
+
+_SCOUT_TS = Path(__file__).resolve().parents[2] / "growqr-demo/frontend/src/scout/data/scout.ts"
+
+
+def _extract_array(name: str, text: str) -> list[str]:
+    m = re.search(rf"{name}\s*[:=]\s*\[(.*?)\]", text, re.S)
+    if not m:
+        return []
+    return re.findall(r'"([^"]+)"', m.group(1))
+
+
+@pytest.mark.skipif(not _SCOUT_TS.exists(), reason="growqr-demo frontend not present (cross-repo)")
+def test_frontend_roles_resolve_in_backend():
+    text = _SCOUT_TS.read_text()
+    roles = _extract_array("ROLE_OPTIONS", text)
+    assert roles, "could not extract ROLE_OPTIONS from scout.ts — regex/contract drift"
+    unmapped = [r for r in roles if r not in C.ROLE_NAUKRI_DEPT and r not in _ROLE_KEYWORD_ONLY]
+    assert not unmapped, f"frontend ROLE options with no coerce.py code (silent keyword-fold): {unmapped}"
+
+
+@pytest.mark.skipif(not _SCOUT_TS.exists(), reason="growqr-demo frontend not present (cross-repo)")
+def test_frontend_industries_resolve_in_backend():
+    text = _SCOUT_TS.read_text()
+    inds = _extract_array("INDUSTRY_OPTIONS", text)
+    assert inds, "could not extract INDUSTRY_OPTIONS from scout.ts — regex/contract drift"
+    unmapped = [i for i in inds if i not in C.INDUSTRY_NAUKRI and i not in _INDUSTRY_KEYWORD_ONLY]
+    assert not unmapped, f"frontend INDUSTRY options with no coerce.py code (silent keyword-fold): {unmapped}"

tests/test_curate.py

@@ -8,18 +8,50 @@ def test_parse_json_strips_fences_and_prose():
     assert C._parse_json('here: {"kept":[{"id":"1"}]} thanks')["kept"][0]["id"] == "1"
 
 
-def test_to_match_builds_full_contract():
-    m = C._to_match({
-        "id": "1", "score": 78, "fit": "fit", "archetype": "The Fit", "one_line": "strong but a gap",
-        "breakdown": [{"name": "Skill match", "score": 80, "level": "Strong", "note": "real SQL"}],
-        "coach_note": "lead with X", "growth": {"text": "close Y", "from": 78, "to": 85},
+def test_scored_card_computes_overall_from_rubric():
+    # Stage 1: the curator RATES dimensions; _scored_card COMPUTES the overall (no model-picked score)
+    from app.engine import rubric as R
+    sc = C._scored_card({
+        "id": "1", "fit": "fit",
+        "dimensions": {
+            "role": {"score": 100, "note": "same function"},
+            "skills": {"score": 80, "note": "real SQL"},
+            "seniority": {"score": 90, "note": "right level"},
+            "location": {"score": 100, "note": "target city"},
+            "industry": {"score": 85, "note": "adjacent"},
+            "experience": {"score": 80, "note": "aligned"},
+        },
     })
-    d = m.as_dict()
-    assert d["score"] == 78 and d["archetype"] == "The Fit" and d["fit"] == "fit"
-    assert d["breakdown"][0]["note"] == "real SQL" and d["coach_note"] == "lead with X"
+    expected = R.aggregate({"role": 100, "skills": 80, "seniority": 90,
+                            "location": 100, "industry": 85, "experience": 80})
+    assert sc["overall"] == expected and sc["fit"] == "fit"
+    assert C._scored_card({"id": "x", "dimensions": {}}) is None      # no dims → dropped
+
+
+def test_to_match_merges_score_and_prose():
+    # Stage 1 scored card + Stage 2 prose → the full MatchResult contract
+    sc = C._scored_card({
+        "id": "1", "fit": "fit",
+        "dimensions": {"role": {"score": 100, "note": "same function"}, "skills": {"score": 80, "note": "real SQL"},
+                       "seniority": {"score": 90}, "location": {"score": 100}, "experience": {"score": 80}},
+    })
+    prose = {"archetype": "The Fit", "one_line": "strong but a gap", "coach_note": "lead with X",
+             "growth": {"text": "close Y", "from": 78, "to": 85}}
+    d = C._to_match(sc, prose).as_dict()
+    assert d["score"] == sc["overall"] and d["archetype"] == "The Fit" and d["coach_note"] == "lead with X"
+    assert any(dim["note"] == "real SQL" for dim in d["breakdown"])
     assert d["growth"]["from"] == 78 and d["growth"]["to"] == 85 and d["proofReady"] is True
 
 
+def test_to_match_falls_back_to_templated_prose():
+    # Stage 2 failed (prose=None) → score + breakdown survive, prose is templated (never blank)
+    sc = C._scored_card({"id": "1", "fit": "fit",
+                         "dimensions": {"role": {"score": 95}, "skills": {"score": 85}, "seniority": {"score": 90},
+                                        "location": {"score": 100}, "experience": {"score": 80}}})
+    d = C._to_match(sc, None).as_dict()
+    assert d["score"] == sc["overall"] and d["archetype"] and d["one_line"]   # templated, not empty
+
+
 def test_curate_disabled_returns_none(monkeypatch):
     monkeypatch.setattr(C, "curate_enabled", lambda: False)
     assert C.curate({"title": "PM"}, None, [{"id": "1", "title": "PM"}]) is None

tests/test_e2e_integration.py

@@ -0,0 +1,112 @@
+"""E2E integration — the full curate machinery (deterministic, offline) + an opt-in LIVE stack run.
+
+The offline test mocks the gateway so the WHOLE two-stage curate path runs in CI: Opus-score JSON →
+rubric.aggregate → calibration → floor filter → best-first sort → Haiku prose merge. It asserts the
+contract that matters: an aligned job reaches the 90s, a weak job is DROPPED at the floor, and every
+surviving match decomposes to real evidence.
+
+The live test (RUN_LIVE_E2E=1) runs the real new stack + Opus/Haiku — documents the manual verification
+(sweep → ~133 jobs incl. TimesJobs full-signal → scores reach 96).
+"""
+from __future__ import annotations
+
+import json
+import os
+
+import pytest
+
+from app.engine import curate as C
+from app.config import get_settings
+
+
+# ── deterministic offline e2e (mocks the gateway) ────────────────────────────
+class _Msg:
+    def __init__(self, content): self.content = content
+class _Choice:
+    def __init__(self, content): self.message = _Msg(content); self.finish_reason = "stop"
+class _Resp:
+    def __init__(self, content):
+        self.choices = [_Choice(content)]
+        self.usage = type("U", (), {"prompt_tokens": 50, "completion_tokens": 50})()
+class _Completions:
+    def __init__(self, score_model, score_json, cards_json):
+        self._sm, self._s, self._c = score_model, score_json, cards_json
+    def create(self, *, model, **kw):
+        return _Resp(self._s if model == self._sm else self._c)
+class _Chat:
+    def __init__(self, *a): self.completions = _Completions(*a)
+class _FakeClient:
+    def __init__(self, score_model, score_json, cards_json): self.chat = _Chat(score_model, score_json, cards_json)
+
+
+def test_curate_e2e_scores_calibrate_floor_and_decompose(monkeypatch):
+    s = get_settings()
+    sifted = [
+        {"id": "j1", "title": "Senior Sales Manager", "organization": "Acme", "details": {"description": "lead B2B sales"}},
+        {"id": "j2", "title": "Sales Manager", "organization": "Beta", "details": {"description": "manage accounts"}},
+        {"id": "j3", "title": "Junior Analyst", "organization": "Gamma", "details": {"description": "data entry"}},
+    ]
+    # Opus rates dimensions; j1 = strongly aligned (→ 90s), j2 = good, j3 = weak (raw < floor → dropped).
+    score_json = json.dumps({"kept": [
+        {"id": "j1", "fit": "fit", "dimensions": {"role": {"score": 100, "note": "same function"},
+            "skills": {"score": 92, "note": "B2B + CRM"}, "seniority": {"score": 95, "note": "right level"},
+            "location": {"score": 100, "note": "target city"}, "industry": {"score": 90, "note": "fintech"},
+            "experience": {"score": 90, "note": "9 yrs"}}},
+        {"id": "j2", "fit": "fit", "dimensions": {"role": {"score": 90}, "skills": {"score": 75},
+            "seniority": {"score": 80}, "location": {"score": 100}, "experience": {"score": 75}}},
+        {"id": "j3", "fit": "stretch", "dimensions": {"role": {"score": 20}, "skills": {"score": 25},
+            "seniority": {"score": 30}, "location": {"score": 100}, "experience": {"score": 20}}},
+    ]})
+    cards_json = json.dumps({"cards": [
+        {"id": "j1", "archetype": "The Perfect Fit", "one_line": "Strong on role and seniority.",
+         "coach_note": "Lead with revenue numbers.", "growth": {"text": "x", "from": 90, "to": 96}},
+        {"id": "j2", "archetype": "The Solid Step", "one_line": "Good role match, lighter on skills."},
+    ]})
+    monkeypatch.setattr(C, "curate_enabled", lambda: True)
+    monkeypatch.setattr(C, "gateway_client", lambda: _FakeClient(s.CURATE_MODEL, score_json, cards_json))
+
+    out = C.curate({"title": "Sales Manager"}, {"skills": ["B2B Sales"]}, sifted)
+    assert out is not None
+    ids = [o["id"] for o in out]
+    assert "j3" not in ids, "weak job (raw < floor) must be DROPPED — floor honesty"
+    assert ids == ["j1", "j2"], "best-first by the COMPUTED score"
+    j1 = out[0]
+    assert j1["matchScore"] >= 90, f"aligned job should reach the 90s, got {j1['matchScore']}"
+    assert all(o["matchScore"] >= s.MATCH_FLOOR for o in out), "no sub-floor match surfaces"
+    # decomposes to real evidence
+    bd = {d["name"]: d["score"] for d in j1["match"]["breakdown"]}
+    assert len(bd) >= 5 and j1["match"]["archetype"] == "The Perfect Fit"
+    assert j1["match"]["coach_note"] == "Lead with revenue numbers."
+
+
+def test_curate_floor_only_returns_empty_not_unfiltered_fallback(monkeypatch):
+    # Opus succeeds but keeps nothing ≥ floor → honest empty deck, NEVER the unfiltered white-box
+    s = get_settings()
+    sifted = [{"id": "j1", "title": "X", "organization": "Y", "details": {}}]
+    score_json = json.dumps({"kept": [{"id": "j1", "fit": "stretch",
+        "dimensions": {"role": {"score": 10}, "skills": {"score": 10}, "seniority": {"score": 10},
+                       "location": {"score": 10}, "experience": {"score": 10}}}]})
+    monkeypatch.setattr(C, "curate_enabled", lambda: True)
+    monkeypatch.setattr(C, "gateway_client", lambda: _FakeClient(s.CURATE_MODEL, score_json, "{}"))
+    out = C.curate({"title": "X"}, None, sifted)
+    assert out == [], "all-below-floor → empty, not a fallback that could leak sub-floor matches"
+
+
+# ── opt-in LIVE e2e (real stack + Opus) ──────────────────────────────────────
+@pytest.mark.skipif(os.environ.get("RUN_LIVE_E2E") != "1", reason="live stack run — set RUN_LIVE_E2E=1")
+def test_live_stack_reaches_90s_with_full_signal():
+    import asyncio
+    from app.engine import search, sift as _sift
+    prefs = {"title": "Sales Manager", "location": ["Mumbai · India"], "role": ["Sales"],
+             "industry": ["Fintech"], "experience": ["Senior"],
+             "user_context": {"id": "live", "skills": ["B2B Sales", "Account Management", "CRM"],
+                              "current_role": "Senior Sales Manager", "years_experience": 9}}
+    ctx = prefs["user_context"]
+    sweep = asyncio.run(search.run_sweep({**prefs, "_fresh": True}, fresh=True))
+    assert len(sweep["sources"]) >= 4, "most of the stack should return"
+    assert len(sweep["opportunities"]) >= 80, "deck should be deep"
+    top, _ = _sift.sift(prefs, ctx, [dict(j) for j in sweep["opportunities"]])
+    cur = C.curate(prefs, ctx, top) or []
+    scores = [o["matchScore"] for o in cur]
+    assert scores and max(scores) >= 90, f"aligned profile should reach 90s, got max {max(scores) if scores else None}"
+    assert min(scores) >= get_settings().MATCH_FLOOR, "floor honored"

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

@@ -0,0 +1,91 @@
+"""Warm-pool gate — the reorder/floor decision that keeps the shelf stocked without over-fetching."""
+from types import SimpleNamespace
+
+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)
+
+
+def test_pool_healthy_serves_no_fetch():
+    assert pool_decision(150, False, FLOOR, REORDER) == "pool"
+    assert pool_decision(100, False, FLOOR, REORDER) == "pool"    # exactly at reorder → still no fetch
+
+
+def test_pool_dipping_serves_plus_background_refill():
+    assert pool_decision(99, False, FLOOR, REORDER) == "pool+refill"   # below reorder → top up in bg
+    assert pool_decision(90, False, FLOOR, REORDER) == "pool+refill"   # at the floor → still serve, refill
+
+
+def test_pool_below_floor_blocks_and_fetches():
+    assert pool_decision(89, False, FLOOR, REORDER) == "sweep"    # under the hard floor → blocking fetch
+    assert pool_decision(0, False, FLOOR, REORDER) == "sweep"     # empty pool (first search) → fetch
+
+
+def test_forced_fresh_always_fetches():
+    assert pool_decision(999, True, FLOOR, REORDER) == "sweep"    # dev _fresh toggle bypasses the pool
+    assert pool_decision(5, True, FLOOR, REORDER, exhausted=True) == "sweep"  # force overrides exhaustion
+
+
+def test_exhausted_query_serves_instead_of_block_fetching():
+    # tapped-out niche query below the floor: serve what's pooled, DON'T re-buy duplicates
+    assert pool_decision(30, False, FLOOR, REORDER, exhausted=True) == "pool"   # below floor but exhausted
+    assert pool_decision(1, False, FLOOR, REORDER, exhausted=True) == "pool"    # even 1 job → serve it
+    # but an EMPTY exhausted pool must still try a fetch (nothing to serve)
+    assert pool_decision(0, False, FLOOR, REORDER, exhausted=True) == "sweep"
+    # exhaustion does NOT suppress the healthy/refill paths
+    assert pool_decision(200, False, FLOOR, REORDER, exhausted=True) == "pool"
+
+
+def test_floor_never_above_reorder_invariant():
+    # the gate only makes sense when floor ≤ reorder (reorder early, floor as the hard backstop)
+    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_and_camel_case_work_mode_have_same_signature():
+    common = {"title": "Data Analyst", "location": ["Bangalore · India"]}
+
+    assert search_signature({**common, "workMode": ["Hybrid"]}) == search_signature({**common, "work_mode": ["Hybrid"]})
+
+
+@pytest.mark.asyncio
+async def test_sweep_marks_total_board_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 = await search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]})
+
+    assert result["all_boards_failed"] is True
+    assert result["failed_boards"] == 2
+    assert result["sources"] == {}

tests/test_rubric.py

@@ -0,0 +1,65 @@
+"""The honest rubric — the displayed score is COMPUTED from dimension ratings, never model-emitted."""
+import pytest
+
+from app.engine import rubric as R
+
+
+def test_weights_sum_to_one():
+    assert abs(sum(R.WEIGHTS.values()) - 1.0) < 1e-9
+
+
+def test_perfect_match_scores_high_90s():
+    # all dimensions genuinely aligned → the overall lands in the high 90s (earned, not prompted)
+    perfect = {k: 100 for k in R.KEYS}
+    assert R.aggregate(perfect) == 100
+    strong = {"role": 100, "skills": 92, "seniority": 95, "location": 100, "industry": 95, "experience": 90}
+    assert R.aggregate(strong) >= 94      # a real strong match reaches the 90s
+
+
+def test_partial_match_scores_mid():
+    partial = {"role": 75, "skills": 50, "seniority": 70, "location": 60, "industry": 65, "experience": 60}
+    s = R.aggregate(partial)
+    assert 55 <= s <= 72                   # honestly a partial, not inflated
+
+
+def test_weak_role_drags_score_down():
+    # role is 25% — a wrong-function job can't score high no matter how good the rest is
+    weak = {"role": 15, "skills": 90, "seniority": 90, "location": 100, "industry": 90, "experience": 90}
+    assert R.aggregate(weak) < 80
+
+
+def test_absent_dimension_redistributes_not_zeroes():
+    # industry absent (no preference) must NOT act as a 0 that tanks the score — weight redistributes
+    no_ind = {"role": 100, "skills": 100, "seniority": 100, "location": 100, "experience": 100}
+    assert R.aggregate(no_ind) == 100      # not dragged down by the missing 10% industry weight
+
+
+def test_empty_is_zero_not_crash():
+    assert R.aggregate({}) == 0
+
+
+def test_scores_are_clamped():
+    assert R.aggregate({k: 150 for k in R.KEYS}) == 100
+    assert R.aggregate({k: -20 for k in R.KEYS}) == 0
+
+
+def test_level_bands():
+    assert R.level_for(90) == "Strong" and R.level_for(60) == "Solid" and R.level_for(40) == "Light"
+
+
+def test_prompt_block_lists_every_dimension():
+    block = R.prompt_block()
+    for k in R.KEYS:
+        assert f'"{k}"' in block           # every rubric key is shown to the curator
+
+
+def test_calibration_is_monotonic_floor_anchored_and_lifts_top():
+    # the transparent "little bias": strictly increasing, floor fixed, top expanded into the 90s
+    assert R.calibrate(50) == 50                     # floor unchanged — weak stays weak
+    assert R.calibrate(49) == 49 and R.calibrate(20) == 20   # below floor untouched
+    assert R.calibrate(100) == 100                   # ceiling fixed
+    assert R.calibrate(90) >= 93                     # genuinely-strong → presents in the 90s
+    assert R.calibrate(60) <= 65                     # a weak match is NOT lifted into the strong zone
+    # strictly monotonic across the range (ordering never lies)
+    vals = [R.calibrate(x) for x in range(50, 101, 5)]
+    assert all(b > a for a, b in zip(vals, vals[1:]))

tests/test_search_retry.py

@@ -0,0 +1,51 @@
+"""Retry-on-empty guard — scrape-flaky boards (Naukri-feed, curious_coder LinkedIn) intermittently
+return 0; one fresh retry rescues the transient case. RCA: the same input flaked 0 then returned 15."""
+import asyncio
+
+from app.engine import search
+
+
+def _run(coro):
+    return asyncio.run(coro)
+
+
+def test_retry_on_empty_recovers_transient_zero(monkeypatch):
+    calls = {"n": 0}
+
+    async def fake_run_actor(actor, inp, cache_mode=None):
+        calls["n"] += 1
+        # first call flakes (0), retry returns a real job
+        if calls["n"] == 1:
+            return []
+        return [{"profile_job_title": "Data Architect", "branch_company_name": "Acme",
+                 "branch_location_city_name": "Delhi", "skills": ["python"], "job_id": "1"}]
+
+    monkeypatch.setattr(search, "run_actor", fake_run_actor)
+    key, jobs = _run(search._fetch_board("workindia", {"title": "Data Architect", "location": ["Delhi · India"]}))
+    assert calls["n"] == 2, "a board returning 0 must be retried exactly once"
+    assert len(jobs) == 1, "the retry's jobs are used"
+
+
+def test_no_retry_when_first_call_succeeds(monkeypatch):
+    calls = {"n": 0}
+
+    async def fake_run_actor(actor, inp, cache_mode=None):
+        calls["n"] += 1
+        return [{"profile_job_title": "X", "branch_company_name": "Y",
+                 "branch_location_city_name": "Delhi", "skills": [], "job_id": "1"}]
+
+    monkeypatch.setattr(search, "run_actor", fake_run_actor)
+    _run(search._fetch_board("workindia", {"title": "X", "location": ["Delhi · India"]}))
+    assert calls["n"] == 1, "a board that returns jobs is NOT retried (no extra cost)"
+
+
+def test_genuinely_empty_returns_empty_after_one_retry(monkeypatch):
+    calls = {"n": 0}
+
+    async def fake_run_actor(actor, inp, cache_mode=None):
+        calls["n"] += 1
+        return []   # genuinely empty (e.g. WorkIndia for a tech role)
+
+    monkeypatch.setattr(search, "run_actor", fake_run_actor)
+    key, jobs = _run(search._fetch_board("workindia", {"title": "AI Architect", "location": ["Delhi · India"]}))
+    assert calls["n"] == 2 and jobs == [], "retry once, then accept the empty result (no infinite loop)"

tests/test_sift.py

@@ -1,12 +1,13 @@
-"""A1 — sift: rank-fusion math + graceful degradation when the vibe-engineer is unavailable.
+"""A1 — sift: magnitude-fusion math + graceful degradation when the vibe-engineer is unavailable.
 (Offline — no network; the embedding path is exercised in the cached e2e verification.)"""
 from app.engine import sift as S
 
 
-def test_ranks_normalization():
-    r = S._ranks({"a": 0.9, "b": 0.5, "c": 0.1})
-    assert r["a"] == 1.0 and r["c"] == 0.0 and 0.0 < r["b"] < 1.0
-    assert S._ranks({"x": 0.5}) == {"x": 1.0}                 # single item → top
+def test_minmax_normalization():
+    # magnitude-preserving (min-max), NOT rank-position: a much-better score stays much better.
+    r = S._minmax({"a": 0.9, "b": 0.5, "c": 0.1})
+    assert r["a"] == 1.0 and r["c"] == 0.0 and abs(r["b"] - 0.5) < 1e-9
+    assert S._minmax({"x": 0.5}) == {"x": 1.0}                # single item / no range → top
 
 
 def test_sift_degrades_to_whitebox_when_no_vibe(monkeypatch):

tests/test_stack_contracts.py

@@ -0,0 +1,158 @@
+"""Step-0 DEEP contract tests for the new signal-weighted stack.
+
+These lock the load-bearing contracts BEFORE the actors land, so nothing is missed:
+  1. Stack alignment — BOARDS_ENABLED == the target stack; Indeed off-but-registered; every enabled
+     board has a builder + normalizer (an enabled board with no normalizer = silent drop).
+  2. Builder contract — every enabled board's input builder emits a NON-EMPTY query (no actor 400s).
+  3. Normalizer contract (on REAL cached data) — every board's normalizer turns real raw items into
+     valid ScoutJobs with the required fields AND DISTINCT dedup keys (the WorkIndia 20→1 collapse class).
+  4. Signal contract — each board's normalizer actually yields the signals we claim it does (skills /
+     salary / seniority), so a regression in the actor or mapping fails loudly.
+  5. Pool contract — floor ≤ reorder, and the gate's three outcomes are well-defined.
+
+Data contracts read .apify_cache (real captured sweeps). A target board with no cache FAILS (so we can't
+'forget' to validate a new actor) — except where explicitly allowed pre-build.
+"""
+from __future__ import annotations
+
+import glob
+import json
+import os
+
+import pytest
+
+from app.engine import search, normalize as N
+from app.engine.search import pool_decision
+from app.config import get_settings
+
+# ── the TARGET stack (what we're building toward) ────────────────────────────
+TARGET_STACK = ["naukri", "foundit", "linkedin", "timesjobs", "workindia"]
+OFF_BUT_REGISTERED = "indeed"            # dropped from enabled, kept registered
+
+# board → cache-file actor prefix (real raw data to validate the normalizer against)
+CACHE_PREFIX = {
+    "naukri": "blackfalcondata_naukri-jobs-feed",
+    "foundit": "shahidirfan_Foundit-Jobs-Scraper",
+    "linkedin": "curious_coder_linkedin-jobs-scraper",
+    "timesjobs": "shahidirfan_Timesjobs-Scraper",
+    "workindia": "shahidirfan_workindia-jobs-scraper",
+}
+# signals each board MUST confirm (guards actor / mapping regressions). 'offsite' = a real apply link.
+SIGNAL_MAP = {
+    "naukri": {"skills", "offsite"},
+    "foundit": {"skills", "offsite"},
+    "linkedin": {"industry"},            # curious_coder structured industries (100%); seniority often N/A in IN
+    "timesjobs": {"skills", "salary"},
+    "workindia": {"skills"},
+}
+
+SAMPLE_PREFS = {
+    "title": "Sales Manager", "location": ["Mumbai · India"], "role": ["Sales"],
+    "industry": ["Fintech"], "experience": ["Senior"],
+    "user_context": {"skills": ["B2B Sales", "Account Management", "CRM"]},
+}
+REQUIRED_FIELDS = {"id", "title", "organization", "location_country", "required_skills", "details"}
+# Stable, COMMITTED per-board raw samples (curated from real sweeps) — so the contract is reproducible
+# and not at the mercy of whatever is currently in the volatile .apify_cache.
+_SAMPLES_PATH = os.path.join(os.path.dirname(__file__), "fixtures", "board_samples.json")
+_SAMPLES = json.load(open(_SAMPLES_PATH)) if os.path.exists(_SAMPLES_PATH) else {}
+
+
+def _board_sample(board: str):
+    return _SAMPLES.get(board)
+
+
+def _dedup_key(j: dict) -> str:
+    return (j.get("apply_url") or "").strip() or f"{j['organization']}|{j['title']}|{j.get('location_city')}".lower()
+
+
+def _has_signal(sj: dict, signal: str) -> bool:
+    d = sj.get("details") or {}
+    if signal == "skills":
+        return bool(d.get("skills") or sj.get("required_skills"))
+    if signal == "salary":
+        return sj.get("salary_lpa") is not None or bool(sj.get("payLabel") and "disclos" not in (sj["payLabel"] or "").lower())
+    if signal == "seniority":
+        return bool(sj.get("seniority_level"))
+    if signal == "industry":
+        return bool(sj.get("industry") or (sj.get("details") or {}).get("industry"))
+    if signal == "offsite":
+        return bool(sj.get("offsite_apply"))
+    return False
+
+
+# ── 1. Stack alignment ───────────────────────────────────────────────────────
+def test_enabled_equals_target_stack():
+    enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip()]
+    assert set(enabled) == set(TARGET_STACK), f"BOARDS_ENABLED {enabled} != target {TARGET_STACK}"
+
+
+def test_indeed_off_but_registered():
+    enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",")]
+    assert OFF_BUT_REGISTERED not in enabled, "indeed must be DROPPED from enabled"
+    assert OFF_BUT_REGISTERED in search.BOARDS, "indeed must stay REGISTERED (off, not deleted)"
+
+
+def test_every_target_board_has_builder_and_normalizer():
+    for b in TARGET_STACK:
+        assert b in search.BOARDS, f"target board '{b}' not registered in BOARDS"
+        assert b in N.NORMALIZERS, f"target board '{b}' has no normalizer (silent drop)"
+
+
+# ── 2. Builder contract ──────────────────────────────────────────────────────
+def test_every_enabled_builder_emits_nonempty_query():
+    enabled = [b for b in TARGET_STACK if b in search.BOARDS]
+    for b in enabled:
+        inp = search.BOARDS[b][1](dict(SAMPLE_PREFS))
+        assert isinstance(inp, dict) and inp, f"{b} builder returned empty input"
+        q = inp.get("keyword") or inp.get("title") or (inp.get("jobTitles") or [None])[0] or inp.get("includeKeyword") or inp.get("urls")
+        assert q, f"{b} builder emitted no query/keyword/title → actor would 400"
+
+
+# ── 3 + 4. Normalizer + signal contracts on REAL cached data ─────────────────
+@pytest.mark.parametrize("board", TARGET_STACK)
+def test_normalizer_contract_on_real_data(board):
+    items = _board_sample(board)
+    if items is None:
+        pytest.fail(f"no committed sample for '{board}' in {_SAMPLES_PATH} — capture one")
+    fn = N.NORMALIZERS[board]
+    jobs = [sj for it in items[:25] if (sj := fn(it))]
+    assert len(jobs) >= 2, f"{board}: normalizer produced < 2 jobs from {len(items)} raw items"
+    # required fields present
+    for j in jobs:
+        missing = REQUIRED_FIELDS - set(j)
+        assert not missing, f"{board} ScoutJob missing {missing}"
+        assert isinstance(j["required_skills"], list) and isinstance(j["details"], dict)
+    # ANTI-COLLAPSE: distinct dedup keys (the WorkIndia 20→1 bug class)
+    keys = [_dedup_key(j) for j in jobs]
+    assert len(set(keys)) >= max(2, int(0.8 * len(jobs))), \
+        f"{board}: dedup keys collapse ({len(set(keys))} unique of {len(jobs)}) — apply_url not unique"
+    # SIGNAL contract: the board confirms what we claim, on a majority of jobs
+    for sig in SIGNAL_MAP[board]:
+        hits = sum(_has_signal(j, sig) for j in jobs)
+        assert hits >= 0.5 * len(jobs), f"{board}: signal '{sig}' present on only {hits}/{len(jobs)} (claimed)"
+
+
+# ── 4b. Legacy-board city aliasing (TimesJobs/Foundit need "Delhi" not "New Delhi") ──
+@pytest.mark.parametrize("builder", ["build_timesjobs_input", "build_foundit_input"])
+def test_legacy_boards_alias_modern_city_names(builder):
+    from app.engine.board_adapters import adapters as A
+    fn = getattr(A, builder)
+    for modern, legacy in [("New Delhi · India", "Delhi"), ("Bengaluru · India", "Bangalore"),
+                           ("Gurugram · India", "Gurgaon")]:
+        loc = fn({"title": "Data Architect", "location": [modern], "role": ["Data"]}).get("location")
+        assert loc == legacy, f"{builder}: {modern!r} must alias to {legacy!r} (board returns 0 otherwise), got {loc!r}"
+
+
+# ── 5. Pool contract ─────────────────────────────────────────────────────────
+def test_pool_floor_le_reorder():
+    s = get_settings()
+    assert s.POOL_SAFETY_FLOOR <= s.POOL_REORDER_AT, "floor must be ≤ reorder (reorder early, floor backstop)"
+
+
+def test_pool_gate_outcomes_well_defined():
+    f, r = 90, 100
+    assert pool_decision(200, False, f, r) == "pool"
+    assert pool_decision(95, False, f, r) == "pool+refill"
+    assert pool_decision(50, False, f, r) == "sweep"
+    assert pool_decision(9999, True, f, r) == "sweep"      # forced fresh always sweeps

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
 
@@ -31,3 +36,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]