scripts/verify_maximum_ats.py

"""
Deterministic regression for Maximum ATS Mode (User-Confirmed Skill Expansion).

Uses StubProvider (no API keys). Proves:
  1. Normal PM/AI missing terms are treated as user-confirmed in Maximum ATS Mode.
  2. Hard anti-fake boundaries still hold (certs / seniority / engineering).
  3. High-risk terms need confirmation; confirming them lets them in.
  4. Pasted Jobalytics feedback improves keyword coverage.
  5. A 65%-style case is repaired toward target instead of being accepted.
  6. The rich coverage report is produced (per-keyword + sections + categories).
  7. Scores come from the re-parsed exported resume.
  8. Word/skills caps do not block keyword coverage (skills can exceed a small cap).
"""
import os, sys, io, shutil
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
sys.path.insert(0, os.path.abspath("."))

# Isolate the vault so prior confirmations don't change results.
import src.candidate_vault as _cv
_cv._VAULT_PATH = "data/_test_vault_maxats.json"
if os.path.exists(_cv._VAULT_PATH):
    os.remove(_cv._VAULT_PATH)

import shutil as _sh
src_pdf = r"C:\Users\Nxtwave\Desktop\resume\Saiteja_Tirunagari_Resume A 26 - Copy.pdf"
dst = "data/resume/resume.pdf"
os.makedirs("data/resume", exist_ok=True)
if not os.path.exists(dst) and os.path.exists(src_pdf):
    _sh.copyfile(src_pdf, dst)

from src.resume_parser_v2 import parse_resume_pdf
from src.providers import StubProvider
from src.jobalytics_repair import repair_with_external_feedback
from src.resume_customizer import _read_docx_text
from src.candidate_fit import (
    classify_fit, severity, LOW, MEDIUM, HIGH, BLOCKED,
)
from src.jd_analyzer import Requirement
from src.fit_gate import (
    READY, READY_REVIEW, READY_MAX_ATS_95_PLUS, READY_90_PLUS_EXTERNAL_ALIGNED,
    READY_95_EXTERNAL_ALIGNED, BELOW_TARGET_REPAIRABLE, NEEDS_USER_CONFIRMATION,
)

ok = True
def check(name, cond, detail=""):
    global ok; ok = ok and cond
    print(f"  [{'PASS' if cond else 'FAIL'}] {name}  {detail}")

BASE = "Product manager with 5 years building SaaS products. Led roadmap and agile delivery."

def _verdict(term, category, max_mode, candidate_years=5, confirmed=None):
    r = Requirement(term=term, category=category)
    return classify_fit(r, BASE, seniority="mid", candidate_years=candidate_years,
                        maximum_ats_mode=max_mode, confirmed=confirmed)

# ── 1. Normal PM/AI terms become user-confirmed (LOW) in Maximum ATS Mode ──────
print("1. Maximum ATS Mode promotes normal PM/AI terms to user-confirmed (LOW):")
for t, cat in [("generative ai", "hard_skill"), ("prompt design", "hard_skill"),
               ("machine learning", "hard_skill"), ("a/b testing", "responsibility"),
               ("stakeholder management", "soft_skill"), ("roadmap", "responsibility"),
               ("saas", "domain"), ("b2b", "domain"), ("product ownership", "responsibility")]:
    v = _verdict(t, cat, max_mode=True)
    check(f"'{t}' -> LOW/include", v.action == "include" and severity(v) == LOW,
          f"action={v.action} sev={severity(v)}")

# ── 2. Hard anti-fake boundaries still hold in Maximum ATS Mode ────────────────
print("2. Hard boundaries unchanged in Maximum ATS Mode:")
v = _verdict("cissp", "certification", max_mode=True)
check("CISSP cert still blocked/ask (never auto LOW)", severity(v) in (HIGH, BLOCKED),
      f"sev={severity(v)} action={v.action}")
v = _verdict("pmp", "certification", max_mode=True)
check("PMP cert not auto-included", severity(v) in (HIGH, BLOCKED), f"sev={severity(v)}")
v = _verdict("12+ years", "seniority", max_mode=True, candidate_years=5)
check("12+ years seniority blocked for 5y candidate", v.action == "block",
      f"action={v.action} reason={v.reason}")
v = _verdict("spring boot", "hard_skill", max_mode=True)
check("engineering 'spring boot' stays HIGH (needs confirm)", severity(v) == HIGH,
      f"sev={severity(v)}")
v = _verdict("siem", "tool", max_mode=True)
check("specialized 'siem' stays HIGH (needs confirm)", severity(v) == HIGH,
      f"sev={severity(v)}")

# ── 3. Confirming a high-risk term lets it in ─────────────────────────────────
print("3. Per-request confirmation promotes a HIGH term:")
v = _verdict("siem", "tool", max_mode=True, confirmed={"siem"})
check("confirmed 'siem' -> include", v.action == "include", f"action={v.action}")

# ── 4-8. Repair flow with Maximum ATS Mode ────────────────────────────────────
base = parse_resume_pdf(dst)
jd = open("tests/fixtures/jds/generic_pm_3_7yrs.txt", encoding="utf-8").read()
job = {"title": "AI Product Manager", "company": "TestCo", "description": jd, "_raw_assessment": {}}

# A 65%-style external result missing many PM/AI terms + 1 blocked + 1 high-risk.
fb = ("Match score 65%. Missing keywords: generative ai, prompt design, machine learning, "
      "a/b testing, product discovery, stakeholder management, roadmap, agile, user stories, "
      "acceptance criteria, saas, b2b, enterprise platform, product ownership, data-driven, "
      "experimentation, CISSP, SIEM.")
res = repair_with_external_feedback(job, feedback_text=fb, provider=StubProvider(),
                                    base_resume=base, maximum_ats_mode=True,
                                    output_dir="data/output/resumes/_maxats_test")
print("4. repair (Maximum ATS Mode):", res.get("status"), res.get("scores"),
      "cov=", res.get("after_coverage", {}).get("pct"))
check("no error", "error" not in res, res.get("error", ""))

before_pct = res.get("before_coverage", {}).get("pct", 0)
after_pct = res.get("after_coverage", {}).get("pct", 0)
check("coverage improved (after >= before)", after_pct >= before_pct,
      f"{before_pct}% -> {after_pct}%")

# Classification guarantee (deterministic): normal PM/AI terms are treated as
# user-confirmed/addable in Maximum ATS Mode — disposition is auto-included, NOT
# blocked or needs-confirmation. (Physical weaving completeness is the live LLM's
# job; the stub places a subset, so we assert intent + partial placement here.)
cov0 = res.get("coverage_report", {})
disp = {k["keyword"].lower(): k["disposition"] for k in cov0.get("keywords", [])}
AUTO = {"included_auto", "included_review", "already_present", "user_confirmed"}
pm_ai = ["generative ai", "prompt design", "machine learning", "a/b testing",
         "product discovery", "stakeholder management", "roadmap", "agile",
         "user stories", "acceptance criteria", "saas", "b2b",
         "enterprise platform", "product ownership", "data-driven", "experimentation"]
auto_count = sum(1 for t in pm_ai if disp.get(t) in AUTO)
check("normal PM/AI terms classified addable (>=12 of 16)", auto_count >= 12,
      f"auto={auto_count}/16")
check("'generative ai' & 'prompt design' treated as user-confirmed (not blocked)",
      disp.get("generative ai") in AUTO and disp.get("prompt design") in AUTO,
      f"gen-ai={disp.get('generative ai')} prompt={disp.get('prompt design')}")
added_l = [a.lower() for a in res.get("added_terms", [])]
check("some normal PM/AI terms physically woven (>=3)", len(added_l) >= 3,
      f"added={res.get('added_terms', [])[:12]}")

# Honesty: blocked/high terms never fabricated.
txt = _read_docx_text(res["resume_path"]).lower()
check("CISSP NOT in resume", "cissp" not in txt)
check("SIEM NOT auto-added (unconfirmed high-risk)", "siem" not in txt)
check("CISSP in blocked_terms", "CISSP" in res.get("blocked_terms", []))
check("SIEM in unresolved_high_risk_terms", "SIEM" in res.get("unresolved_high_risk_terms", []))

# Status: a 65% PM/AI case must NOT be a terminal failure — it is repaired toward
# target or flagged repairable / confirmation, never silently accepted at 65%.
acceptable = {READY, READY_REVIEW, READY_MAX_ATS_95_PLUS, READY_90_PLUS_EXTERNAL_ALIGNED,
              READY_95_EXTERNAL_ALIGNED, BELOW_TARGET_REPAIRABLE, NEEDS_USER_CONFIRMATION}
check("status is aggressive/repairable (not accepted at 65%)",
      res.get("status") in acceptable, res.get("status"))

# Rich coverage report present and structured.
cov = res.get("coverage_report", {})
check("coverage_report present", bool(cov.get("keywords")), f"keys={list(cov.keys())}")
check("coverage_report has by_category", bool(cov.get("by_category")), str(cov.get("by_category")))
secs = {k.get("section") for k in cov.get("keywords", []) if k.get("placed_in_resume")}
check("terms placed across multiple sections (not Skills only)",
      len(secs - {"Skills"}) >= 1 or len(secs) >= 2, f"sections={secs}")

# Scores from re-parsed export.
check("scores present from parsed export",
      isinstance(res.get("scores", {}).get("internal_jd_match"), int)
      and isinstance(res.get("scores", {}).get("independent_jd_match"), int),
      str(res.get("scores")))

# Word/skills cap does not block coverage: skills list can exceed a small cap.
internal = res.get("scores", {}).get("internal_jd_match", 0)
check("internal score high (caps did not block coverage)", internal >= 80, f"internal={internal}")

# ── 9. Confirming high-risk term in the repair flow adds it ───────────────────
print("5. Confirm SIEM in repair flow:")
res2 = repair_with_external_feedback(job, feedback_text=fb, provider=StubProvider(),
                                     base_resume=base, maximum_ats_mode=True,
                                     confirmed_terms=["SIEM"],
                                     output_dir="data/output/resumes/_maxats_test2")
check("SIEM no longer unresolved after confirmation",
      "SIEM" not in res2.get("unresolved_high_risk_terms", []),
      str(res2.get("unresolved_high_risk_terms", [])))

print("\n" + ("\u2713 ALL MAXIMUM-ATS CHECKS PASS" if ok else "\u2717 SOME FAILED"))
shutil.rmtree("data/output/resumes/_maxats_test", ignore_errors=True)
shutil.rmtree("data/output/resumes/_maxats_test2", ignore_errors=True)
if os.path.exists(_cv._VAULT_PATH):
    os.remove(_cv._VAULT_PATH)
sys.exit(0 if ok else 1)