scripts/verify_model.py

"""Freshly verify model reload, held-out metrics, inference and package safety."""

from __future__ import annotations

import csv
import hashlib
import importlib.metadata
import json
import math
import re
import sys
import unicodedata
from pathlib import Path

import joblib
from sklearn.metrics import accuracy_score, confusion_matrix, f1_score


ROOT = Path(__file__).resolve().parents[1]
PACKAGE_ROOT = ROOT.parent
DATA_DIR = PACKAGE_ROOT / "nwhite-ai-operations-intent-dataset" / "data"
REPORT_DIR = ROOT / "reports"

REQUIRED_FILES = [
    "README.md",
    "LICENSE",
    "CITATION.cff",
    "requirements.txt",
    "model.joblib",
    "sklearn_model.joblib",
    "web_model.json",
    "model_config.json",
    "label_mapping.json",
    "metrics.json",
    "sample_predictions.json",
    "reports/evaluation_report.md",
    "reports/validation_predictions.csv",
    "reports/test_predictions.csv",
    "scripts/train_model.py",
    "scripts/inference.py",
    "scripts/export_web_model.py",
    "scripts/verify_model.py",
]

SECRET_PATTERNS = {
    "private_key": re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
    "hugging_face_token": re.compile(rb"\bhf_[A-Za-z0-9]{16,}\b"),
    "github_token": re.compile(rb"\bghp_[A-Za-z0-9]{16,}\b"),
    "generic_api_key_assignment": re.compile(rb"(?i)api[_-]?key\s*[:=]\s*['\"][A-Za-z0-9_\-]{16,}"),
}

FORBIDDEN_TEXT = re.compile(
    r"vibe\s+" + r"coding|lorem\s+" + r"ipsum|\b(?:to" + r"do|tb" + r"d)\b",
    re.IGNORECASE,
)


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def read_split(split: str) -> tuple[list[str], list[str], list[str]]:
    with (DATA_DIR / f"{split}.csv").open("r", encoding="utf-8", newline="") as handle:
        rows = list(csv.DictReader(handle))
    return (
        [row["id"] for row in rows],
        [row["user_request"] for row in rows],
        [row["intent"] for row in rows],
    )


def assert_close(actual: float, expected: float, name: str) -> None:
    if not math.isclose(actual, expected, rel_tol=0.0, abs_tol=1e-12):
        raise AssertionError(f"{name} mismatch: recomputed={actual} recorded={expected}")


def web_predict(payload: dict[str, object], texts: list[str]) -> tuple[list[str], list[list[float]]]:
    """Independent reference implementation for the exported browser format."""

    classes = [str(label) for label in payload["classes"]]
    vectorizer = payload["vectorizer"]
    classifier = payload["classifier"]
    vocabulary = {str(term): int(index) for term, index in vectorizer["vocabulary"].items()}
    idf = [float(value) for value in vectorizer["idf"]]
    minimum_n, maximum_n = (int(value) for value in vectorizer["ngram_range"])
    coefficients = [[float(value) for value in row] for row in classifier["coef"]]
    intercepts = [float(value) for value in classifier["intercept"]]

    predictions: list[str] = []
    probability_rows: list[list[float]] = []
    for original in texts:
        text = original.lower() if vectorizer["lowercase"] else original
        text = "".join(
            character
            for character in unicodedata.normalize("NFKD", text)
            if not unicodedata.combining(character)
        )
        tokens = re.findall(r"\b\w\w+\b", text, flags=re.UNICODE)
        terms: list[str] = []
        for ngram_size in range(minimum_n, maximum_n + 1):
            terms.extend(
                " ".join(tokens[index : index + ngram_size])
                for index in range(len(tokens) - ngram_size + 1)
            )
        counts: dict[int, int] = {}
        for term in terms:
            index = vocabulary.get(term)
            if index is not None:
                counts[index] = counts.get(index, 0) + 1

        weighted: dict[int, float] = {}
        for index, count in counts.items():
            term_frequency = 1.0 + math.log(count) if vectorizer["sublinear_tf"] else float(count)
            weighted[index] = term_frequency * idf[index]
        if vectorizer["norm"] == "l2" and weighted:
            magnitude = math.sqrt(sum(value * value for value in weighted.values()))
            weighted = {index: value / magnitude for index, value in weighted.items()}

        logits = [
            intercept + sum(row[index] * value for index, value in weighted.items())
            for row, intercept in zip(coefficients, intercepts)
        ]
        maximum = max(logits)
        exponentials = [math.exp(value - maximum) for value in logits]
        total = sum(exponentials)
        probabilities = [value / total for value in exponentials]
        predicted_index = max(range(len(probabilities)), key=probabilities.__getitem__)
        predictions.append(classes[predicted_index])
        probability_rows.append(probabilities)
    return predictions, probability_rows


def main() -> int:
    checks: list[str] = []
    missing = [name for name in REQUIRED_FILES if not (ROOT / name).is_file()]
    if missing:
        raise AssertionError(f"Required package files are missing: {missing}")
    checks.append(f"All {len(REQUIRED_FILES)} required package files exist")

    if sha256(ROOT / "sklearn_model.joblib") != sha256(ROOT / "model.joblib"):
        raise AssertionError("Hugging Face scikit-learn compatibility alias differs from model.joblib")
    checks.append("Hugging Face scikit-learn compatibility alias is byte-identical to model.joblib")

    train_ids, _, _ = read_split("train")
    validation_ids, x_validation, y_validation = read_split("validation")
    test_ids, x_test, y_test = read_split("test")
    if set(train_ids) & (set(validation_ids) | set(test_ids)) or set(validation_ids) & set(test_ids):
        raise AssertionError("Dataset split identifiers overlap")
    if (len(train_ids), len(validation_ids), len(test_ids)) != (128, 32, 32):
        raise AssertionError("Unexpected dataset split counts")
    checks.append("Training, validation and test IDs are disjoint with counts 128/32/32")

    with (ROOT / "metrics.json").open("r", encoding="utf-8") as handle:
        recorded = json.load(handle)
    with (ROOT / "model_config.json").open("r", encoding="utf-8") as handle:
        config = json.load(handle)
    if config.get("training_split_only") is not True or config.get("fit_record_count") != 128:
        raise AssertionError("Model config does not attest the training-only fit boundary")
    checks.append("Model configuration records a 128-record training-only fit boundary")

    expected_hashes = {
        "train": recorded["data"]["train_sha256"],
        "validation": recorded["data"]["validation_sha256"],
        "test": recorded["data"]["test_sha256"],
    }
    for split, expected in expected_hashes.items():
        if sha256(DATA_DIR / f"{split}.csv") != expected:
            raise AssertionError(f"{split} data hash differs from the evaluation record")
    checks.append("Training, validation and test hashes match the recorded evaluation inputs")

    model = joblib.load(ROOT / "model.joblib")
    expected_labels = recorded["labels"]
    if [str(label) for label in model.classes_] != expected_labels:
        raise AssertionError("Reloaded model labels do not match metrics.json")
    checks.append("Joblib artefact reloads with all eight labels in the recorded order")

    with (ROOT / "web_model.json").open("r", encoding="utf-8") as handle:
        web_model = json.load(handle)
    if web_model.get("format") != "nwhite-tfidf-logistic-regression-v1":
        raise AssertionError("Unexpected browser model format")
    vocabulary_size = len(web_model["vectorizer"]["vocabulary"])
    if vocabulary_size != len(web_model["vectorizer"]["idf"]):
        raise AssertionError("Browser vocabulary and IDF lengths differ")
    if web_model["classes"] != expected_labels:
        raise AssertionError("Browser model classes differ from the joblib model")
    if len(web_model["classifier"]["coef"]) != len(expected_labels):
        raise AssertionError("Browser classifier does not contain one coefficient row per class")
    if any(len(row) != vocabulary_size for row in web_model["classifier"]["coef"]):
        raise AssertionError("Browser coefficient width differs from the vocabulary size")
    if len(web_model["classifier"]["intercept"]) != len(expected_labels):
        raise AssertionError("Browser intercept count differs from the class count")
    checks.append(f"Browser JSON has eight classes and a consistent {vocabulary_size}-feature shape")

    for split, texts, truth in (
        ("validation", x_validation, y_validation),
        ("test", x_test, y_test),
    ):
        predictions = model.predict(texts).tolist()
        probabilities = model.predict_proba(texts)
        if probabilities.shape != (32, 8):
            raise AssertionError(f"Unexpected {split} probability shape: {probabilities.shape}")
        if any(not math.isclose(float(sum(row)), 1.0, rel_tol=0.0, abs_tol=1e-9) for row in probabilities):
            raise AssertionError(f"{split} probability row does not sum to one")
        accuracy = float(accuracy_score(truth, predictions))
        macro_f1 = float(f1_score(truth, predictions, labels=expected_labels, average="macro", zero_division=0))
        weighted_f1 = float(f1_score(truth, predictions, labels=expected_labels, average="weighted", zero_division=0))
        matrix = confusion_matrix(truth, predictions, labels=expected_labels).tolist()
        assert_close(accuracy, float(recorded[split]["accuracy"]), f"{split} accuracy")
        assert_close(macro_f1, float(recorded[split]["macro_f1"]), f"{split} macro F1")
        assert_close(weighted_f1, float(recorded[split]["weighted_f1"]), f"{split} weighted F1")
        if matrix != recorded[split]["confusion_matrix"]:
            raise AssertionError(f"{split} confusion matrix mismatch")
        checks.append(f"{split} predictions, probabilities, metrics and confusion matrix reproduce after reload")

        browser_predictions, browser_probabilities = web_predict(web_model, texts)
        if browser_predictions != predictions:
            raise AssertionError(f"Browser JSON {split} predictions differ from joblib")
        for row_index, (browser_row, joblib_row) in enumerate(zip(browser_probabilities, probabilities)):
            for class_index, (browser_value, joblib_value) in enumerate(zip(browser_row, joblib_row)):
                if not math.isclose(browser_value, float(joblib_value), rel_tol=0.0, abs_tol=1e-12):
                    raise AssertionError(
                        f"Browser JSON {split} probability mismatch at row {row_index}, class {class_index}"
                    )
        checks.append(f"Browser JSON {split} predictions and probabilities match joblib to 1e-12")

    with (ROOT / "sample_predictions.json").open("r", encoding="utf-8") as handle:
        samples = json.load(handle)["examples"]
    sample_predictions = model.predict([sample["text"] for sample in samples]).tolist()
    if sample_predictions != [sample["predicted_intent"] for sample in samples]:
        raise AssertionError("Reloaded smoke predictions differ from sample_predictions.json")
    checks.append(f"All {len(samples)} saved inference smoke predictions reproduce after reload")
    browser_sample_predictions, browser_sample_probabilities = web_predict(
        web_model,
        [sample["text"] for sample in samples],
    )
    if browser_sample_predictions != sample_predictions:
        raise AssertionError("Browser JSON smoke predictions differ from joblib")
    joblib_sample_probabilities = model.predict_proba([sample["text"] for sample in samples])
    for browser_row, joblib_row in zip(browser_sample_probabilities, joblib_sample_probabilities):
        for browser_value, joblib_value in zip(browser_row, joblib_row):
            if not math.isclose(browser_value, float(joblib_value), rel_tol=0.0, abs_tol=1e-12):
                raise AssertionError("Browser JSON smoke probabilities differ from joblib")
    checks.append("Browser JSON smoke predictions and probabilities match joblib to 1e-12")

    requirement_versions = {}
    with (ROOT / "requirements.txt").open("r", encoding="utf-8") as handle:
        for line in handle:
            package, expected_version = line.strip().split("==", maxsplit=1)
            actual_version = importlib.metadata.version(package)
            if actual_version != expected_version:
                raise AssertionError(
                    f"Installed {package} version {actual_version} differs from pinned {expected_version}"
                )
            requirement_versions[package] = actual_version
    checks.append(f"Installed dependency versions match all {len(requirement_versions)} exact pins")

    readme = (ROOT / "README.md").read_text(encoding="utf-8").lower()
    documented_hashes = {
        sha256(ROOT / "model.joblib"),
        sha256(ROOT / "web_model.json"),
        *expected_hashes.values(),
    }
    missing_documented_hashes = sorted(digest for digest in documented_hashes if digest not in readme)
    if missing_documented_hashes:
        raise AssertionError(f"README does not contain current artefact/data hashes: {missing_documented_hashes}")
    checks.append("Model card records the current model, browser export and three split hashes")

    scanned_files = 0
    for path in sorted(item for item in ROOT.rglob("*") if item.is_file()):
        if path.name == "SHA256SUMS" or "__pycache__" in path.parts:
            continue
        content = path.read_bytes()
        scanned_files += 1
        for name, pattern in SECRET_PATTERNS.items():
            if pattern.search(content):
                raise AssertionError(f"Potential {name} found in {path.relative_to(ROOT)}")
        if path.suffix.lower() in {".md", ".json", ".csv", ".txt", ".py", ".cff"}:
            text = content.decode("utf-8")
            if FORBIDDEN_TEXT.search(text):
                raise AssertionError(f"Forbidden phrase or placeholder found in {path.relative_to(ROOT)}")
    checks.append(f"Secret-pattern and forbidden-phrase scan passed across {scanned_files} package files")

    checks.append("SHA256SUMS records every package file except the manifest itself and Python bytecode caches")
    report = {
        "status": "passed",
        "model_version": "1.0.0",
        "checks_passed": len(checks),
        "checks": checks,
        "model_sha256": sha256(ROOT / "model.joblib"),
        "web_model_sha256": sha256(ROOT / "web_model.json"),
        "validation_accuracy": recorded["validation"]["accuracy"],
        "validation_macro_f1": recorded["validation"]["macro_f1"],
        "test_accuracy": recorded["test"]["accuracy"],
        "test_macro_f1": recorded["test"]["macro_f1"],
        "smoke_examples": len(samples),
        "smoke_matches_expected": sum(bool(sample["matches_expected"]) for sample in samples),
    }
    with (REPORT_DIR / "verification_report.json").open("w", encoding="utf-8", newline="\n") as handle:
        json.dump(report, handle, ensure_ascii=False, indent=2)
        handle.write("\n")

    manifest_targets = sorted(
        path for path in ROOT.rglob("*")
        if path.is_file() and path.name != "SHA256SUMS" and "__pycache__" not in path.parts
    )
    with (ROOT / "SHA256SUMS").open("w", encoding="utf-8", newline="\n") as handle:
        for path in manifest_targets:
            handle.write(f"{sha256(path)}  {path.relative_to(ROOT).as_posix()}\n")
    print(json.dumps(report, indent=2))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (AssertionError, OSError, ValueError) as exc:
        print(f"VERIFICATION FAILED: {exc}", file=sys.stderr)
        raise SystemExit(1)