diff --git a/app/agent/session.py b/app/agent/session.py index 20fe232..0707c90 100644 --- a/app/agent/session.py +++ b/app/agent/session.py @@ -9,6 +9,7 @@ say "not implemented yet" so we can fill them in one full-stack slice at a time. """ from __future__ import annotations +import asyncio import logging logger = logging.getLogger(__name__) @@ -36,7 +37,7 @@ class MatchmakingAgentSession: handlers = { # existing contract "sync_preferences": self.handle_sync_preferences, - "get_feed": self.handle_get_feed, + "get_scout_feed": self.handle_get_feed, # unique name (avoids the course-service "get_feed" collision) "record_feedback": self.handle_record_feedback, "get_opportunity_detail": self.handle_get_opportunity_detail, # new in v2 @@ -60,7 +61,10 @@ class MatchmakingAgentSession: await self._stub("preferences_synced", "persist ScoutPrefs → label store") async def handle_get_feed(self, params: dict): - await self._stub("feed_loaded", "return cached ranked feed") + """Replay the user's last persisted feed so matches survive navigation (no re-search).""" + 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}) async def handle_record_feedback(self, params: dict): await self._stub("feedback_recorded", "store SAVE/DISMISS/APPLY label") @@ -73,13 +77,40 @@ class MatchmakingAgentSession: from app.engine import search 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 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="Scanning live roles across job boards…") - result = await search.run_sweep(prefs) + await self.push("agent_thinking", + message="Fetching fresh roles across job boards…" if fresh + else "Scanning live roles across job boards…") + sweep = await search.run_sweep(prefs, fresh=fresh) + # 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. + from app.engine import curate as _curate + from app.engine import rank as _rank + from app.engine import sift as _sift + user_context = params.get("user_context") # resume skills / experience / education / QScore + await self.push("agent_thinking", message="Scoring roles against your profile…") + # Run the blocking LLM work (embeddings + Opus) OFF the event loop, or the long sync OpenAI + # call freezes the loop and the Redis response can't publish (→ the loader hangs forever). + top, dbg = await asyncio.to_thread(_sift.sift, prefs, user_context, sweep["opportunities"]) + await self.push("agent_thinking", message="Scout is reading your top roles…") + curated = await asyncio.to_thread(_curate.curate, prefs, user_context, top) # Opus's call is final + if curated is None: # safety net: sift + templated cards + curated = _rank.select(top, prefs.get("stretch", "balanced")) + engine = f"fallback:{dbg['mode']}" + else: + engine = "opus" + result = {"opportunities": curated, "sources": sweep["sources"], + "shortlisted": len(curated), "scanned": dbg["scored"], "engine": engine, + "has_profile": bool(user_context and user_context.get("skills"))} await self.push("agent_data", action="search_complete", data=result) + # Persist the feed so it survives navigation/refresh (replayed by handle_get_feed). + from app.db.repo import save_feed + clean_prefs = {k: v for k, v in prefs.items() if k not in ("user_context", "_fresh")} + await save_feed(self.user_id, clean_prefs, result) async def handle_tailor_resume(self, params: dict): await self._stub("resume_tailored", "resume-builder: tailor resume to the chosen role") diff --git a/app/config.py b/app/config.py index be38095..46f0da2 100644 --- a/app/config.py +++ b/app/config.py @@ -32,6 +32,15 @@ class Settings(BaseSettings): APIFY_TOKEN: str | None = None OPENAI_API_KEY: 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. + 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 + EMBED_MODEL: str = "text-embedding-3-small" # the vibe-engineer (direct OpenAI — gateway has no embeddings route) + 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 + # Result cache (dev cost-saver). Keyed by (actor, exact input) → saved job set on disk. # off = always live, never save (PRODUCTION default) # readwrite = replay if saved else live + save (DEV: one live sweep, then $0 replays) diff --git a/app/db/__init__.py b/app/db/__init__.py new file mode 100644 index 0000000..1c1211e --- /dev/null +++ b/app/db/__init__.py @@ -0,0 +1,5 @@ +"""Persistence layer — a dedicated Postgres for the per-user feed (so matches survive navigation). + +Optional by design: if `DATABASE_URL` is unset or the DB is unreachable, the engine still serves live +searches — it just can't replay them. Every call here degrades gracefully (logs + no-ops), never crashes. +""" diff --git a/app/db/models.py b/app/db/models.py new file mode 100644 index 0000000..6415339 --- /dev/null +++ b/app/db/models.py @@ -0,0 +1,27 @@ +"""ORM models. One table for now: the per-user cached feed (the last completed search).""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import JSON, DateTime, Integer, String, func +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + + +class Base(DeclarativeBase): + pass + + +class UserFeed(Base): + """The user's last completed search — replayed on load so matches survive navigation. + Upserted on each `search_complete`; read by `get_feed`. One row per user (PK = user_id).""" + __tablename__ = "user_feed" + + user_id: Mapped[str] = mapped_column(String, primary_key=True) + prefs: Mapped[dict] = mapped_column(JSON, default=dict) # the ScoutPrefs that produced it + opportunities: Mapped[list] = mapped_column(JSON, default=list) # the curated deck (with match blocks) + sources: Mapped[dict] = mapped_column(JSON, default=dict) # per-board counts + engine: Mapped[str] = mapped_column(String, default="") # "opus" | "fallback:…" + scanned: Mapped[int] = mapped_column(Integer, default=0) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) diff --git a/app/db/repo.py b/app/db/repo.py new file mode 100644 index 0000000..3b36eb1 --- /dev/null +++ b/app/db/repo.py @@ -0,0 +1,78 @@ +"""Feed repository — upsert + read the per-user cached feed. All calls degrade to no-op/None +when the DB is off or unreachable (persistence is optional; the engine must never crash on it).""" +from __future__ import annotations + +import logging + +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert + +from app.db.models import Base, UserFeed +from app.db.session import engine, session_factory + +logger = logging.getLogger(__name__) + + +async def init_db() -> None: + """Create tables if the DB is configured (best-effort; logged, never fatal).""" + eng = engine() + if eng is None: + logger.info("DATABASE_URL unset — feed persistence disabled") + return + try: + async with eng.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + logger.info("feed store ready") + except Exception as e: # noqa: BLE001 + logger.warning("feed store init failed (persistence off): %s", e) + + +async def save_feed(user_id: str, prefs: dict, result: dict) -> None: + """Upsert the user's last completed search. No-op if the DB is off.""" + factory = session_factory() + if factory is None or not user_id: + return + row = { + "user_id": user_id, + "prefs": prefs or {}, + "opportunities": result.get("opportunities") or [], + "sources": result.get("sources") or {}, + "engine": result.get("engine") or "", + "scanned": int(result.get("scanned") or 0), + } + try: + async with factory() as s: + stmt = pg_insert(UserFeed).values(**row) + stmt = stmt.on_conflict_do_update( + index_elements=[UserFeed.user_id], + set_={k: row[k] for k in ("prefs", "opportunities", "sources", "engine", "scanned")}, + ) + await s.execute(stmt) + await s.commit() + except Exception as e: # noqa: BLE001 + logger.warning("save_feed failed (non-fatal): %s", e) + + +async def get_feed(user_id: str) -> dict | None: + """The user's last cached feed as a search_complete-shaped dict, or None.""" + factory = session_factory() + if factory is None or not user_id: + return None + 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("get_feed failed (non-fatal): %s", e) + return None + if not row: + return None + return { + "opportunities": row.opportunities or [], + "sources": row.sources or {}, + "prefs": row.prefs or {}, + "engine": row.engine, + "scanned": row.scanned, + "shortlisted": len(row.opportunities or []), + "cached": True, + "updated_at": row.updated_at.isoformat() if row.updated_at else None, + } diff --git a/app/db/session.py b/app/db/session.py new file mode 100644 index 0000000..7329a56 --- /dev/null +++ b/app/db/session.py @@ -0,0 +1,37 @@ +"""Async SQLAlchemy engine + session factory (mirrors qscore-service/app/db/session.py). + +Lazily built so the service boots even when no DB is configured — `engine()` returns None then, +and every caller treats that as "persistence off". +""" +from __future__ import annotations + +import logging +from functools import lru_cache + +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.config import get_settings + +logger = logging.getLogger(__name__) + + +@lru_cache +def _engine_and_factory(): + s = get_settings() + if not s.DATABASE_URL: + return None, None + eng = create_async_engine(s.DATABASE_URL, pool_pre_ping=True, pool_size=5, max_overflow=10) + factory = async_sessionmaker(eng, class_=AsyncSession, expire_on_commit=False) + return eng, factory + + +def engine(): + return _engine_and_factory()[0] + + +def session_factory(): + return _engine_and_factory()[1] + + +def db_enabled() -> bool: + return engine() is not None diff --git a/app/engine/apify_client.py b/app/engine/apify_client.py index 24a1014..aa9716c 100644 --- a/app/engine/apify_client.py +++ b/app/engine/apify_client.py @@ -49,12 +49,14 @@ _MAX_ATTEMPTS = 2 _BACKOFF = 2.0 -async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0) -> list[dict]: +async def run_actor(actor_id: str, run_input: dict, *, timeout: float = 240.0, + cache_mode: str | None = None) -> list[dict]: """POST input → return the dataset items (list). One retry on transient 5xx; raises otherwise. - Result cache (APIFY_CACHE_MODE): dev replays a saved job set for $0; production stays live. + Result cache: dev replays a saved job set for $0; production stays live. `cache_mode` + overrides the configured APIFY_CACHE_MODE per call (e.g. "refresh" for a forced fresh search). """ - mode = get_settings().APIFY_CACHE_MODE + mode = cache_mode or get_settings().APIFY_CACHE_MODE path = _cache_path(actor_id, run_input) if mode != "off" else None if path and mode in ("readwrite", "read"): diff --git a/app/engine/curate.py b/app/engine/curate.py new file mode 100644 index 0000000..2cfb0d1 --- /dev/null +++ b/app/engine/curate.py @@ -0,0 +1,140 @@ +"""The senior recruiter — Opus curates the sifted ~18 into the honest shortlist and writes the report +cards in the interview-service's Gemini video-analysis voice. **Opus's call is final.** + +One gateway call: profile + the ~18 sifted jobs → strict JSON of the KEPT jobs (honest count ≤ the +pool — fewer when warranted, never padded), each carrying the full report-card contract. Returns `None` +on any failure (no key / bad JSON / API error) so the caller falls back to the sift + templated cards. +""" +from __future__ import annotations + +import json +import logging + +from app.config import get_settings +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: + +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. + +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). + +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 _profile_brief(prefs: dict, ctx: dict | None) -> dict: + ctx = ctx or {} + return { + "target_title": prefs.get("title"), + "target_roles": prefs.get("role"), + "target_location": prefs.get("location"), + "seniority": prefs.get("experience"), + "years": prefs.get("years"), + "target_industry": prefs.get("industry"), + "skills": ctx.get("skills"), + "current_role": ctx.get("current_role"), + "stretch_appetite": prefs.get("stretch", "balanced"), + } + + +def _job_brief(j: dict) -> dict: + d = j.get("details") or {} + return { + "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": (d.get("skills") or [])[:12], + "snippet": (d.get("description") or "")[:500], + "prelim_score": j.get("matchScore"), + } + + +def _parse_json(text: str) -> dict: + """Tolerant parse — strip markdown fences / surrounding prose if the model added any.""" + t = (text or "").strip() + if t.startswith("```"): + t = t.split("```", 2)[1].removeprefix("json").strip() if "```" in t[3:] else t.strip("`") + i, j = t.find("{"), t.rfind("}") + if i != -1 and j != -1: + t = t[i : j + 1] + 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 + 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={}, + ) + + +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.""" + 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, + ) + 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) + return None + + by_id = {j["id"]: j for j in sifted_jobs} + out: list[dict] = [] + for k in data.get("kept") or []: + job = by_id.get(k.get("id")) + if not job: + continue + try: + m = _to_match(k) + except Exception: # noqa: BLE001 — skip a malformed card, keep the rest + continue + job["match"] = m.as_dict() + job["matchScore"] = m.score + out.append(job) + return out or None diff --git a/app/engine/embed.py b/app/engine/embed.py new file mode 100644 index 0000000..c9a02e9 --- /dev/null +++ b/app/engine/embed.py @@ -0,0 +1,60 @@ +"""The vibe-engineer — embedding similarity (ENGINE_DESIGN §3 embedding ranker). + +Cosine(profile-text, job-text) via `text-embedding-3-small`: one batched call (profile + all jobs), +the profile embedded alongside. Returns id → cosine in [0,1]; `None` when embeddings are unavailable +(no key / API error / empty profile) so the sift degrades to white-box only. +""" +from __future__ import annotations + +from app.config import get_settings +from app.engine.llm import embed_client + + +def _job_text(job: dict) -> str: + d = job.get("details") or {} + parts = [ + job.get("title", ""), + job.get("organization", ""), + " ".join(d.get("skills") or []), + (d.get("description") or "")[:600], + ] + return " · ".join(p for p in parts if p) + + +def _profile_text(prefs: dict, ctx: dict | None) -> str: + ctx = ctx or {} + parts = [ + prefs.get("title", ""), + " ".join(prefs.get("role") or []), + " ".join(ctx.get("skills") or []), + ctx.get("current_role", "") or "", + " ".join(prefs.get("industry") or []), + ] + return " · ".join(p for p in parts if p) + + +def _cos(a, b) -> float: + import numpy as np + + a, b = np.asarray(a, dtype=float), np.asarray(b, dtype=float) + n = float(np.linalg.norm(a) * np.linalg.norm(b)) + return float(a @ b / n) if n else 0.0 + + +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: + 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: + 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)} diff --git a/app/engine/llm.py b/app/engine/llm.py new file mode 100644 index 0000000..b93eff3 --- /dev/null +++ b/app/engine/llm.py @@ -0,0 +1,40 @@ +"""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`). + +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). +""" +from __future__ import annotations + +from functools import lru_cache + +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.""" + s = get_settings() + if not (s.DSPY_API_BASE and s.DSPY_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) + + +@lru_cache +def embed_client(): + """Direct OpenAI client for embeddings (the gateway has no embeddings route). 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) + + +def curate_enabled() -> bool: + """Is the Opus curator usable? (flag on + gateway key present).""" + s = get_settings() + return bool(s.ENGINE_LLM_ENABLED and s.DSPY_API_BASE and s.DSPY_API_KEY) diff --git a/app/engine/rank.py b/app/engine/rank.py new file mode 100644 index 0000000..3fed5d9 --- /dev/null +++ b/app/engine/rank.py @@ -0,0 +1,290 @@ +"""Coverage-aware, FLOOR-FREE white-box ranker (Phase 1 — ENGINE_DESIGN §3, floor-free per §3.3). + +The cheap cold-start ranker — one of the three rankers fused in Phase 3, NOT "the engine." +Hard rule (the one the old engine broke): **every factor returns a real 0–1 score OR `None`**; +an absent factor leaves the weighted average (its weight redistributes) — never a constant floor. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +from app.engine import skills as S +from app.engine.board_adapters import coerce as C +from app.engine.schema import Growth, MatchDim, MatchResult + +# Expert-prior weights (§3.4 — to calibrate, not optima). semantic = keyword placeholder → embedding in §3. +WEIGHTS = {"skill": 0.35, "experience": 0.25, "location": 0.15, "salary": 0.10, "industry": 0.05, + "role": 0.05, "semantic": 0.10} +FACTOR_LABELS = {"skill": "Skill match", "experience": "Experience", "location": "Location fit", + "salary": "Salary fit", "industry": "Industry", "role": "Role fit", "semantic": "Role relevance"} +# job seniority_level → (min, max) years band +BAND_YEARS = {"junior": (0, 3), "mid": (3, 8), "senior": (8, 13), "lead": (12, 99)} +SELECT = {"safe": 0.85, "balanced": 0.70, "reach": 0.55} +FLOOR, CAP = 10, 25 +_STOP = {"the", "and", "for", "with", "you", "our", "your", "are", "this", "that", "will", "job", "role"} + + +@dataclass +class UserProfile: + title: str | None = None + skills: set[str] = field(default_factory=set) + years: int | None = None + cities: list[str] = field(default_factory=list) + countries: list[str] = field(default_factory=list) + onsite_only: bool = False + has_loc_pref: bool = False + industries: list[str] = field(default_factory=list) + roles: list[str] = field(default_factory=list) + comp_lpa: float | None = None # target minimum, in lakhs + deal_breakers: list[str] = field(default_factory=list) + stretch: str = "balanced" + profile_tokens: set[str] = field(default_factory=set) + + +def _comp_min_lpa(prefs: dict) -> float | None: + nums = re.findall(r"\d+", str(prefs.get("targetComp") or "")) + return float(nums[0]) if nums else None + + +def profile_from(prefs: dict, user_context: dict | None = None) -> UserProfile: + locs = C.parsed_locations(prefs) + cities = [lp["city"].lower() for lp in locs if lp.get("city")] + 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) + 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), + cities=cities, countries=countries, + onsite_only=("on-site" in wm or "onsite" in wm), + has_loc_pref=bool(cities or countries), + industries=[i for i in (prefs.get("industry") or []) if i and i != "Any"], + roles=[r for r in (prefs.get("role") or []) if r], + comp_lpa=_comp_min_lpa(prefs), + deal_breakers=prefs.get("dealBreakers") or [], + stretch=(prefs.get("stretch") or "balanced"), + profile_tokens=toks, + ) + + +# ── job accessors ── +def _jskills(job): return (job.get("details") or {}).get("skills") or job.get("required_skills") or [] +def _jindustry(job): return (job.get("details") or {}).get("industry") or job.get("industry") +def _jrole(job): return (job.get("details") or {}).get("role_category") + + +# ── factors: each returns float in [0,1] OR None (absent → weight redistributes) ── +def f_skill(p, job): + o = S.overlap(p.skills, _jskills(job)) + return o["score"] if o else None + + +def f_experience(p, job): + band = BAND_YEARS.get((job.get("seniority_level") or "").lower()) + if band is None or p.years is None: + return None + lo, hi = band + if p.years < lo: + return max(0.2, 1.0 - (lo - p.years) / max(lo, 1) * 1.2) + if p.years > hi: + return max(0.4, 1.0 - (p.years - hi) / max(p.years, 1) * 0.9) + return 1.0 + + +def f_location(p, job): + if (job.get("location_mode") or "").lower() == "remote": + return 0.0 if p.onsite_only else 1.0 + if not p.has_loc_pref: + return None # "Anywhere"/open → absent, NOT a 60 floor + jl = f"{job.get('location_city') or ''} {job.get('location_country') or ''}".lower().strip() + if not jl: + return None + if any(c and c in jl for c in p.cities): + return 1.0 # exact city / metro + if any(c and c in jl for c in p.countries): + return 0.6 # same country, different city + return 0.0 # different country → hard-filtered out + + +def f_salary(p, job): + jl = job.get("salary_lpa") + if jl is None or p.comp_lpa is None: + return None + if jl >= p.comp_lpa: + return 1.0 + gap = (p.comp_lpa - jl) / max(p.comp_lpa, 1) + return max(0.3, 1.0 - gap) + + +def _cat_match(wants, value): + if not wants or not value: + return None + v = str(value).lower() + return 1.0 if any(w.lower() in v or v in w.lower() for w in wants) else 0.3 + + +def f_industry(p, job): return _cat_match(p.industries, _jindustry(job)) +def f_role(p, job): return _cat_match(p.roles, _jrole(job)) + + +def f_semantic(p, job): + if not p.profile_tokens: + return None + blob = f"{job.get('title') or ''} {(job.get('details') or {}).get('description') or ''}".lower() + jt = {t for t in re.findall(r"[a-z0-9+#]+", blob) if len(t) > 2 and t not in _STOP} + if not jt: + return None + return len(p.profile_tokens & jt) / len(p.profile_tokens) + + +FACTORS = {"skill": f_skill, "experience": f_experience, "location": f_location, "salary": f_salary, + "industry": f_industry, "role": f_role, "semantic": f_semantic} + + +def utility(p: UserProfile, job: dict): + 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) + U = (num / den) if den else None # NO present signals → None (filtered out) + return U, factors + + +def passes_filter(p: UserProfile, job: dict) -> bool: + loc = f_location(p, job) + if loc == 0.0: # remote-for-onsite-only OR explicit no-fit + return False + band = BAND_YEARS.get((job.get("seniority_level") or "").lower()) + if band and p.years is not None and p.stretch != "reach": + lo, hi = band + if p.years + (3 if p.stretch == "balanced" else 0) < lo: # job needs much more seniority + return False + if p.years - 8 > hi: # wildly over-qualified + return False + return True + + +def _level(score: int) -> str: + return "Strong" if score >= 72 else "Solid" if score >= 52 else "Light" + + +# per-dimension one-line notes (the Gemini report-card voice). Templated → Opus replaces later. +_DIM_NOTES = { + "skill": {"Strong": "Covers most of the role's must-have skills", "Solid": "Covers several must-haves — a few gaps", "Light": "Light on the listed skills"}, + "experience": {"Strong": "Your level sits right in their band", "Solid": "Just outside their experience band", "Light": "Off their experience band"}, + "location": {"Strong": "Right in your target location", "Solid": "Same country, different city", "Light": "Off your preferred locations"}, + "salary": {"Strong": "Meets your target pay", "Solid": "Near your target pay", "Light": "Below your target pay"}, + "industry": {"Strong": "In your target industry", "Solid": "An adjacent industry", "Light": "A different industry"}, + "role": {"Strong": "Matches your target role area", "Solid": "An adjacent role area", "Light": "A different role area"}, + "semantic": {"Strong": "Reads very close to your search", "Solid": "Reads close to your search", "Light": "Loosely related to your search"}, +} + + +def _breakdown(factors) -> list[MatchDim]: + dims = [] + for k, v in factors.items(): + if v is None: + continue + sc = round(v * 100) + lvl = _level(sc) + dims.append(MatchDim(name=FACTOR_LABELS[k], score=sc, level=lvl, note=_DIM_NOTES.get(k, {}).get(lvl))) + dims.sort(key=lambda d: -d.score) + return dims[:4] + + +def _archetype(fit: str, score: int) -> str: + if fit == "fit": + return "The Strong Match" if score >= 85 else "The Solid Fit" + return "The Stretch Worth Taking" if score >= 68 else "The Long Shot" + + +def _one_line(dims: list[MatchDim]) -> str: + strong = [d.name.lower() for d in dims if d.level == "Strong"][:2] + light = [d.name.lower() for d in dims if d.level == "Light"][:1] + if strong and light: + return f"Your {' and '.join(strong)} land cleanly — {light[0]} is the gap to bridge." + if strong: + return f"Your {' and '.join(strong)} make this a clean match." + if light: + return f"A reach — {light[0]} is the stretch, but the rest lines up." + return "A balanced match worth a closer look." + + +def _coach(dims: list[MatchDim]) -> str: + strong = next((d.name.lower() for d in dims if d.level == "Strong"), None) + weak = min((d for d in dims if d.level == "Light"), key=lambda d: d.score, default=None) + if weak and strong: + return f"Lead with your {strong}; shore up {weak.name.lower()} before you apply and this reads as a clean fit." + if weak: + return f"Worth a look — close the {weak.name.lower()} gap to strengthen your case." + return "A strong fit — apply with confidence." + + +def _fit(p: UserProfile, U: float, factors) -> str: + sk = factors.get("skill") + if p.stretch == "reach": + return "stretch" if U < 0.8 else "fit" + strong = (sk is None or sk >= 0.5) and U >= 0.7 + return "fit" if strong else "stretch" + + +def _reason(job, dims: list[MatchDim], fit: str) -> str: + board = (job.get("note") or "").replace("Live from ", "") or "the boards" + top = [d.name.lower() for d in dims if d.level in ("Strong", "Solid")][:2] + lead = " · ".join(top) if top else ("a reach worth a look" if fit == "stretch" else "a solid fit") + return f"{lead.capitalize()} — fresh from {board}" + + +def _growth(dims: list[MatchDim], score: int) -> Growth | None: + light = [d for d in dims if d.level == "Light"] + if not light: + return None + w = min(light, key=lambda d: d.score) + return Growth(text=f"Close the {w.name.lower()} gap to lift this match", **{"from": score, "to": min(99, score + 8)}) + + +def _match(p: UserProfile, job: dict, U: float, factors) -> MatchResult: + score = round(U * 100) + dims = _breakdown(factors) + fit = _fit(p, U, factors) + return MatchResult( + score=score, fit=fit, archetype=_archetype(fit, score), one_line=_one_line(dims), + reason=_reason(job, dims, fit), breakdown=dims, growth=_growth(dims, score), coach_note=_coach(dims), + proofReady=(factors.get("skill") is not None and factors["skill"] >= 0.72), + factors={k: (round(v, 3) if v is not None else None) for k, v in factors.items()}, + ) + + +def rank(prefs: dict, user_context: dict | None, jobs: list[dict]) -> dict: + """Score ALL jobs (no filter, no cut). The bouncer is fired — Apify already guarantees city/title + precision, so location/seniority are *scoring* signals, not gates. This is the cheap SIFTER pass: + it ranks everything; `sift.sift()` then blends it with the embedding vibe to pick the top ~18 the + curator (Opus) reads. The final cut is Opus's call (or `select()` in the fallback path).""" + p = profile_from(prefs, user_context) + scored = [] + for job in jobs: + U, factors = utility(p, job) + if U is None: # no scorable factor at all → can't rank it + continue + scored.append((U, factors, job)) + scored.sort(key=lambda x: -x[0]) + + out = [] + for U, factors, job in scored: + m = _match(p, job, U, factors) + job["match"] = m.as_dict() + job["matchScore"] = m.score + out.append(job) + return {"opportunities": out, "ranked": len(out), "scanned": len(jobs)} + + +def select(jobs: list[dict], stretch: str = "balanced", floor: int = FLOOR, cap: int = CAP) -> list[dict]: + """Final shortlist cut — applied AFTER fusion (Phase 3), NEVER on the cheap white-box alone. + Cuts by the (fused) `matchScore` against the stretch threshold, with a floor (never empty) and + cap (~25). `jobs` must already be ranked best-first.""" + bar = round(SELECT.get(stretch, 0.70) * 100) + passing = [j for j in jobs if j.get("matchScore", 0) >= bar] + if len(passing) < floor: + passing = jobs[:floor] + return passing[:cap] diff --git a/app/engine/schema.py b/app/engine/schema.py new file mode 100644 index 0000000..c294bf4 --- /dev/null +++ b/app/engine/schema.py @@ -0,0 +1,59 @@ +"""The `match` block — the ONE contract between the engine and the card. + +Defined once here (Pydantic) and mirrored as a `JobMatch` TS interface on the frontend +`ScoutJob`. Every engine phase writes this exact shape; `scoutJobToMatchRole` reads it. +Keeping the breakdown/growth shapes identical to the card's existing `MatchDim`/`growth` +means the UI renders real numbers with **no shape change**. +""" +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + +Level = Literal["Strong", "Solid", "Light"] +Fit = Literal["fit", "stretch"] + + +class MatchDim(BaseModel): + """One row of the "how you match" breakdown — mirrors the frontend `MatchDim`. + `note` is a one-line, Gemini-toned observation for this dimension (the report-card voice).""" + name: str + score: int # 0–100 + level: Level + note: str | None = None + + +class Growth(BaseModel): + """The weakest *reachable* dim + the lift closing it buys — mirrors `MatchRole.growth`. + `from` is a Python keyword, so it's aliased (JSON stays `{text, from, to}`).""" + model_config = ConfigDict(populate_by_name=True) + text: str + from_: int = Field(alias="from") + to: int + + +class MatchResult(BaseModel): + """Attached to each surviving ScoutJob as `job["match"]` — the engine's report card. + Voice mirrors interview-service's Gemini video-analysis: a characterful archetype, an honest + balanced one-liner, per-dimension notes, and a warm coach note. Templated now → Opus later.""" + score: int # 0–100 — the real utility (NOT fetch order) + fit: Fit + archetype: str | None = None # characterful label, e.g. "The Stretch Worth Taking" + one_line: str | None = None # balanced headline verdict (strength AND gap) + reason: str # short "why picked" + breakdown: list[MatchDim] = [] # per-dimension dims, each with a `note` + growth: Growth | None = None + coach_note: str | None = None # warm, actionable closing line + proofReady: bool = False # user is Strong on the job's key skill + factors: dict[str, float | None] = {} # debug/observability (per-factor φ; None = absent) + + def as_dict(self) -> dict: + """Serialize for the ScoutJob payload — by alias so growth emits `from`/`to`.""" + return self.model_dump(by_alias=True) + + +def attach_match(job: dict, result: MatchResult) -> dict: + """Set `job['match']` from a MatchResult (in place) and return the job.""" + job["match"] = result.as_dict() + return job diff --git a/app/engine/search.py b/app/engine/search.py index 2f40526..08bf214 100644 --- a/app/engine/search.py +++ b/app/engine/search.py @@ -47,16 +47,19 @@ BOARDS = { } -async def _fetch_board(key: str, prefs: dict): +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)) + items = await run_actor(actor, build(prefs), cache_mode=cache_mode) jobs = [sj for it in items if (sj := norm(it))] return key, jobs -async def run_sweep(prefs: dict) -> dict: +async def run_sweep(prefs: dict, *, fresh: bool = False) -> dict: + # fresh=True forces a live fetch + cache overwrite (dev "fresh search" toggle); + # otherwise honor the configured cache mode (dev replays cached results for $0). + cache_mode = "refresh" if fresh else None enabled = [b.strip() for b in get_settings().BOARDS_ENABLED.split(",") if b.strip() in BOARDS] - results = await asyncio.gather(*(_fetch_board(k, prefs) for k in enabled), return_exceptions=True) + results = await asyncio.gather(*(_fetch_board(k, prefs, cache_mode) for k in enabled), return_exceptions=True) merged: list[dict] = [] seen: set[str] = set() @@ -74,10 +77,7 @@ async def run_sweep(prefs: dict) -> dict: seen.add(dk) merged.append(j) - # placeholder ranking (no engine yet): a descending spread so cards aren't all the same % - for i, j in enumerate(merged): - j["matchScore"] = max(55, 95 - i) - + # No ranking here — the engine (app/engine/rank.py) scores + selects after the sweep. return {"opportunities": merged, "sources": sources} diff --git a/app/engine/sift.py b/app/engine/sift.py new file mode 100644 index 0000000..0ec09f2 --- /dev/null +++ b/app/engine/sift.py @@ -0,0 +1,43 @@ +"""The sift — combine the two cheap rankers (white-box ‖ embedding vibe) by weighted-average +RANK fusion and return the **top-K candidate pool** for the curator (Opus). + +This is NOT the final cut — it only decides which ~18 jobs the expensive brain reads (retrieve cheap, +rerank expensive). Each returned job already carries its white-box `match` block (the templated +fallback the curator's output later replaces). +""" +from __future__ import annotations + +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 + + +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 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.""" + 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: + 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 + 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" + else: + fused = _ranks(wb) + mode = "whitebox-only" + + top = sorted(ranked, key=lambda j: -fused[j["id"]])[:k] + return top, {"mode": mode, "scored": len(ranked), "k": len(top)} diff --git a/app/engine/skills.py b/app/engine/skills.py new file mode 100644 index 0000000..26030d3 --- /dev/null +++ b/app/engine/skills.py @@ -0,0 +1,93 @@ +"""Skill normalization + coverage-aware overlap — the core accuracy lever (ENGINE_DESIGN §3.2). + +v1: synonym canonicalization + lowercase/punctuation fold (no ESCO/KG yet — upgrade later). +**FLOOR-FREE** (§3.3): `overlap` returns a real score ONLY when both sides have skills; otherwise +`None` (the factor is *absent* → its weight redistributes — never a neutral constant like 60). +""" +from __future__ import annotations + +import re + +# Seed synonym map → canonical. Extend over time; swap to ESCO + KG traversal later. +_SYNONYMS = { + "k8s": "kubernetes", "kube": "kubernetes", + "js": "javascript", "ts": "typescript", "react.js": "react", "reactjs": "react", + "node.js": "node", "nodejs": "node", "next.js": "nextjs", "nextjs": "nextjs", + "ml": "machine learning", "a.i.": "artificial intelligence", "ai": "artificial intelligence", + "dl": "deep learning", "nlp": "natural language processing", "genai": "generative ai", + "llm": "large language models", "llms": "large language models", + "pm": "product management", "product manager": "product management", "apm": "product management", + "postgres": "postgresql", "pg": "postgresql", "psql": "postgresql", + "gcp": "google cloud", "aws": "amazon web services", + "ds": "data science", "da": "data analytics", "ba": "business analysis", + "ux": "user experience", "ui": "user interface", "qa": "quality assurance", + "sde": "software engineering", "swe": "software engineering", "fe": "frontend", "be": "backend", + "fullstack": "full stack", "full-stack": "full stack", "rest": "rest api", "restful": "rest api", + "ci/cd": "cicd", "scrum": "agile", "go": "golang", "py": "python", "k8": "kubernetes", +} + +_LEVEL_SCORE = {"Strong": 90, "Solid": 65, "Light": 35} + + +def norm(s: str) -> str: + s = re.sub(r"\s+", " ", (s or "").strip().lower()) + if s in _SYNONYMS: # raw lookup first — keys keep punctuation (react.js, k8s, ci/cd) + return _SYNONYMS[s] + s = re.sub(r"\s+", " ", re.sub(r"[._/]+", " ", s)).strip() # then fold punctuation + return _SYNONYMS.get(s, s) + + +def canon(skills) -> set[str]: + out: set[str] = set() + for s in skills or []: + if not s: + continue + n = norm(str(s)) + if n: + out.add(n) + return out + + +def _tok(s: str) -> set[str]: + return {t for t in s.split() if len(t) > 2} + + +def _level(job_skill: str, user_canon: set[str], user_tokens: set[str]) -> str: + if job_skill in user_canon: + return "Strong" + jt = _tok(job_skill) + if jt and (jt & user_tokens): + return "Solid" + return "Light" + + +def overlap(user_skills, job_skills) -> dict | None: + """Coverage-aware skill F1 (precision & recall each ≤ 1). Returns None when EITHER side is + empty (absent factor — never a constant floor).""" + user = canon(user_skills) + job = canon(job_skills) + if not user or not job: + return None + user_tokens: set[str] = set().union(*(_tok(u) for u in user)) if user else set() + job_tokens: set[str] = set().union(*(_tok(j) for j in job)) if job else set() + matched_job = [j for j in job if _level(j, user, user_tokens) != "Light"] # job needs the user meets + matched_user = [u for u in user if u in job or (_tok(u) & job_tokens)] # user skills relevant here + recall = len(matched_job) / len(job) # ≤ 1 + precision = len(matched_user) / len(user) # ≤ 1 + f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) else 0.0 + return { + "score": f1, "recall": recall, + "matched": sorted(matched_job), + "missing": sorted(j for j in job if j not in matched_job), + } + + +def dims(user_skills, job_skills, k: int = 4) -> list[dict]: + """Per-skill breakdown rows (job's top skills marked Strong/Solid/Light vs the user).""" + user = canon(user_skills) + user_tokens: set[str] = set().union(*(_tok(u) for u in user)) if user else set() + rows = [] + for j in list(canon(job_skills))[:k]: + lvl = _level(j, user, user_tokens) if user else "Light" + rows.append({"name": j.title(), "score": _LEVEL_SCORE[lvl], "level": lvl}) + return rows diff --git a/app/main.py b/app/main.py index 67f9d44..2476d7d 100644 --- a/app/main.py +++ b/app/main.py @@ -25,6 +25,8 @@ _worker = RedisStreamWorker(redis_url=_settings.ORCHESTRATOR_REDIS_URL or _setti @asynccontextmanager async def lifespan(_: FastAPI): + from app.db.repo import init_db + await init_db() # create the feed table if DATABASE_URL is set (best-effort) await _worker.start() try: yield diff --git a/docker-compose.yml b/docker-compose.yml index 21230dc..b15bf86 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -7,4 +7,32 @@ services: - "8006:8000" env_file: - .env + 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 + depends_on: + postgres: + condition: service_healthy restart: unless-stopped + + # Durable per-user feed store (fixes "matches disappear on navigation"). Dedicated, not shared. + postgres: + image: postgres:16-alpine + container_name: matchmaking-postgres + environment: + POSTGRES_USER: matchmaking + POSTGRES_PASSWORD: matchmaking + POSTGRES_DB: matchmaking + ports: + - "5446:5432" # host-dev uvicorn connects via localhost:5446 + volumes: + - mmpgdata:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U matchmaking"] + interval: 5s + timeout: 3s + retries: 10 + restart: unless-stopped + +volumes: + mmpgdata: diff --git a/requirements.txt b/requirements.txt index c23f565..42d51a9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,3 +7,7 @@ redis>=5.0 # engine/auto-apply deps land in later slices: # openai>=1.40 # GPT-5.4 rerank/explain + tailoring # apify-client>=1.7 # on-demand board fetch (or call run-sync via httpx) +openai>=1.40 +numpy +sqlalchemy[asyncio]>=2.0 +asyncpg diff --git a/research/docs/ENGINE_BUILD_ANALYSIS.md b/research/docs/ENGINE_BUILD_ANALYSIS.md new file mode 100644 index 0000000..cdfbac7 --- /dev/null +++ b/research/docs/ENGINE_BUILD_ANALYSIS.md @@ -0,0 +1,171 @@ +# Engine Build Analysis — from 90 raw cards to a ranked, explained shortlist + +> Written 2026-06-19. Companion to `ENGINE_DESIGN.md` (the locked design) + `SIGNAL_AUDIT_V2.md` +> (the signal inventory). This doc is the **build-readiness analysis**: what we have *today*, the gap, +> and the smallest engine that turns the live multi-board sweep into "only the recommended ones, with +> aligned stats." It translates the design into a phased, buildable plan against the real data surface. + +--- + +## 1. The gap (current vs target) + +**Today (Slice 1a):** fetch ~90 live jobs → `search.run_sweep` assigns a **placeholder** score +(`matchScore = max(55, 95-i)`, i.e. rank = fetch order) → **all 90** shown. Critically: +- `user_context` (the user's resume/skills/experience) is **fetched by the orchestrator but ignored** + by `handle_run_search` — so nothing is actually matched *to the user*. +- The card's "stats" are **fabricated in the frontend**: `scoutJobToMatchRole()` builds the QX + `breakdown` from `required_skills[:3]` + fixed deltas `[10,-7,-32]`, derives `growth` from that, sets + `reason = "Live from Naukri"`, and `proofReady = matchScore≥84`. **None of it reflects the user.** + +**Target:** fetch ~90 → **engine** (filter + rank *against the user* + select) → **top ~20–30** → +each card carries a **real** match %, a real "why picked", a real skill breakdown, and a FIT/STRETCH tag. + +Two asks map to two engine outputs: +- *"only show the ones recommended"* → **hard filter + score + top-N selection**. +- *"stats more aligned"* → **move the breakdown/why/growth from fabricated-frontend to real-backend**, + and have the card render what the engine emits. + +--- + +## 2. The input surface we actually have NOW + +### Job side — per fetched `ScoutJob` (already extracted by the normalizers) +`title` · `required_skills` (full list in `details.skills`) · `seniority_level` · `location_city` + +`location_mode` · `salary_lpa` / `payLabel` · `industry` · `role_category` · `employment_type` · +`description` (clean text) · `note` (board) · `offsite_apply` · freshness (**all on-demand = fresh by +construction**; we already drop `isExpired`). + +### User side +- **Configure prefs (`ScoutPrefs`) — always present:** `title`, `role[]`, `location[]`, `workMode[]`, + `experience`/seniority, `industry[]`, `years`, `targetComp`, `companyStage`, `availability`, + `dealBreakers[]`, `priorities[]` (rank), `stretch` (safe/balanced/reach). +- **`user_context` (orchestrator-assembled, currently ignored) — present when the user has a profile:** + `skills[]` (resume + live LinkedIn, deduped), `current_role`, `experience_history`, `education`, + `linkedin_headline/summary/experience`. Plus the **QScore quotient vector** (qscore-service). + +**Coverage reality:** Configure prefs are guaranteed; `user_context` may be sparse/empty (the "No +resume" state). The engine **must** degrade gracefully (design §3.3 coverage-renormalization) — rank on +what's present, never fabricate, never collapse to a neutral 60 (the v1 bug). + +--- + +## 3. The design (recap) vs what's buildable first + +`ENGINE_DESIGN.md` §3 cascade: ① normalize → ② hard-filter → ③ white-box utility → ④ rank-fusion → +⑤ LLM rerank+explain → ⑥ output → (log → learned ranker). + +The **full** vision needs ESCO taxonomy + depth-2 KG, contrastive skill embeddings, GPT-5.4 rerank, and +a learned pairwise ranker. **None of that is needed to ship the core value.** Stages ② + ③ + ⑥ — +deterministic, white-box, no LLM, no taxonomy service — already deliver *filter + rank + select + +explain* against the current data. That is the first engine; the rest are precision upgrades. + +--- + +## 4. Phased build + +### Phase 0 — Deterministic white-box engine (`app/engine/rank.py`), no LLM, no `user_context` required +The shippable core. Filters 90 → top-N, scores, emits real stats. Works on Configure prefs alone, gets +sharper when `user_context` lands (Phase 1). + +**② Hard filter (deterministic gate — the design's point: embeddings can't do this):** +- **Location / work-mode fit** — port `_score_location_fit` + `_location_filter_blocks_match` from the + *old* `matchmaking/app/engine/core.py` (the audit says it did this better): remote→100 (0 if seeker + onsite-only); else city/country substring vs preferred locations; unknown→neutral 60; **gate** blocks + no-fit unless prefs are "Anywhere". +- **Seniority band** — drop jobs far outside the user's level (e.g. a 2-yr seeker vs a Lead role), + governed by `stretch` (reach allows +1 level). +- **Deal-breakers** — exclude on stated breakers where detectable (on-call, on-site, etc.). +- **Freshness** — already fresh; drop any `isExpired`/dead. + +**③ White-box utility, coverage-renormalized** — `U = Σ_present c_f·w_f·φ_f / Σ_present c_f·w_f`: +| factor | prior w | φ (0–1) from | +|---|---|---| +| skill_match | .35 | overlap(user.skills, job.required_skills), synonym-normalized; required>preferred | +| experience | .25 | user `years`/seniority vs job `seniority_level` band | +| location | .15 | ported location-fit score | +| salary | .10 | job `salary_lpa` vs `targetComp` band — **present-only** (most are "Not disclosed" → weight redistributes) | +| industry/role | .05+.05 | job `industry`/`role_category` vs prefs | +| semantic | .10 | cheap title/description keyword overlap (NOT embeddings in v0) | + +Sparse profile (no `skills`) → skill weight redistributes to location/experience/recency. Never penalize absence. + +**⑤′ Selection (the "only recommended" lever)** — sort by U, cut by `stretch`: +`safe`→ U≥0.85 · `balanced`→ U≥0.70 (+ a few stretch) · `reach`→ U≥0.55. **Floor of ~8–10 so the deck +is never empty** (relax threshold if too few pass); **cap ~25–30** (never the full 90). + +**FIT vs STRETCH (§3.5):** FIT = skill-gap≈0 (score current overlap, mild over-qual penalty); +STRETCH = one level up / adjacent, scored by **reachability** (fraction of required skills the user has +or is one hop from). Tag each card; the `stretch` pref sets how many STRETCH cards ride along. + +**⑥ Aligned stats (emitted by the engine, rendered by the card):** +- `match` = round(U×100) — **real**, not fetch-order. +- `breakdown[]` = the real top dims (matched skills Strong / partial Solid / missing Light) — **from + user↔job skill overlap**, replacing the fabricated `[10,-7,-32]`. +- `growth` = the weakest *reachable* dim + the lift it buys. +- `reason` = a real templated "why" ("Matches 6/8 must-have skills · Senior fit · fresh from Naukri"). +- `proofReady` = user is Strong on the job's key skill (real). +- `fit | stretch` tag. + +### Phase 1 — Wire `user_context` (the accuracy unlock) +`handle_run_search` reads `params["user_context"]` (already delivered by the worker) → feeds +`skills/current_role/experience_history/education` + QScore quotient into ③. QScore/assessments are +**boosters** (present-only, coverage-penalized), never a penalty when absent. This is the user-arm the +whole product is about — Phase 0 ranks on prefs; Phase 1 ranks on *who the user actually is*. + +### Phase 2 — LLM rerank + explanation (GPT-5.4 on top ~20) +Calibrated 0–1 fit + a genuine 1-line "why" + growth nudge; replaces the templated reason on the top +slice. Log soft-scores as cold-start supervision. **Only the top ~20, cached per (role, location)** — +never LLM all 90 (cost/latency). + +### Phase 3 — Learned ranker +Accrue SAVE/DISMISS/APPLY labels (the `record_feedback` action) → pairwise/listwise ranker → shift +fusion weight off the expert priors. + +--- + +## 5. The stats-alignment refactor (a contract change) + +Right now the card's "intelligence" is **invented in `scoutJobToMatchRole`**. For stats to be honest, +the engine must own them and the ScoutJob must carry a `match` block: + +``` +match: { + score: number, // U×100 + fit: "fit" | "stretch", + reason: string, // real why-picked + breakdown: [{name, score, level}], // real user↔job skill dims + growth?: {text, from, to}, + proofReady: boolean, +} +``` +The frontend `scoutJobToMatchRole` then **reads `job.match.*`** instead of fabricating it. Net: delete +the fake-breakdown logic; the card becomes a pure renderer of engine output. (Same modal/stats UI — it +just gets real numbers.) + +--- + +## 6. Key decisions & risks +- **Skill normalization without ESCO (v0):** a **seed synonym map** (k8s→Kubernetes, React.js→React, + ML→Machine Learning, JS→JavaScript…) + India tech terms + lowercase/punctuation fold. Beats exact + match; upgrade to ESCO+KG+contrastive later. This is the single biggest accuracy lever (§3.2). +- **Sparse profiles are the norm, not the exception** — coverage-renormalization is load-bearing, not + a nicety. Test the "No resume" path first. +- **Never an empty deck** — selection has a floor; if few pass, lower the bar and label them honestly + ("a few reaches"). +- **Don't double-count QScore** (already digests interview/roleplay) — §2 of the audit. +- **LLM only on the top slice, cached** — the §3 "retrieve cheap, rerank expensive" discipline. +- **Freshness is free here** — on-demand fetch means every shown listing is seconds old (§6.1). + +--- + +## 7. Recommendation (smallest first step) + +Build **Phase 0 + Phase 1 together** as `app/engine/rank.py` — a deterministic, coverage-aware, +white-box ranker that consumes `user_context`, filters 90 → top ~25, and emits the real `match` block. +No LLM, no ESCO service, no new infra; it ships *both* asks ("only recommended" + "aligned stats"). +Then add Phase 2 (LLM rerank/explain) once the deterministic core + the frontend `match`-contract are +in. Phase 3 (learned) follows the live save/apply loop. + +**Open scoping questions for the build:** (a) wire `user_context` now or rank on Configure prefs first? +(b) how aggressive should default selection be (target deck size)? (c) include the GPT-5.4 rerank in v1 +or ship deterministic-only first? (d) seed-synonym map now vs minimal exact+fold to start? diff --git a/tests/test_contract.py b/tests/test_contract.py new file mode 100644 index 0000000..b2128da --- /dev/null +++ b/tests/test_contract.py @@ -0,0 +1,50 @@ +"""Phase 0 — the `match` contract test. + +The engine's `match` block must (a) validate against the Pydantic model and (b) serialize +to EXACTLY the keys the frontend `scoutJobToMatchRole` reads (scout/lib/jobMatching.ts → +`JobMatch`). If either drifts, backend ⇄ frontend have silently diverged — this test fails. +""" +import pytest +from pydantic import ValidationError + +from app.engine.schema import Growth, MatchDim, MatchResult + +# The exact field sets the frontend reads. Keep in lockstep with `JobMatch` / `MatchDim`. +FRONTEND_MATCH_KEYS = {"score", "fit", "archetype", "one_line", "reason", "breakdown", "growth", + "coach_note", "proofReady", "factors"} +FRONTEND_DIM_KEYS = {"name", "score", "level", "note"} +FRONTEND_GROWTH_KEYS = {"text", "from", "to"} + + +def _golden() -> MatchResult: + return MatchResult( + score=88, fit="stretch", reason="Matches 5/7 skills · Senior fit", + breakdown=[MatchDim(name="Product Management", score=92, level="Strong"), + MatchDim(name="SQL", score=48, level="Light")], + growth=Growth(text="Add an SQL proof point", **{"from": 88, "to": 95}), + proofReady=True, factors={"skill": 0.71, "experience": 0.9, "salary": None}, + ) + + +def test_match_block_keys_align_with_frontend(): + d = _golden().as_dict() + assert set(d) == FRONTEND_MATCH_KEYS, f"match keys drifted: {set(d) ^ FRONTEND_MATCH_KEYS}" + assert set(d["breakdown"][0]) == FRONTEND_DIM_KEYS + assert set(d["growth"]) == FRONTEND_GROWTH_KEYS + # the `from` keyword must serialize under its alias, not `from_` + assert d["growth"]["from"] == 88 and d["growth"]["to"] == 95 + + +def test_constrained_enums_reject_bad_values(): + with pytest.raises(ValidationError): + MatchResult(score=1, fit="maybe", reason="x") # fit ∉ {fit, stretch} + with pytest.raises(ValidationError): + MatchDim(name="x", score=1, level="Great") # level ∉ {Strong, Solid, Light} + + +def test_minimal_match_is_valid_and_floor_free_shaped(): + d = MatchResult(score=50, fit="fit", reason="ok").as_dict() + assert d["growth"] is None and d["breakdown"] == [] and d["factors"] == {} + # factors may carry None (an *absent* factor) — that's the coverage-aware contract, not a floor + d2 = MatchResult(score=50, fit="fit", reason="ok", factors={"salary": None}).as_dict() + assert d2["factors"]["salary"] is None diff --git a/tests/test_coverage.py b/tests/test_coverage.py new file mode 100644 index 0000000..000e717 --- /dev/null +++ b/tests/test_coverage.py @@ -0,0 +1,49 @@ +"""Phase 1 — the floor-free + coverage-aware invariants (the v1-collapse guardrails). + +This is the test that would have caught the old engine: every factor is float-or-`None` +(no constant floors), and a sparse profile must NOT collapse all jobs to one neutral score. +""" +from app.engine import rank as R + +_JOBS = [ + {"title": "Product Manager", "seniority_level": "senior", "location_city": "New Delhi", + "location_country": "India", "salary_lpa": 30, + "details": {"skills": ["Product Management", "SQL"], "industry": "Fintech", + "description": "product manager roadmap stakeholders"}}, + {"title": "Sales Lead", "seniority_level": "lead", "location_city": "Mumbai", + "location_country": "India", "salary_lpa": None, + "details": {"skills": ["Sales", "CRM"], "industry": "SaaS", "description": "sales quota targets"}}, + {"title": "Product Manager", "seniority_level": "mid", "location_mode": "remote", + "location_city": "Remote", "details": {"skills": ["Product Management"], "description": "product manager"}}, + {"title": "Designer", "seniority_level": None, "location_city": None, "salary_lpa": None, "details": {}}, +] + + +def test_every_factor_returns_float_or_none(): + p_full = R.profile_from({"title": "Product Manager", "location": ["New Delhi · India"], + "industry": ["Fintech"], "experience": ["Senior"], "years": 9, + "role": ["Product"], "targetComp": "₹25–30L"}) + p_sparse = R.profile_from({"title": "Product Manager"}) + for p in (p_full, p_sparse): + for job in _JOBS: + for name, fn in R.FACTORS.items(): + v = fn(p, job) + assert v is None or (isinstance(v, float) and 0.0 <= v <= 1.0), f"{name} → {v!r}" + + +def test_sparse_profile_does_not_collapse(): + # The v1 bug: sparse profile → everyone ~60. Assert real score variance across varied jobs. + p = R.profile_from({"title": "Product Manager", "location": ["New Delhi · India"]}) + us = [u for u in (R.utility(p, j)[0] for j in _JOBS) if u is not None] + assert len(us) >= 2 and (max(us) - min(us)) > 0.1, f"collapsed to a flat band: {us}" + + +def test_absent_factor_redistributes_weight_not_floors(): + # A job missing salary/industry/skills must still score on what IS present (no 60 injected). + p = R.profile_from({"title": "Product Manager", "location": ["New Delhi · India"]}) + bare = {"title": "Product Manager", "location_city": "New Delhi", "location_country": "India", + "details": {"description": "product manager"}} + U, factors = R.utility(p, bare) + assert U is not None + assert factors["salary"] is None and factors["skill"] is None # absent + assert factors["location"] == 1.0 # present, real diff --git a/tests/test_curate.py b/tests/test_curate.py new file mode 100644 index 0000000..b7c7e5a --- /dev/null +++ b/tests/test_curate.py @@ -0,0 +1,25 @@ +"""A2 — curator: tolerant JSON parse, contract mapping, and the disabled→None safety net. +(Offline — the live Opus call is exercised in the cached e2e verification.)""" +from app.engine import curate as C + + +def test_parse_json_strips_fences_and_prose(): + assert C._parse_json('```json\n{"kept":[]}\n```') == {"kept": []} + 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}, + }) + 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" + assert d["growth"]["from"] == 78 and d["growth"]["to"] == 85 and d["proofReady"] is True + + +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 diff --git a/tests/test_rank.py b/tests/test_rank.py new file mode 100644 index 0000000..c2ef179 --- /dev/null +++ b/tests/test_rank.py @@ -0,0 +1,75 @@ +"""Phase 1 — factor primitives + end-to-end shortlist on CACHED data ($0).""" +import glob +import json + +import pytest + +from app.engine import normalize as N +from app.engine import rank as R + +_NORM = {"blackfalcondata": N.naukri_feed_to_scoutjob, "shahidirfan": N.foundit_to_scoutjob, + "harvestapi": N.linkedin_to_scoutjob} + + +def _cached_jobs() -> list[dict]: + jobs = [] + for f in glob.glob(".apify_cache/*.json"): + fn = _NORM.get(f.split("/")[-1].split("_")[0]) + if not fn: + continue + for it in json.load(open(f)): + sj = fn(it) + if sj: + jobs.append(sj) + return jobs + + +# ── factor primitives ── +def test_location_gate_and_tiers(): + p = R.profile_from({"title": "PM", "location": ["New Delhi · India"]}) + assert R.f_location(p, {"location_city": "New Delhi", "location_country": "India"}) == 1.0 + assert R.f_location(p, {"location_city": "Bengaluru", "location_country": "India"}) == 0.6 + assert R.f_location(p, {"location_city": "London", "location_country": "UK"}) == 0.0 + assert R.f_location(p, {"location_mode": "remote"}) == 1.0 + p_open = R.profile_from({"title": "PM"}) + assert R.f_location(p_open, {"location_city": "X", "location_country": "Y"}) is None # Anywhere → absent + + +def test_experience_band(): + p = R.profile_from({"title": "PM", "years": 5}) + assert R.f_experience(p, {"seniority_level": "mid"}) == 1.0 # 5 in [3,8] + assert R.f_experience(p, {"seniority_level": "lead"}) < 1.0 # under-qualified + assert R.f_experience(p, {"seniority_level": None}) is None # absent + + +def test_e2e_ranks_all_gate_passing_no_early_cut(): + jobs = _cached_jobs() + if len(jobs) < 30: + pytest.skip("no cached sweep data — run a search first") + prefs = {"title": "Product Manager", "role": ["Product"], "location": ["New Delhi · India"], + "experience": ["Mid-level"], "stretch": "balanced"} + res = R.rank(prefs, None, jobs) + opp = res["opportunities"] + # the cheap white-box does NOT cut to 25 — it ranks ALL gate-passing jobs (cut is Phase 3) + assert len(opp) >= 1 + assert res["ranked"] == len(opp) and res["scanned"] == len(jobs) + scores = [j["matchScore"] for j in opp] + assert scores == sorted(scores, reverse=True) # ranked best-first + for j in opp: + m = j["match"] + assert m["fit"] in ("fit", "stretch") and isinstance(m["breakdown"], list) + assert m["score"] == j["matchScore"] + # determinism + res2 = R.rank(prefs, None, _cached_jobs()) + assert [j["id"] for j in res2["opportunities"]] == [j["id"] for j in opp] + + +def test_select_is_separate_and_capped(): + # the final cut (Phase 3) — caps at ~25, floor so never empty, by fused matchScore + ranked = [{"id": str(i), "matchScore": 100 - i} for i in range(80)] + sel = R.select(ranked, "balanced") + assert 1 <= len(sel) <= R.CAP + assert [j["matchScore"] for j in sel] == sorted((j["matchScore"] for j in sel), reverse=True) + # floor: even if nothing clears the bar, return at least `floor` + low = [{"id": str(i), "matchScore": 10} for i in range(40)] + assert len(R.select(low, "safe")) == R.FLOOR diff --git a/tests/test_sift.py b/tests/test_sift.py new file mode 100644 index 0000000..89e4e30 --- /dev/null +++ b/tests/test_sift.py @@ -0,0 +1,25 @@ +"""A1 — sift: rank-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_sift_degrades_to_whitebox_when_no_vibe(monkeypatch): + monkeypatch.setattr(S._embed, "vibe_scores", lambda *a, **k: None) # embeddings unavailable + jobs = [{"id": str(i), "title": "Product Manager", "organization": "X", + "location_city": "New Delhi", "location_country": "India", + "details": {"description": "product manager roadmap"}} for i in range(30)] + prefs = {"title": "Product Manager", "location": ["New Delhi · India"]} + top, dbg = S.sift(prefs, None, jobs, k=18) + assert dbg["mode"] == "whitebox-only" and len(top) == 18 + assert all("match" in j for j in top) # white-box match block rides along + + +def test_sift_empty_input(): + top, dbg = S.sift({"title": "PM"}, None, [], k=18) + assert top == [] and dbg["mode"] == "empty" diff --git a/tests/test_skills.py b/tests/test_skills.py new file mode 100644 index 0000000..22c4c12 --- /dev/null +++ b/tests/test_skills.py @@ -0,0 +1,24 @@ +"""Phase 1 — skill normalization + coverage-aware (floor-free) overlap.""" +from app.engine import skills as S + + +def test_synonym_and_fold(): + assert S.norm("React.js") == "react" + assert S.norm("K8s") == "kubernetes" + assert S.norm(" ML ") == "machine learning" + c = S.canon(["ML", "Python", "react.js"]) + assert {"machine learning", "python", "react"} <= c + + +def test_overlap_is_floor_free(): + assert S.overlap([], ["python"]) is None # empty user → absent (NOT 0 or 60) + assert S.overlap(["python"], []) is None # empty job → absent + o = S.overlap(["python", "sql", "django"], ["Python", "Django", "AWS"]) + assert o is not None and 0.0 < o["score"] <= 1.0 + assert "python" in o["matched"] and "amazon web services" in o["missing"] + + +def test_dims_levels(): + rows = {r["name"]: r["level"] for r in S.dims(["product management"], ["Product Management", "SQL"], k=2)} + assert rows["Product Management"] == "Strong" + assert rows["Sql"] == "Light"