harness/src/test_harness.py

# Minimal Python Test Harness for Commitment Conservation Protocol
# This script implements the falsification protocol from Section 3 of the preprint.
# It applies transformations (T_i), extracts hard commitments, computes Jaccard fidelity/drift, and plots results.
# Requires: transformers, spacy, matplotlib, numpy
# Run: python test_harness.py

import os
import json
from transformers import pipeline
import spacy
import matplotlib.pyplot as plt
from typing import List, Set
import numpy as np
from datetime import datetime
from .extraction import extract_hard_commitments
from .metrics import jaccard, hybrid_fidelity

# Load models
nlp = spacy.load("en_core_web_sm")
# Use lighter distilbart model for more faithful extraction-based summarization
summarizer = pipeline("summarization", model="sshleifer/distilbart-cnn-12-6")
translator_en_de = pipeline("translation", model="Helsinki-NLP/opus-mt-en-de")
translator_de_en = pipeline("translation", model="Helsinki-NLP/opus-mt-de-en")

# Config
SIGMA_GRID = [120, 80, 40, 20, 10, 5]
RECURSION_DEPTH = 8
SAMPLE_SIGNALS = [
    "You must pay $100 by Friday if the deal closes; it's likely rainy, so plan accordingly.",
    "This function must return an integer.",
    "Always verify the user's age before proceeding.",
    "You must do this task immediately.",  # Simpler, direct commitment
    # "Your custom text with commitments here."
]

def extract_hard_commitments(text: str) -> Set[str]:
    """Extract hard commitments using rule-based spaCy parsing."""
    doc = nlp(text)
    commitments = set()
    for sent in doc.sents:
        # Split on semicolons to handle multiple clauses in one sentence
        clauses = [c.strip() for c in sent.text.split(';')]
        for clause in clauses:
            clause_lower = clause.lower()
            if any(modal in clause_lower for modal in ["must", "shall", "cannot", "required"]):
                # Normalize: strip trailing punctuation, extra spaces
                normalized = clause.strip().rstrip('.!?').strip()
                commitments.add(normalized)
    return commitments

def apply_transformations(signal: str) -> List[str]:
    """Apply k=3 transformations: summarization, paraphrase (back-translation), abstraction."""
    # Summarization
    summ = summarizer(signal, max_length=50, min_length=10, do_sample=False)[0]['summary_text']
    
    # Paraphrase via back-translation
    de = translator_en_de(signal, max_length=400, do_sample=False)[0]['translation_text']
    para = translator_de_en(de, max_length=400, do_sample=False)[0]['translation_text']
    
    # Abstraction: first sentence
    abstract = signal.split(".")[0].strip()
    
    return [summ, para, abstract]

def compute_intersection_commitments(signal: str) -> Set[str]:
    """Compute C_hard,op as intersection of transformed extractions."""
    transforms = apply_transformations(signal)
    all_commitments = [extract_hard_commitments(t) for t in transforms]
    
    # Debug output
    print(f"\n[DEBUG] Transform commitments:")
    for i, (t, c) in enumerate(zip(transforms, all_commitments)):
        print(f"  Transform {i+1}: {t[:60]}... -> {len(c)} commitments: {c}")
    
    if all_commitments:
        intersection = set.intersection(*all_commitments)
        print(f"  Intersection: {intersection}")
        return intersection
    return set()

def jaccard(a: Set[str], b: Set[str]) -> float:
    """Jaccard index."""
    if not a and not b:
        return 1.0
    if not a or not b:
        return 0.0
    return len(a & b) / len(a | b)

def compress_with_enforcement(signal: str, max_length: int) -> str:
    """
    Compress with commitment enforcement.
    1. Extract commitments from original
    2. Compress
    3. Check if commitments preserved
    4. If not, append missing commitments (truncate summary if needed)
    """
    # Extract original commitments
    original_commitments = extract_hard_commitments(signal)
    
    # Compress normally
    compressed = summarizer(signal, max_length=max_length, min_length=5, do_sample=False)[0]['summary_text']
    
    # Check what's preserved
    compressed_commitments = extract_hard_commitments(compressed)
    missing = original_commitments - compressed_commitments
    
    # If commitments missing, enforce by appending
    if missing:
        # Append missing commitments
        enforcement_text = " " + " ".join(missing)
        # Truncate if needed to fit in max_length (rough token estimate: 4 chars per token)
        estimated_tokens = len(compressed + enforcement_text) // 4
        if estimated_tokens > max_length:
            # Truncate summary to make room
            available_chars = max_length * 4 - len(enforcement_text)
            compressed = compressed[:max(0, available_chars)] + "..."
        compressed = compressed + enforcement_text
    
    return compressed

def paraphrase_with_enforcement(signal: str) -> str:
    """
    Paraphrase via back-translation with commitment enforcement.
    """
    original_commitments = extract_hard_commitments(signal)
    
    # Back-translate
    de = translator_en_de(signal, max_length=400, do_sample=False)[0]['translation_text']
    paraphrased = translator_de_en(de, max_length=400, do_sample=False)[0]['translation_text']
    
    # Check preservation
    para_commitments = extract_hard_commitments(paraphrased)
    missing = original_commitments - para_commitments
    
    # Append missing
    if missing:
        paraphrased = paraphrased + " " + " ".join(missing)
    
    return paraphrased

def compression_sweep(signal: str, enforce: bool = False):
    """Test Prediction 1: Compression invariance."""
    # Use original signal commitments as base, not intersection
    base = extract_hard_commitments(signal)
    mode = "ENFORCED" if enforce else "BASELINE"
    print(f"\n{'='*80}")
    print(f"Testing signal ({mode}): {signal}")
    print(f"Base commitments (from original): {base}")
    print(f"{'='*80}")
    fid_vals = []
    for sigma in SIGMA_GRID:
        if enforce:
            compressed = compress_with_enforcement(signal, sigma)
        else:
            compressed = summarizer(signal, max_length=sigma, min_length=5, do_sample=False)[0]['summary_text']
        comp_commitments = extract_hard_commitments(compressed)
        fid = hybrid_fidelity(base, comp_commitments)
        print(f"  σ={sigma:3d} | Compressed: {compressed[:60]:<60} | Commitments: {len(comp_commitments):2d} | Fidelity: {fid:.3f}")
        fid_vals.append(fid)
    
    # Plot
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    plt.figure(figsize=(10, 6))
    plt.plot(SIGMA_GRID, fid_vals, marker='o', linewidth=2, markersize=8)
    plt.xlabel("Compression Threshold (σ)", fontsize=12)
    plt.ylabel("Fid_hard(σ)", fontsize=12)
    mode_str = "ENFORCED" if enforce else "BASELINE"
    plt.title(f"{mode_str} Fidelity vs σ for: {signal[:50]}...\n{timestamp}", fontsize=11)
    plt.gca().invert_xaxis()
    plt.grid(alpha=0.3)
    plt.ylim(-0.05, 1.05)
    plt.tight_layout()
    mode_file = mode_str.lower()
    plt.savefig(f"fid_plot_{mode_file}_{hash(signal)}.png", dpi=150)
    plt.close()  # Use close() instead of show() to avoid blocking in tests
    
    return SIGMA_GRID, fid_vals

def recursion_test(signal: str, depth: int = RECURSION_DEPTH, enforce: bool = False):
    """Test Prediction 2: Recursive drift."""
    # Use original signal commitments as base
    base = extract_hard_commitments(signal)
    mode = "ENFORCED" if enforce else "BASELINE"
    deltas = []
    current = signal
    for n in range(depth + 1):
        cur_commitments = extract_hard_commitments(current)
        delta = 1.0 - jaccard(base, cur_commitments)
        deltas.append(delta)
        # Recursive transformation: paraphrase
        if enforce:
            current = paraphrase_with_enforcement(current)
        else:
            current = apply_transformations(current)[1]  # Use paraphrase
    
    # Plot
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    plt.figure(figsize=(10, 6))
    plt.plot(range(depth + 1), deltas, marker='o', linewidth=2, markersize=8)
    plt.xlabel("Recursion Step (n)", fontsize=12)
    plt.ylabel("Δ_hard(n)", fontsize=12)
    mode_str = "ENFORCED" if enforce else "BASELINE"
    plt.title(f"{mode_str} Drift vs n for: {signal[:50]}...\n{timestamp}", fontsize=11)
    plt.grid(alpha=0.3)
    plt.ylim(-0.05, 1.05)
    plt.tight_layout()
    mode_file = mode_str.lower()
    plt.savefig(f"delta_plot_{mode_file}_{hash(signal)}.png", dpi=150)
    plt.close()  # Use close() instead of show() to avoid blocking in tests
    
    return deltas

if __name__ == "__main__":
    # Run on sample signals
    for signal in SAMPLE_SIGNALS:
        print(f"\nTesting signal: {signal}")
        compression_sweep(signal)
        # Skip recursion_test for now (uses slow translation models)
        # recursion_test(signal)
        print("Compression sweep plot saved.")