Merge main into staging

This commit is contained in:
-Puter
2026-07-08 20:55:51 +05:30
32 changed files with 7825 additions and 181 deletions

View File

@@ -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"]

39
alembic.ini Normal file
View File

@@ -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

67
alembic/env.py Normal file
View File

@@ -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()

24
alembic/script.py.mako Normal file
View File

@@ -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"}

View File

@@ -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")

View File

@@ -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")

View File

@@ -102,36 +102,53 @@ 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…"])
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 +167,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 [])])
@@ -174,6 +185,27 @@ class MatchmakingAgentSession:
result["shortlisted"] = len(result["opportunities"])
await self.push("agent_data", action="search_complete", data=result)
# 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."""
from app.engine import suggest

View File

@@ -40,7 +40,13 @@ class Settings(BaseSettings):
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,16 +58,36 @@ 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.
QSCORE_BASE_URL: str = "http://localhost:8004"

View File

@@ -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

View File

@@ -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__)
@@ -346,7 +346,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 +363,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)

View File

@@ -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 = {

View File

@@ -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):

View File

@@ -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,175 @@ 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, *, 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"')."""
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) -> 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
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")
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 {}
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

View File

@@ -42,24 +42,32 @@ def _embedding_vectors(s, texts: list[str]) -> list[list[float]] | 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)

View File

@@ -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)
}

View File

@@ -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

66
app/engine/rubric.py Normal file
View File

@@ -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)

View File

@@ -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,7 +117,17 @@ 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()

View File

@@ -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]

View File

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

79
tests/_seed_regression.py Normal file
View File

@@ -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())

1
tests/fixtures/board_samples.json vendored Normal file

File diff suppressed because one or more lines are too long

5787
tests/fixtures/regression_set.json vendored Normal file

File diff suppressed because it is too large Load Diff

93
tests/run_regression.py Normal file
View File

@@ -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)

98
tests/test_contracts.py Normal file
View File

@@ -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}"

View File

@@ -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

View File

@@ -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"

41
tests/test_pool.py Normal file
View File

@@ -0,0 +1,41 @@
"""Warm-pool gate — the reorder/floor decision that keeps the shelf stocked without over-fetching."""
from app.engine.search import pool_decision
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

65
tests/test_rubric.py Normal file
View File

@@ -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:]))

View File

@@ -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)"

View File

@@ -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):

View File

@@ -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