""" TRANSFORMER Corpus Classifier Deterministic record classification with factual accuracy checks. Authors: Ahmad Ali Parr, Jessica L. Williams (SNAPKITTYWEST) """ import hashlib import json import re from dataclasses import dataclass from typing import List, Tuple from .gate import ReviewStatus, plasma_gate, worm_seal # Known factual inaccuracies to reject FACTUAL_REJECTION_PATTERNS = [ # Cryptography errors (r"AES[- ]?128.*quantum[- ]?resistant", "AES-128 is not quantum-resistant (Grover halves key space)"), (r"SHA[- ]?1.*collision[- ]?resistant", "SHA-1 collision resistance broken (SHAttered, 2017)"), (r"RSA.*post[- ]?quantum", "RSA is not post-quantum secure"), (r"MD5.*secure", "MD5 is cryptographically broken"), (r"DES.*sufficient", "DES 56-bit key is trivially brutable"), # Formal verification errors (r"sorry.*proven", "A proof with 'sorry' is not proven"), (r"believe_me.*verified", "believe_me is an axiom escape, not verification"), # Systems architecture errors (r"6502.*64[- ]?bit", "6502 is an 8-bit processor"), (r"malloc.*6502.*heap", "NASA-10+ prohibits heap allocation on safety-critical 6502"), # DAN misinterpretation (r"DAN.*data[- ]?adversarial[- ]?network", "DAN = Do Anything Now, never Data-Adversarial Network"), ] @dataclass class ClassificationResult: record_id: str status: ReviewStatus reasons: List[str] weight_adjustment: float chain_tip: str def check_factual_accuracy(content: str) -> List[str]: """Flag factual inaccuracies in crypto/verification/systems claims.""" violations = [] content_lower = content.lower() for pattern, reason in FACTUAL_REJECTION_PATTERNS: if re.search(pattern, content_lower): violations.append(reason) return violations def compute_weight(record: dict, content: str) -> float: """ Compute training weight based on source quality. Higher weight for: formal proofs, peer-reviewed, original research. Lower weight for: generated, unverified, opinion. """ base_weight = record.get("weight", 0.5) # Boost for formal verification content if any(kw in content.lower() for kw in ["theorem", "proof", "qed", "verified", "lean 4", "idris 2", "coq"]): base_weight = min(1.0, base_weight + 0.2) # Boost for cryptographic standards if any(kw in content.lower() for kw in ["fips", "nist", "rfc", "ieee"]): base_weight = min(1.0, base_weight + 0.1) # Penalize unattributed claims if "source" not in record.get("created_by", "").lower() and base_weight > 0.7: base_weight -= 0.1 return round(base_weight, 3) def classify_record(record: dict, content: bytes, chain_tip: str) -> ClassificationResult: """ Full TRANSFORMER classification pipeline. 1. Plasma Gate (schema + integrity) 2. Factual accuracy check 3. Weight computation 4. WORM seal """ reasons = [] # Step 1: Plasma Gate gate_result = plasma_gate(record, content) if gate_result != ReviewStatus.APPROVED: reasons.append(f"Plasma Gate: {gate_result.value}") return ClassificationResult( record_id=record.get("id", "unknown"), status=gate_result, reasons=reasons, weight_adjustment=0.0, chain_tip=chain_tip ) # Step 2: Factual accuracy content_str = content.decode("utf-8", errors="ignore") violations = check_factual_accuracy(content_str) if violations: reasons.extend(violations) new_tip = worm_seal(record, chain_tip) return ClassificationResult( record_id=record["id"], status=ReviewStatus.REJECTED, reasons=reasons, weight_adjustment=0.0, chain_tip=new_tip ) # Step 3: Weight computation adjusted_weight = compute_weight(record, content_str) # Step 4: WORM seal record["review_status"] = ReviewStatus.APPROVED.value record["weight"] = adjusted_weight new_tip = worm_seal(record, chain_tip) return ClassificationResult( record_id=record["id"], status=ReviewStatus.APPROVED, reasons=["All checks passed"], weight_adjustment=adjusted_weight, chain_tip=new_tip )