tests/test_explain.py

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