validate.py

"""Verbatim-grounding validator for English fine-tune JSONL (D-12).

Each record's `output` is checked against the union of source markdown in
build/extracted/. Records whose output cannot be substantially attributed to
the corpus are flagged as fabricated and the run exits 1.

Translated records (zh, ar) are spot-checked manually per D-12, not by this script.
"""
from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path


def _validate_path(path: str) -> str:
    root = Path.cwd().resolve()
    candidate = Path(path).resolve()
    try:
        candidate.relative_to(root)
    except ValueError as exc:
        raise ValueError(
            f"Path traversal detected: {path} is outside allowed directory."
        ) from exc
    return str(candidate)


_WORD_RE = re.compile(r"\w+", re.UNICODE)


def _bigrams(text: str) -> set[tuple[str, str]]:
    words = [w.lower() for w in _WORD_RE.findall(text)]
    return {(words[i], words[i + 1]) for i in range(len(words) - 1)}


def verbatim_grounded(output_text: str, source_corpus: str, threshold: float = 0.6) -> bool:
    """Return True if `output_text` shares >=threshold of its bigrams with corpus.

    Threshold tuned for paraphrase tolerance: Data Recipes legitimately rephrases
    source text, so we don't require exact substring match. Bigram overlap of 0.6
    captures meaningful grounding while rejecting hallucinated facts.
    """
    out_bg = _bigrams(output_text)
    if not out_bg:
        return False
    corpus_bg = _bigrams(source_corpus)
    if not corpus_bg:
        return False
    overlap = len(out_bg & corpus_bg) / len(out_bg)
    return overlap >= threshold


def load_corpus(corpus_dir: Path) -> str:
    files = sorted(Path(corpus_dir).glob("*.md"))
    return "\n\n".join(f.read_text(encoding="utf-8") for f in files)


def validate_jsonl(jsonl_path: Path, corpus_dir: Path, threshold: float = 0.6) -> dict:
    corpus = load_corpus(corpus_dir)
    records = [json.loads(l) for l in jsonl_path.read_text(encoding="utf-8").splitlines() if l.strip()]
    passed = 0
    failed_records: list[int] = []
    for i, rec in enumerate(records):
        if rec.get("language", "en") != "en":
            continue  # only English is auto-validated; zh/ar spot-checked per D-12
        if verbatim_grounded(rec["output"], corpus, threshold):
            passed += 1
        else:
            failed_records.append(i)
    return {
        "total_en": passed + len(failed_records),
        "passed": passed,
        "failed": len(failed_records),
        "fail_records": failed_records,
        "_records": records,
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="Verbatim-grounding validator (D-12)")
    parser.add_argument("--input", required=True, help="JSONL file to validate")
    parser.add_argument("--corpus-dir", required=True, help="Directory of source .md files")
    parser.add_argument("--threshold", type=float, default=0.6, help="Bigram overlap threshold (0..1)")
    parser.add_argument("--output-clean", help="Write passing records to this file (filters out fabricated)")
    args = parser.parse_args()
    input_path = Path(_validate_path(args.input))
    corpus_path = Path(_validate_path(args.corpus_dir))
    result = validate_jsonl(input_path, corpus_path, args.threshold)
    print(
        f"[validate] total_en={result['total_en']}  "
        f"passed={result['passed']}  failed={result['failed']}  "
        f"threshold={args.threshold}",
        flush=True,
    )
    if args.output_clean:
        failed_set = set(result["fail_records"])
        records = result["_records"]
        clean_path = Path(_validate_path(args.output_clean))
        with clean_path.open("w", encoding="utf-8") as f:
            for i, rec in enumerate(records):
                if i not in failed_set:
                    f.write(json.dumps(rec, ensure_ascii=False) + "\n")
        print(f"[validate] wrote {result['passed']} clean records to {clean_path}", flush=True)
    if result["failed"]:
        print(f"[validate] FAIL  fabricated record indices: {result['fail_records']}", flush=True)
        if args.output_clean:
            print("[validate] clean file written; exiting 2 to signal partial pass", flush=True)
            sys.exit(2)
        sys.exit(1)
    print("[validate] OK  all English records grounded in corpus", flush=True)


if __name__ == "__main__":
    main()