Engine: signal-weighted India stack + cursorless pool + honest-card hardening

Boards (signal-weighted, ~150/sweep):
- Add TimesJobs (shahidirfan) — richest board: skills+salary+experience+offsite. Build/normalize +
  register; legacy city alias (New Delhi→Delhi, Bengaluru→Bangalore, Gurugram→Gurgaon) — verified live
  it returns 0 for "New Delhi", 35 for "Delhi".
- Swap LinkedIn harvestapi → curious_coder (structured industry/jobFunction/applicants); harvestapi kept
  registered as linkedin_v1 fallback.
- Drop Indeed from enabled (0% skills, redundant). Budgets: Foundit 40, TimesJobs 35, Naukri 30,
  LinkedIn 30, WorkIndia 15.

Pool — cursorless freshness + safety:
- REMOVE pagination/cursors: curious_coder only honors start=0 (start>0 → empty), proven; the rest are
  date feeds. Freshness now = boards' date-sort + the per-user seen-net + pool_save id-dedup.
- Exhaustion guard: a refill adding < POOL_MIN_NEW_PER_REFILL new jobs flags the (user,query) exhausted →
  the gate relaxes the floor instead of block-fetching dupes (makes the 100/90 floor safe on niche queries).
- Storage cap: pool_save trims beyond POOL_MAX_PER_QUERY freshest (kills DB swell). 72h TTL verified.
- Thresholds 100/90 (deep pool, pay-for-volume).

Scoring / cards:
- Evidence-based fallback prose: when Haiku's cards stage gives nothing, the card uses Opus's REAL
  dimension notes (not a generic "Strong on X") + coverage logging + generic salvage parser.
- "Skills & requirements" → "Skills fit" (consistent dimension labels).

Bug fixes:
- mark_seen dedups ids (was CardinalityViolationError on a duplicate id in one batch).

Tests: deep stack contracts + committed board-sample fixture (no cache-pollution flakiness) + offline
e2e + opt-in live e2e. 64 pass.

tests/test_e2e_integration.py

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

tests/test_pool.py

@@ -1,26 +1,37 @@
 """Warm-pool gate — the reorder/floor decision that keeps the shelf stocked without over-fetching."""
 from app.engine.search import pool_decision
 
-FLOOR, REORDER = 70, 80   # the configured thresholds (POOL_SAFETY_FLOOR / POOL_REORDER_AT)
+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(80, False, FLOOR, REORDER) == "pool"     # exactly at reorder → still no fetch
+    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(79, False, FLOOR, REORDER) == "pool+refill"   # below reorder → top up in bg
-    assert pool_decision(70, False, FLOOR, REORDER) == "pool+refill"   # at the floor → still serve, 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(69, False, FLOOR, REORDER) == "sweep"    # under the hard floor → blocking fetch
+    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():

tests/test_stack_contracts.py

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