File size: 4,303 Bytes
ab48644 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """
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
)
|