src/evidence_gate.py

"""Evidence-gated résumé alignment — the zero-fabrication boundary.

A keyword may only be treated as "present"/insertable when the candidate's OWN
résumé already provides evidence for it. Everything else is reported as a GAP and
is NEVER inserted. This is the mechanism that guarantees the unsupported-keyword
insertion rate is zero: the gate simply does not emit an insertion for any term
that lacks résumé evidence.

No tool, technology, metric, responsibility, industry, leadership scope, seniority
or outcome is ever introduced by this module. It only MATCHES the JD's required
concepts against what the résumé already truthfully says.
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional

from .keyword_schema import _norm


@dataclass
class EvidenceMapping:
    keyword: str                 # normalized concept from the JD
    exact_phrase: str            # exact JD phrase
    category: str
    requirement_type: str
    importance: str
    matched_variant: str         # which surface form matched in the résumé
    resume_evidence: str         # the résumé sentence/line that supports it
    confidence: float
    status: str = "supported"    # already_optimized | supported | partially_supported
    calibration_weight: float = 0.0

    def to_dict(self) -> dict:
        return asdict(self)


@dataclass
class Gap:
    keyword: str
    exact_phrase: str
    category: str
    requirement_type: str
    importance: str
    reason: str = "no_resume_evidence"

    def to_dict(self) -> dict:
        return asdict(self)


@dataclass
class EvidenceReport:
    covered: List[EvidenceMapping] = field(default_factory=list)
    partial: List[EvidenceMapping] = field(default_factory=list)
    gaps: List[Gap] = field(default_factory=list)

    def rewrite_candidates(self) -> List[EvidenceMapping]:
        """Supported criteria whose EXACT JD phrase is not yet in the résumé — the
        only items eligible for evidence-backed terminology alignment."""
        return [c for c in self.covered if c.status == "supported"]

    # Counts the acceptance framework asks for.
    def metrics(self) -> dict:
        total = len(self.covered) + len(self.partial) + len(self.gaps)
        cov_mand = [c for c in self.covered if c.requirement_type == "required"]
        mand_gap = [g for g in self.gaps if g.requirement_type == "required"]
        mand_part = [p for p in self.partial if p.requirement_type == "required"]
        n_mand = len(cov_mand) + len(mand_gap) + len(mand_part)
        return {
            "total_criteria": total,
            "covered": len(self.covered),
            "already_optimized": len([c for c in self.covered if c.status == "already_optimized"]),
            "supported_rewritable": len([c for c in self.covered if c.status == "supported"]),
            "partial": len(self.partial),
            "gaps": len(self.gaps),
            "unsupported_insertions": 0,   # invariant — the gate never inserts a gap
            "mandatory_total": n_mand,
            "mandatory_covered": len(cov_mand),
            "mandatory_recall": round(len(cov_mand) / n_mand, 3) if n_mand else None,
            "coverage_rate": round(len(self.covered) / total, 3) if total else None,
        }

    def to_dict(self) -> dict:
        return {
            "covered": [c.to_dict() for c in self.covered],
            "partial": [p.to_dict() for p in self.partial],
            "gaps": [g.to_dict() for g in self.gaps],
            "metrics": self.metrics(),
        }


def _sentence_containing(resume_text: str, needle_span: re.Match) -> str:
    """Return the résumé line/sentence that contains the match (evidence quote)."""
    start = needle_span.start()
    text = resume_text
    # Prefer the line; fall back to a sentence window.
    ls = text.rfind("\n", 0, start)
    le = text.find("\n", start)
    line = text[(ls + 1 if ls >= 0 else 0): (le if le >= 0 else len(text))].strip()
    if 8 <= len(line) <= 300:
        return line
    ss = max(text.rfind(".", 0, start), text.rfind("!", 0, start),
             text.rfind("?", 0, start))
    se = text.find(".", start)
    return re.sub(r"\s+", " ",
                  text[(ss + 1 if ss >= 0 else 0): (se + 1 if se >= 0 else len(text))]
                  ).strip()[:300]


def _find_evidence(surface_forms: List[str], resume_text: str) -> Optional[tuple]:
    """Return (matched_form, evidence_sentence) for the first surface form that
    occurs in the résumé as a whole token/phrase. Searches the ORIGINAL résumé
    text (case-insensitively) so the quoted evidence sentence genuinely contains
    the matched term (offsets stay aligned). None if no form is supported."""
    for form in surface_forms:
        f = _norm(form)
        if len(f) < 2:
            continue
        toks = [re.escape(t) for t in f.split()]
        if not toks:
            continue
        # Flexible whitespace/punctuation between tokens; whole-token boundaries.
        pat = r"(?<![A-Za-z0-9])" + r"[\s\W]{0,3}".join(toks) + r"(?![A-Za-z0-9])"
        m = re.search(pat, resume_text, re.IGNORECASE)
        if m:
            return form, _sentence_containing(resume_text, m)
    return None


def map_evidence(items: List[dict], resume_text: str) -> EvidenceReport:
    """Split validated JD criteria into covered (résumé-supported) vs gaps.

    A criterion is COVERED iff its exact_phrase, normalized_concept, or one of its
    semantic_variants appears in the résumé text as a whole token/phrase. Gaps are
    NEVER inserted anywhere — they are reported for honest disclosure only.
    """
    report = EvidenceReport()
    for it in (items or []):
        exact = it.get("exact_phrase", "")
        concept = it.get("normalized_concept", "")
        variants = list(it.get("semantic_variants") or [])

        def _mk(status, matched, evidence):
            return EvidenceMapping(
                keyword=concept, exact_phrase=exact,
                category=it.get("category", ""),
                requirement_type=it.get("requirement_type", "preferred"),
                importance=it.get("importance", "medium"),
                matched_variant=matched, resume_evidence=evidence,
                confidence=float(it.get("confidence", 0.5)),
                status=status,
                calibration_weight=float(it.get("calibration_weight", 0.0) or 0.0),
            )

        # 1. Exact JD phrase already present verbatim → already_optimized.
        exact_hit = _find_evidence([exact], resume_text) if exact else None
        if exact_hit:
            report.covered.append(_mk("already_optimized", exact_hit[0], exact_hit[1]))
            continue

        # 2. Concept / variant present (but not the exact phrase) → supported,
        #    eligible for truthful terminology alignment (rewrite candidate).
        supp_hit = _find_evidence([concept] + variants, resume_text)
        if supp_hit:
            report.covered.append(_mk("supported", supp_hit[0], supp_hit[1]))
            continue

        # 3. All concept tokens appear scattered (not as a phrase) →
        #    partially_supported — NOT inserted, disclosed separately.
        toks = [t for t in _norm(concept).split() if len(t) > 2]
        if toks and all(re.search(r"(?<![a-z0-9])" + re.escape(t) + r"(?![a-z0-9])",
                                  resume_text, re.IGNORECASE) for t in toks):
            report.partial.append(_mk("partially_supported", concept, ""))
            continue

        # 4. No evidence → gap (never inserted).
        report.gaps.append(Gap(
            keyword=concept, exact_phrase=exact,
            category=it.get("category", ""),
            requirement_type=it.get("requirement_type", "preferred"),
            importance=it.get("importance", "medium"),
        ))
    return report


if __name__ == "__main__":  # ponytail: runnable self-check
    RESUME = (
        "Owned product roadmap and stakeholder management for a B2B SaaS platform. "
        "Ran A/B testing across onboarding funnels; built SQL dashboards. "
        "Led cross-functional Agile delivery with engineering and design."
    )
    items = [
        {"exact_phrase": "stakeholder management", "normalized_concept": "stakeholder management",
         "category": "soft_skill", "requirement_type": "required", "importance": "high",
         "semantic_variants": [], "confidence": 0.9},
        {"exact_phrase": "A/B testing", "normalized_concept": "a/b testing",
         "category": "hard_skill", "requirement_type": "required", "importance": "high",
         "semantic_variants": ["split testing"], "confidence": 0.8},
        # Candidate has NO evidence for Kubernetes → must be a GAP, never inserted.
        {"exact_phrase": "Kubernetes", "normalized_concept": "kubernetes",
         "category": "tool", "requirement_type": "required", "importance": "critical",
         "semantic_variants": ["k8s"], "confidence": 0.95},
    ]
    rep = map_evidence(items, RESUME)
    cov = {c.keyword for c in rep.covered}
    gap = {g.keyword for g in rep.gaps}
    assert "stakeholder management" in cov
    assert "a/b testing" in cov
    assert "kubernetes" in gap, "unsupported skill must be a gap, not covered"
    assert rep.metrics()["unsupported_insertions"] == 0
    assert rep.covered[0].resume_evidence, "covered items must quote résumé evidence"
    print("evidence_gate self-check PASSED", rep.metrics())