"""Step 11 multi-band explainability — bands, FactEngine matchers, NLG composition, tier gap-gating, artifact.""" from __future__ import annotations from pathlib import Path import orjson import polars as pl from config import ReasoningSettings, Settings from jd import JDSpec from rank import add_explanation from rank.explain import _clean_note, _is_grounded, _matched_skills, _title_match from rank.explain.bands import experience_fit, location_tier, notice_band, recency_band, skill_coverage from rank.explain.engine import matched_education, matched_industries, title_fit from rank.explain.signals import signal_bands _SPEC = JDSpec( must_have_skills=["Python", "embeddings", "ranking"], nice_to_have=["RAG", "HNSW"], job_titles=["AI Engineer", "Search Engineer"], locations_preferred=["Pune", "Noida"], locations_acceptable=["Bangalore"], years_min=5, years_max=9, target_industries=["Fintech", "Healthcare"], education_fields=["Computer Science"], ) def _row(**over: object) -> dict[str, object]: base = { "candidate_id": "C1", "current_title": "Senior AI Engineer", "years_of_experience": 7.4, "experience_band": "senior", "location": "Pune, MH", "country": "India", "current_industry": "Fintech", "current_company": "Acme", "flat_skills": ["Python", "embeddings", "ranking"], "flat_career_industries": ["Fintech"], "flat_certifications": [], "education": [{"field_of_study": "Computer Science"}], "days_since_active": 10, "notice_period_days": 15, "is_available_soon": True, "open_to_work_flag": True, "willing_to_relocate": False, "preferred_work_mode": "hybrid", "verification_count": 3, "recruiter_response_rate": 0.8, "saved_by_recruiters_30d": 6, "profile_views_received_30d": 25, "interview_completion_rate": 0.9, "profile_completeness_score": 85, "has_github": True, "github_score_clean": 70.0, "avg_skill_assessment": 75.0, "verified_skills": ["Python"], "bm25_hit_skills": True, "bm25_hit_titles": True, "bm25_hit_education": True, "bm25_hit_locations": True, "bm25_hit_industries": True, "potion_score": 0.9, "soft_score": 0.8, "bm25_score": 0.7, "composite_score": 0.85, "granite_score": 0.9, "reranker_score": 1.9, "candidate_quality": 0.7, "final_score": 1.0, } base.update(over) return base def _frame(rows: list[dict[str, object]]) -> pl.DataFrame: return pl.DataFrame(rows) def _settings(tmp_path: Path, **reasoning: object) -> Settings: opts = {"mode": "template", "variant_select": "random", "explanations_path": str(tmp_path / "explanations.json")} opts.update(reasoning) return Settings(reasoning=ReasoningSettings(**opts)) # ── band mappers ────────────────────────────────────────────────────────────────────────────────────────────── def test_band_mappers() -> None: assert experience_fit(7.0, 5, 9) == "in" and experience_fit(3.0, 5, 9) == "below" assert experience_fit(11.0, 5, 9) == "above" assert skill_coverage(3, 3) == "strong" and skill_coverage(1, 3) == "partial" and skill_coverage(0, 3) == "thin" assert location_tier("Pune", ["Pune"], ["Bangalore"]) == "preferred" assert location_tier("Bangalore", ["Pune"], ["Bangalore"]) == "acceptable" assert location_tier("Berlin", ["Pune"], ["Bangalore"]) == "outside" assert recency_band(10, 90) == "month" and recency_band(60, 90) == "quarter" and recency_band(400, 90) == "stale" assert notice_band(15, False) == "short" and notice_band(90, False) == "long" def test_fact_matchers() -> None: assert _matched_skills(["Python", "Go"], ["python", "sql"]) == ["Python"] assert _title_match(["AI Engineer"], "Senior AI Engineer") == "AI Engineer" assert title_fit(["AI Engineer"], "Senior AI Engineer", True) == ("AI Engineer", "align") # skills match but title totally off → mismatch trap assert title_fit(["AI Engineer"], "Accountant", True) == ("AI Engineer", "mismatch") assert matched_industries(["Fintech"], "FinTech Solutions", ["SaaS"]) == ["Fintech"] assert matched_education(["Computer Science"], [{"field_of_study": "Computer Science"}]) == ["Computer Science"] def test_signal_bands_sentinel_and_gate() -> None: # github sentinel (has_github False) → skipped; active_market gate fires (open + active + responsive + interest) sig = signal_bands(_row(has_github=False), recency="month", location_outside=False) assert sig["github"] is None assert sig["active_market"] is True and sig["reachability"] == "full" # passive candidate: not enough flags → no soft pattern passive = signal_bands(_row(open_to_work_flag=False, recruiter_response_rate=0.2, saved_by_recruiters_30d=0, profile_views_received_30d=0), recency="stale", location_outside=False) assert passive["active_market"] is False # ── composed output ─────────────────────────────────────────────────────────────────────────────────────────── def test_grounded_multisentence_reasoning(tmp_path: Path) -> None: out = add_explanation(_frame([_row()]), _SPEC, _settings(tmp_path)) top = out.row(0, named=True) assert "Python" in top["reasoning"] and "Pune" in top["reasoning"] # skills + location surfaced assert top["reasoning"].count(".") >= 3 # multi-sentence (T1 ~ 4) assert "AI Engineer" in top["explanation"] # title detail in the fuller note def test_t1_has_no_gap_but_explanation_does(tmp_path: Path) -> None: # a candidate with a real gap (missing skill + outside location), forced into T1 (rank 1) row = _row(flat_skills=["Python"], location="Berlin, DE", country="Germany") out = add_explanation(_frame([row]), _SPEC, _settings(tmp_path, tier1_top=1)) top = out.row(0, named=True) assert "though" not in top["reasoning"].lower() and "outside" not in top["reasoning"].lower() # T1 = clean assert "Watch-outs:" in top["explanation"] # gaps live in the explanation/audit def test_t3_surfaces_one_soft_gap(tmp_path: Path) -> None: # two rows; tier bounds put rank 2 into T3 (allow_neg) — a gap candidate there shows ≤1 soft gap in CSV rows = [_row(candidate_id="A"), _row(candidate_id="B", flat_skills=["Python"], location="Berlin, DE", days_since_active=400, reranker_score=0.1, final_score=0.1)] out = add_explanation(_frame(rows), _SPEC, _settings(tmp_path, tier1_top=1, tier2_top=1, tier3_top=2)) b = out.filter(pl.col("candidate_id") == "B").row(0, named=True) pivots = ("though", "albeit", "caveat", "watch-out", "however", "main gap") # covers every _GAP_PIVOT phrase hits = sum(b["reasoning"].lower().count(p) for p in pivots) assert hits == 1 # exactly one soft gap (any pivot wording) # ── salvage + grounding gates (unchanged behavior) ──────────────────────────────────────────────────────────── def test_clean_note_salvages_and_rejects() -> None: assert _clean_note("Sure! Here it is: A senior engineer with deep retrieval experience.", "fb") \ == "A senior engineer with deep retrieval experience." assert _clean_note("Recruiter Note: /no_think", "fallback line here") == "fallback line here" def test_is_grounded() -> None: source = "Senior AI Engineer; strong on embeddings, Pinecone, Weaviate; based in Pune." assert _is_grounded("Senior AI Engineer strong on embeddings and Pinecone, based in Pune.", source) assert not _is_grounded("Senior AI Engineer strong on Kubernetes.", source) # invented skill rejected # ── artifact + empty guard ──────────────────────────────────────────────────────────────────────────────────── def test_artifact_schema(tmp_path: Path) -> None: settings = _settings(tmp_path) add_explanation(_frame([_row()]), _SPEC, settings) payload = orjson.loads((tmp_path / "explanations.json").read_bytes()) entry = payload[0] assert entry["rank"] == 1 and entry["tier"] == "T1" assert entry["matched_skills"] == ["Python", "embeddings", "ranking"] assert "bands" in entry and "signals" in entry and "positives" in entry and "sentences" in entry assert entry["bm25_hits"]["skills"] is True assert entry["llm_polished"] is False # template mode never polishes def test_empty_passthrough(tmp_path: Path) -> None: out = add_explanation(_frame([_row()]).head(0), _SPEC, _settings(tmp_path)) assert out.height == 0 and {"reasoning", "explanation"} <= set(out.columns)