"""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)