pipeline/classify.py

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