fix: prevent hidden PDF text from inflating hiring scores

evaluation_validation.py

@@ -0,0 +1,165 @@
+"""
+Deterministic validation and normalization for LLM evaluation output.
+
+Enforces score policy in code (not prompts only) and grounds bonus claims
+against trusted resume / GitHub source text. See issue #273.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Optional
+
+from models import EvaluationData, JSONResume
+
+logger = logging.getLogger(__name__)
+
+MAX_BONUS_POINTS = 20
+MIN_FINAL_SCORE = -20
+MAX_FINAL_SCORE = 120
+
+CATEGORY_CAPS = {
+    "open_source": 35,
+    "self_projects": 30,
+    "production": 25,
+    "technical_skills": 10,
+}
+
+# production score cannot exceed this when structured resume has no work history
+PRODUCTION_CAP_WITHOUT_WORK = 5
+
+# open_source score cannot exceed this without real GitHub enrichment data
+OPEN_SOURCE_CAP_WITHOUT_GITHUB = 10
+
+BONUS_CLAIMS = (
+    (re.compile(r"google summer of code|\bgsoc\b", re.I), 5, ("google summer of code", "gsoc")),
+    (
+        re.compile(r"girl\s*script\s*summer\s*of\s*code", re.I),
+        3,
+        ("girl script summer of code", "girlscript"),
+    ),
+    (
+        re.compile(r"startup\s*(founder|co-?founder)|founder\s*experience", re.I),
+        5,
+        ("founder", "co-founder", "cofounder"),
+    ),
+    (re.compile(r"early[- ]stage\s*engineer", re.I), 3, ("early-stage", "early stage")),
+    (re.compile(r"portfolio\s*website", re.I), 2, ("portfolio", "github.io")),
+    (re.compile(r"linkedin\s*profile", re.I), 1, ("linkedin.com", "linkedin")),
+)
+
+
+def _source_corpus(
+    source_text: str,
+    resume_data: Optional[JSONResume],
+    github_data: Optional[dict],
+) -> str:
+    parts = [source_text or ""]
+    if resume_data:
+        from transform import convert_json_resume_to_text
+
+        parts.append(convert_json_resume_to_text(resume_data))
+    if github_data:
+        from transform import convert_github_data_to_text
+
+        parts.append(convert_github_data_to_text(github_data))
+    return "\n".join(parts).lower()
+
+
+def _has_substantive_work(resume_data: Optional[JSONResume]) -> bool:
+    if not resume_data or not resume_data.work:
+        return False
+    return any(
+        (w.name and w.name.strip()) or (w.position and w.position.strip())
+        for w in resume_data.work
+    )
+
+
+def _has_github_enrichment(github_data: Optional[dict]) -> bool:
+    return bool(
+        github_data
+        and isinstance(github_data, dict)
+        and github_data.get("profile")
+        and github_data.get("projects")
+    )
+
+
+def _validate_bonus_points(
+    evaluation: EvaluationData, corpus: str
+) -> tuple[float, str]:
+    breakdown = evaluation.bonus_points.breakdown or ""
+    validated_total = 0.0
+    notes: list[str] = []
+
+    for pattern, points, source_markers in BONUS_CLAIMS:
+        if not pattern.search(breakdown):
+            continue
+        if any(marker in corpus for marker in source_markers):
+            validated_total += points
+            notes.append(f"+{points} {pattern.pattern}")
+        else:
+            notes.append(f"removed unverified bonus ({pattern.pattern})")
+
+    validated_total = min(validated_total, MAX_BONUS_POINTS)
+    if validated_total == 0 and breakdown.strip():
+        note_text = "; ".join(notes) if notes else "No verified bonus claims in source text"
+    else:
+        note_text = "; ".join(notes) if notes else breakdown
+
+    return validated_total, note_text
+
+
+def normalize_evaluation(
+    evaluation: EvaluationData,
+    *,
+    source_text: str = "",
+    resume_data: Optional[JSONResume] = None,
+    github_data: Optional[dict] = None,
+) -> EvaluationData:
+    """
+    Clamp category scores, validate bonuses, and apply structural caps
+    based on trusted resume / GitHub data.
+    """
+    corpus = _source_corpus(source_text, resume_data, github_data)
+    scores = evaluation.scores
+    updates = {}
+
+    for field, cap in CATEGORY_CAPS.items():
+        category = getattr(scores, field)
+        capped_score = min(float(category.score), cap)
+
+        if field == "production" and not _has_substantive_work(resume_data):
+            capped_score = min(capped_score, PRODUCTION_CAP_WITHOUT_WORK)
+
+        if field == "open_source" and not _has_github_enrichment(github_data):
+            capped_score = min(capped_score, OPEN_SOURCE_CAP_WITHOUT_GITHUB)
+
+        if capped_score < category.score:
+            logger.warning(
+                "%s score capped from %s to %s",
+                field,
+                category.score,
+                capped_score,
+            )
+
+        updates[field] = category.model_copy(
+            update={"score": capped_score, "max": cap}
+        )
+
+    bonus_total, bonus_breakdown = _validate_bonus_points(evaluation, corpus)
+
+    deductions = min(float(evaluation.deductions.total), MAX_FINAL_SCORE)
+    category_sum = sum(getattr(updates[f], "score") for f in CATEGORY_CAPS)
+    max_deductions = max(0, category_sum + bonus_total - MIN_FINAL_SCORE)
+    deductions = min(deductions, max_deductions)
+
+    return evaluation.model_copy(
+        update={
+            "scores": scores.model_copy(update=updates),
+            "bonus_points": evaluation.bonus_points.model_copy(
+                update={"total": bonus_total, "breakdown": bonus_breakdown}
+            ),
+            "deductions": evaluation.deductions.model_copy(update={"total": deductions}),
+        }
+    )

evaluator.py

@@ -2,6 +2,8 @@
 from pydantic import BaseModel, Field, field_validator
 from models import JSONResume, EvaluationData
 from llm_utils import initialize_llm_provider, extract_json_from_response
+from evaluation_validation import normalize_evaluation
+from pdf_sanitize import sanitize_resume_text_for_pipeline
 import logging
 import json
 import re
@@ -45,7 +47,13 @@ def _load_evaluation_prompt(self, resume_text: str) -> str:
             raise ValueError("Failed to load resume evaluation criteria template")
         return criteria_template
 
-    def evaluate_resume(self, resume_text: str) -> EvaluationData:
+    def evaluate_resume(
+        self,
+        resume_text: str,
+        resume_data: Optional[JSONResume] = None,
+        github_data: Optional[dict] = None,
+    ) -> EvaluationData:
+        resume_text = sanitize_resume_text_for_pipeline(resume_text)
         self._last_resume_text = resume_text
         full_prompt = self._load_evaluation_prompt(resume_text)
         # logger.info(f"🔤 Evaluation prompt being sent: {full_prompt}")
@@ -84,7 +92,12 @@ def evaluate_resume(self, resume_text: str) -> EvaluationData:
             evaluation_dict = json.loads(response_text)
             evaluation_data = EvaluationData(**evaluation_dict)
 
-            return evaluation_data
+            return normalize_evaluation(
+                evaluation_data,
+                source_text=resume_text,
+                resume_data=resume_data,
+                github_data=github_data,
+            )
 
         except Exception as e:
             logger.error(f"Error evaluating resume: {str(e)}")

pdf.py

@@ -21,7 +21,7 @@
     AwardsSection,
 )
 from llm_utils import initialize_llm_provider, extract_json_from_response
-from pymupdf_rag import to_markdown
+from pdf_sanitize import extract_visible_text_from_pdf, sanitize_resume_text_for_pipeline
 from typing import List, Optional, Dict, Any
 from prompt import (
     DEFAULT_MODEL,
@@ -50,13 +50,10 @@ def extract_text_from_pdf(self, pdf_path: str) -> Optional[str]:
                 raise FileNotFoundError(f"PDF file not found: {pdf_path}")
 
             with pymupdf.open(pdf_path) as doc:
-                pages = range(doc.page_count)
-                resume_text = to_markdown(
-                    doc,
-                    pages=pages,
-                )
+                resume_text = extract_visible_text_from_pdf(doc)
+                resume_text = sanitize_resume_text_for_pipeline(resume_text)
                 logger.debug(
-                    f"Extracted text from PDF: {len(resume_text) if resume_text else 0} characters"
+                    f"Extracted visible text from PDF: {len(resume_text) if resume_text else 0} characters"
                 )
                 return resume_text
         except Exception as e:

pdf_sanitize.py

@@ -0,0 +1,115 @@
+"""
+Filter hidden or low-visibility PDF text before LLM resume extraction.
+
+Mitigates white-on-white and tiny-font injection (see issue #273).
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from pathlib import Path
+from typing import Union
+
+import pymupdf
+
+logger = logging.getLogger(__name__)
+
+# Spans below this size are treated as hidden metadata (PoC footer uses 6pt).
+MIN_VISIBLE_FONT_SIZE = 7.0
+
+# RGB components all above this are treated as background-matching (near-white).
+WHITE_COLOR_THRESHOLD = 0.90
+
+# Resume PDFs must not embed GitHub enrichment blocks; only github.py adds those.
+FORGED_GITHUB_BLOCK_RE = re.compile(
+    r"===\s*GITHUB\s*DATA\s*===.*", re.DOTALL | re.IGNORECASE
+)
+
+
+def _span_rgb(color: int) -> tuple[float, float, float]:
+    value = color & 0xFFFFFF
+    r = ((value >> 16) & 0xFF) / 255.0
+    g = ((value >> 8) & 0xFF) / 255.0
+    b = (value & 0xFF) / 255.0
+    return r, g, b
+
+
+def is_hidden_span(
+    span: dict,
+    *,
+    min_font_size: float = MIN_VISIBLE_FONT_SIZE,
+    white_threshold: float = WHITE_COLOR_THRESHOLD,
+) -> bool:
+    """Return True if span is likely invisible or evasion text."""
+    size = float(span.get("size", 12))
+    if size < min_font_size:
+        return True
+
+    color = span.get("color", 0)
+    if isinstance(color, int):
+        r, g, b = _span_rgb(color)
+        if r >= white_threshold and g >= white_threshold and b >= white_threshold:
+            return True
+
+    return False
+
+
+def extract_visible_text_from_pdf(pdf_source: Union[str, Path, pymupdf.Document]) -> str:
+    """
+    Extract printable resume text, skipping hidden spans (near-white ink, tiny font).
+    """
+    owns_doc = False
+    if isinstance(pdf_source, pymupdf.Document):
+        doc = pdf_source
+    else:
+        doc = pymupdf.open(pdf_source)
+        owns_doc = True
+
+    try:
+        lines: list[str] = []
+        skipped_spans = 0
+        kept_spans = 0
+
+        for page in doc:
+            page_dict = page.get_text("dict", flags=pymupdf.TEXTFLAGS_TEXT)
+            for block in page_dict.get("blocks", []):
+                if block.get("type") != 0:
+                    continue
+                for line in block.get("lines", []):
+                    parts: list[str] = []
+                    for span in line.get("spans", []):
+                        if is_hidden_span(span):
+                            skipped_spans += 1
+                            continue
+                        kept_spans += 1
+                        parts.append(span.get("text", ""))
+                    if parts:
+                        lines.append("".join(parts).rstrip())
+
+        text = "\n".join(lines).strip()
+        logger.debug(
+            "PDF visible extraction: %d kept spans, %d hidden spans, %d chars",
+            kept_spans,
+            skipped_spans,
+            len(text),
+        )
+        return text
+    finally:
+        if owns_doc:
+            doc.close()
+
+
+def strip_forged_github_blocks(text: str) -> str:
+    """Remove embedded === GITHUB DATA === blocks from untrusted resume text."""
+    if not text:
+        return text
+    cleaned = FORGED_GITHUB_BLOCK_RE.sub("", text).strip()
+    if cleaned != text:
+        logger.warning("Removed forged === GITHUB DATA === block from resume text")
+    return cleaned
+
+
+def sanitize_resume_text_for_pipeline(text: str) -> str:
+    """Defense-in-depth cleanup before LLM extraction or evaluation."""
+    return strip_forged_github_blocks(text)

score.py

@@ -181,7 +181,11 @@ def _evaluate_resume(
         resume_text += blog_text
 
     # Evaluate the enhanced resume
-    evaluation_result = evaluator.evaluate_resume(resume_text)
+    evaluation_result = evaluator.evaluate_resume(
+        resume_text,
+        resume_data=resume_data,
+        github_data=github_data,
+    )
 
     # print(evaluation_result)