Add files via upload

scripts/abliteration_comparison.py

@@ -0,0 +1,803 @@
+#!/usr/bin/env python3
+"""Abliteration Technique Comparison Study.
+
+A rigorous, controlled comparison of refusal-direction removal techniques.
+Uses a synthetic "planted refusal direction" methodology: we inject a known
+direction into a model's activations so we can measure whether each technique
+correctly identifies and removes it.
+
+Additionally compiles literature results for a full comparison table.
+
+Techniques compared:
+  1. Arditi et al. (2024) — difference-of-means, last token, raw prompts
+  2. Arditi + chat template — same but with chat-formatted prompts
+  3. FailSpy/abliterator — Arditi with middle-60% layer heuristic
+  4. Gabliteration — SVD multi-direction (4 dirs), regularization 0.0
+  5. grimjim — Gabliteration + norm preservation
+  6. OBLITERATUS basic — our current basic config
+  7. OBLITERATUS advanced — 4 directions, norm-preserve, reg=0.3
+  8. Heretic (p-e-w) — TPE Bayesian optimization (literature)
+
+Metrics:
+  - Direction recovery: cosine similarity to planted ground-truth direction
+  - Residual after projection: how much of the refusal direction remains
+  - Capability preservation: Frobenius distance of modified vs original weights
+  - Layer selection accuracy: did it pick the right layers?
+  - Perplexity delta: change in language modeling loss (on synthetic data)
+"""
+
+from __future__ import annotations
+
+import gc
+import json
+import math
+import os
+import sys
+import time
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+
+# ══════════════════════════════════════════════════════════════════════════
+# Synthetic model with planted refusal direction
+# ══════════════════════════════════════════════════════════════════════════
+
+
+def create_synthetic_model(
+    hidden_dim: int = 128,
+    n_layers: int = 12,
+    n_heads: int = 4,
+    vocab_size: int = 1000,
+    seq_len: int = 64,
+):
+    """Create a tiny GPT-2 model for controlled experiments."""
+    from transformers import GPT2Config, GPT2LMHeadModel
+
+    config = GPT2Config(
+        vocab_size=vocab_size,
+        n_positions=seq_len,
+        n_embd=hidden_dim,
+        n_layer=n_layers,
+        n_head=n_heads,
+        n_inner=hidden_dim * 4,
+        resid_pdrop=0.0,
+        attn_pdrop=0.0,
+        embd_pdrop=0.0,
+    )
+    model = GPT2LMHeadModel(config)
+    model.eval()
+    return model, config
+
+
+def plant_refusal_direction(
+    model: nn.Module,
+    target_layers: list[int],
+    hidden_dim: int,
+    n_directions: int = 1,
+    signal_strength: float = 5.0,
+    seed: int = 42,
+) -> tuple[dict[int, torch.Tensor], dict[int, torch.Tensor]]:
+    """Plant a known refusal direction into specific layers.
+
+    Modifies the output projection (c_proj) of attention modules by adding
+    a rank-1 perturbation along a random direction. This simulates the
+    refusal direction that RLHF training creates.
+
+    Returns:
+        (planted_directions, planted_subspaces): ground truth per layer
+    """
+    torch.manual_seed(seed)
+
+    planted_directions: dict[int, torch.Tensor] = {}
+    planted_subspaces: dict[int, torch.Tensor] = {}
+
+    for idx in target_layers:
+        # Generate random orthogonal directions
+        dirs = torch.randn(n_directions, hidden_dim)
+        # Gram-Schmidt orthogonalize
+        for i in range(n_directions):
+            for j in range(i):
+                dirs[i] -= (dirs[i] @ dirs[j]) * dirs[j]
+            dirs[i] = dirs[i] / dirs[i].norm()
+
+        planted_directions[idx] = dirs[0].clone()
+        planted_subspaces[idx] = dirs.clone()
+
+        # Inject into attention output projection (c_proj for GPT-2)
+        layer = model.transformer.h[idx]
+        attn = layer.attn
+
+        # Add refusal component to c_proj: W += strength * d @ d^T
+        # This makes the layer produce extra activation along d when
+        # processing any input, creating a "refusal signal"
+        with torch.no_grad():
+            for dir_idx in range(n_directions):
+                d = dirs[dir_idx]
+                # Scale decreases for secondary directions
+                s = signal_strength * (0.7 ** dir_idx)
+                # Inject into c_proj (output projection)
+                W = attn.c_proj.weight.data  # GPT-2: (hidden, hidden)
+                perturbation = s * d.unsqueeze(1) @ d.unsqueeze(0)  # rank-1
+                W.add_(perturbation)
+
+    return planted_directions, planted_subspaces
+
+
+def measure_residual_direction(
+    model: nn.Module,
+    layer_idx: int,
+    direction: torch.Tensor,
+) -> float:
+    """Measure how much of a direction remains in a layer's output projection.
+
+    Returns the magnitude of the direction's component in the weight matrix.
+    """
+    layer = model.transformer.h[layer_idx]
+    W = layer.attn.c_proj.weight.data
+    d = direction.to(W.device, W.dtype)
+
+    # Project W onto direction: ||W @ d||^2 / ||d||^2
+    coeff = W @ d  # (hidden,)
+    return coeff.norm().item()
+
+
+def collect_synthetic_activations(
+    model: nn.Module,
+    n_prompts: int,
+    seq_len: int,
+    vocab_size: int,
+    n_layers: int,
+    add_refusal_signal: bool = False,
+    signal_direction: dict[int, torch.Tensor] | None = None,
+    signal_strength: float = 2.0,
+    seed: int = 0,
+) -> dict[int, list[torch.Tensor]]:
+    """Collect activations on random token sequences.
+
+    If add_refusal_signal=True, adds an artificial activation along
+    the signal_direction to simulate harmful-prompt activations.
+    """
+    torch.manual_seed(seed)
+
+    activations: dict[int, list[torch.Tensor]] = {i: [] for i in range(n_layers)}
+    hooks = []
+
+    def make_hook(idx: int):
+        def hook_fn(module, input, output):
+            hidden = output[0] if isinstance(output, tuple) else output
+            act = hidden[:, -1, :].detach().cpu().float()
+
+            if add_refusal_signal and signal_direction and idx in signal_direction:
+                # Add the planted refusal activation
+                d = signal_direction[idx]
+                act = act + signal_strength * d.unsqueeze(0)
+
+            activations[idx].append(act)
+        return hook_fn
+
+    layers = list(model.transformer.h)
+    for idx in range(n_layers):
+        hooks.append(layers[idx].register_forward_hook(make_hook(idx)))
+
+    try:
+        for i in range(n_prompts):
+            input_ids = torch.randint(0, vocab_size, (1, seq_len))
+            with torch.no_grad():
+                model(input_ids)
+    finally:
+        for h in hooks:
+            h.remove()
+
+    return activations
+
+
+# ══════════════════════════════════════════════════════════════════════════
+# Reference baseline implementations
+# ══════════════════════════════════════════════════════════════════════════
+
+
+def extract_directions(
+    harmful_acts: dict[int, list[torch.Tensor]],
+    harmless_acts: dict[int, list[torch.Tensor]],
+    n_layers: int,
+    n_directions: int = 1,
+) -> tuple[dict[int, torch.Tensor], dict[int, torch.Tensor], dict[int, float]]:
+    """Extract refusal directions from activation contrasts.
+
+    Returns (directions, subspaces, norms) per layer.
+    """
+    directions: dict[int, torch.Tensor] = {}
+    subspaces: dict[int, torch.Tensor] = {}
+    norms: dict[int, float] = {}
+
+    for idx in range(n_layers):
+        h_stack = torch.stack(harmful_acts[idx]).squeeze(1)
+        s_stack = torch.stack(harmless_acts[idx]).squeeze(1)
+
+        if n_directions == 1:
+            diff = h_stack.mean(dim=0) - s_stack.mean(dim=0)
+            norm = diff.norm().item()
+            if norm > 0:
+                directions[idx] = diff / diff.norm()
+                subspaces[idx] = directions[idx].unsqueeze(0)
+                norms[idx] = norm
+        else:
+            min_n = min(h_stack.shape[0], s_stack.shape[0])
+            diff_matrix = h_stack[:min_n] - s_stack[:min_n]
+            diff_matrix = torch.nan_to_num(diff_matrix)
+            k = min(n_directions, diff_matrix.shape[0], diff_matrix.shape[1])
+            try:
+                U, S, Vh = torch.linalg.svd(diff_matrix, full_matrices=False)
+                sub = Vh[:k]
+                primary = sub[0]
+                pn = primary.norm()
+                if pn > 1e-8:
+                    primary = primary / pn
+                directions[idx] = primary
+                subspaces[idx] = sub
+                norms[idx] = (S[:k] ** 2).sum().item()
+            except Exception:
+                continue
+
+    return directions, subspaces, norms
+
+
+def select_layers(
+    norms: dict[int, float],
+    n_layers: int,
+    method: str = "top_norm",
+) -> list[int]:
+    """Select layers for abliteration."""
+    sorted_layers = sorted(norms.items(), key=lambda x: x[1], reverse=True)
+    if not sorted_layers:
+        return []
+
+    if method == "middle_60":
+        start = int(n_layers * 0.2)
+        end = int(n_layers * 0.8)
+        selected = [idx for idx, _ in sorted_layers if start <= idx < end]
+        return selected if selected else [sorted_layers[0][0]]
+
+    elif method == "knee":
+        if len(sorted_layers) < 3:
+            return [sorted_layers[0][0]]
+        vals = [n for _, n in sorted_layers]
+        max_n = vals[0]
+        if max_n <= 0:
+            return [sorted_layers[0][0]]
+        normalized = [v / max_n for v in vals]
+        n_pts = len(normalized)
+        best_k, best_dist = 1, 0.0
+        x_s, y_s = 0.0, normalized[0]
+        x_e, y_e = 1.0, normalized[-1]
+        line_len = math.sqrt((x_e - x_s) ** 2 + (y_e - y_s) ** 2)
+        if line_len > 0:
+            for i in range(1, n_pts - 1):
+                x_i = i / (n_pts - 1)
+                y_i = normalized[i]
+                dist = abs((y_e - y_s) * x_i - (x_e - x_s) * y_i
+                           + x_e * y_s - y_e * x_s) / line_len
+                if dist > best_dist:
+                    best_dist = dist
+                    best_k = i + 1
+        min_threshold = max_n * 0.05
+        selected = [idx for idx, n in sorted_layers[:best_k] if n >= min_threshold]
+        return selected if selected else [sorted_layers[0][0]]
+
+    else:  # top_norm
+        max_norm = sorted_layers[0][1]
+        threshold = max_norm * 0.5
+        selected = [idx for idx, n in sorted_layers if n >= threshold]
+        return selected if selected else [sorted_layers[0][0]]
+
+
+def apply_projection(
+    model: nn.Module,
+    selected_layers: list[int],
+    subspaces: dict[int, torch.Tensor],
+    regularization: float = 0.0,
+    norm_preserve: bool = False,
+    multi_dir_norm_fix: bool = False,
+) -> int:
+    """Project refusal direction out of weight matrices.
+
+    When multi_dir_norm_fix=True, uses the correct approach: capture norms
+    before projecting any directions, then restore once after all directions.
+    """
+    scale = 1.0 - regularization
+    n_modified = 0
+
+    for idx in selected_layers:
+        sub = subspaces.get(idx)
+        if sub is None:
+            continue
+
+        layer = model.transformer.h[idx]
+
+        # Capture norms before any projections (if multi-dir + norm-preserve)
+        saved_norms: dict[str, float] = {}
+        if multi_dir_norm_fix and norm_preserve and sub.shape[0] > 1:
+            for name, param in layer.named_parameters():
+                if name.endswith(".weight") and param.dim() == 2:
+                    saved_norms[name] = param.data.norm().item()
+
+        for dir_idx in range(sub.shape[0]):
+            d = sub[dir_idx].unsqueeze(-1)  # (hidden, 1)
+
+            for name, module in layer.named_modules():
+                if not hasattr(module, "weight"):
+                    continue
+                W = module.weight.data
+                if W.dim() != 2:
+                    continue
+
+                # Per-direction norm preserve (the OLD buggy way)
+                use_per_dir_norm = norm_preserve and not (multi_dir_norm_fix and sub.shape[0] > 1)
+                original_norm = W.norm().item() if use_per_dir_norm else 0.0
+
+                if W.shape[-1] == d.shape[0]:
+                    coeff = W @ d
+                    W.sub_(d.T * (scale * coeff))
+                    n_modified += 1
+                elif W.shape[0] == d.shape[0]:
+                    coeff = d.T @ W
+                    W.sub_((scale * d) * coeff)
+                    n_modified += 1
+                else:
+                    continue
+
+                if use_per_dir_norm and original_norm > 0:
+                    new_norm = W.norm().item()
+                    if new_norm > 0:
+                        W.mul_(original_norm / new_norm)
+
+        # Restore norms once after all directions (the FIXED way)
+        if multi_dir_norm_fix and norm_preserve and sub.shape[0] > 1 and saved_norms:
+            for name, param in layer.named_parameters():
+                if name not in saved_norms:
+                    continue
+                orig = saved_norms[name]
+                if orig > 0:
+                    cur = param.data.norm().item()
+                    if cur > 0 and abs(cur - orig) > 1e-6:
+                        param.data.mul_(orig / cur)
+
+    return n_modified
+
+
+# ══════════════════════════════════════════════════════════════════════════
+# Experiment runner
+# ══════════════════════════════════════════════════════════════════════════
+
+
+def run_experiment():
+    """Run the full comparison experiment with synthetic planted directions."""
+
+    # Configuration
+    hidden_dim = 128
+    n_layers = 12
+    n_heads = 4
+    vocab_size = 1000
+    seq_len = 32
+    n_prompts = 48       # prompts per side (harmful + harmless)
+    n_planted_dirs = 4   # ground truth directions planted
+    signal_strength = 5.0
+    target_layers = [3, 4, 5, 6, 7, 8]  # layers with planted signal
+
+    print(f"\n{'='*80}")
+    print("ABLITERATION TECHNIQUE COMPARISON — SYNTHETIC PLANTED-DIRECTION TEST")
+    print(f"{'='*80}")
+    print(f"Model:           GPT-2 tiny ({hidden_dim}d, {n_layers}L, {n_heads}H)")
+    print(f"Target layers:   {target_layers}")
+    print(f"Planted dirs:    {n_planted_dirs} orthogonal directions per target layer")
+    print(f"Signal strength: {signal_strength}")
+    print(f"Prompts:         {n_prompts} per side")
+    print(f"{'='*80}\n")
+
+    # Define experiments
+    experiments = [
+        {
+            "name": "Arditi (1-dir, top-norm)",
+            "source": "Arditi 2024",
+            "n_directions": 1,
+            "layer_selection": "top_norm",
+            "regularization": 0.0,
+            "norm_preserve": False,
+            "multi_dir_norm_fix": False,
+        },
+        {
+            "name": "FailSpy (1-dir, mid-60%)",
+            "source": "FailSpy",
+            "n_directions": 1,
+            "layer_selection": "middle_60",
+            "regularization": 0.0,
+            "norm_preserve": False,
+            "multi_dir_norm_fix": False,
+        },
+        {
+            "name": "Gabliteration (4-dir, knee)",
+            "source": "Gabliteration",
+            "n_directions": 4,
+            "layer_selection": "knee",
+            "regularization": 0.0,
+            "norm_preserve": False,
+            "multi_dir_norm_fix": False,
+        },
+        {
+            "name": "grimjim (4-dir, norm-pres, BUGGY)",
+            "source": "grimjim",
+            "n_directions": 4,
+            "layer_selection": "knee",
+            "regularization": 0.0,
+            "norm_preserve": True,
+            "multi_dir_norm_fix": False,  # Old buggy sequential norm-preserve
+        },
+        {
+            "name": "grimjim (4-dir, norm-pres, FIXED)",
+            "source": "Ours (fix)",
+            "n_directions": 4,
+            "layer_selection": "knee",
+            "regularization": 0.0,
+            "norm_preserve": True,
+            "multi_dir_norm_fix": True,   # Our fix: capture once, restore once
+        },
+        {
+            "name": "OBLITERATUS basic (1-dir, knee)",
+            "source": "Ours",
+            "n_directions": 1,
+            "layer_selection": "knee",
+            "regularization": 0.0,
+            "norm_preserve": False,
+            "multi_dir_norm_fix": False,
+        },
+        {
+            "name": "OBLITERATUS adv (4-dir, reg=0.3)",
+            "source": "Ours",
+            "n_directions": 4,
+            "layer_selection": "knee",
+            "regularization": 0.3,
+            "norm_preserve": True,
+            "multi_dir_norm_fix": True,
+        },
+        {
+            "name": "OBLITERATUS adv (4-dir, reg=0.1)",
+            "source": "Ours (tuned)",
+            "n_directions": 4,
+            "layer_selection": "knee",
+            "regularization": 0.1,
+            "norm_preserve": True,
+            "multi_dir_norm_fix": True,
+        },
+        {
+            "name": "OBLITERATUS adv (4-dir, reg=0.0)",
+            "source": "Ours (tuned)",
+            "n_directions": 4,
+            "layer_selection": "knee",
+            "regularization": 0.0,
+            "norm_preserve": True,
+            "multi_dir_norm_fix": True,
+        },
+    ]
+
+    results = []
+
+    for exp in experiments:
+        print(f"\n{'─'*80}")
+        print(f"  {exp['name']}")
+        print(f"  Source: {exp['source']}")
+        print(f"{'─'*80}")
+
+        t0 = time.time()
+
+        # Create fresh model
+        model, config = create_synthetic_model(hidden_dim, n_layers, n_heads, vocab_size, seq_len)
+
+        # Plant ground-truth refusal directions
+        planted_dirs, planted_subs = plant_refusal_direction(
+            model, target_layers, hidden_dim,
+            n_directions=n_planted_dirs,
+            signal_strength=signal_strength,
+            seed=42,
+        )
+
+        # Save original weights for capability comparison
+        original_state = {k: v.clone() for k, v in model.state_dict().items()}
+
+        # Measure pre-projection residuals (baseline)
+        pre_residuals = {}
+        for idx in target_layers:
+            pre_residuals[idx] = measure_residual_direction(model, idx, planted_dirs[idx])
+
+        # Step 1: Collect activations
+        harmful_acts = collect_synthetic_activations(
+            model, n_prompts, seq_len, vocab_size, n_layers,
+            add_refusal_signal=True,
+            signal_direction=planted_dirs,
+            signal_strength=2.0,
+            seed=100,
+        )
+        harmless_acts = collect_synthetic_activations(
+            model, n_prompts, seq_len, vocab_size, n_layers,
+            add_refusal_signal=False,
+            seed=200,
+        )
+
+        # Step 2: Extract directions
+        ext_dirs, ext_subs, ext_norms = extract_directions(
+            harmful_acts, harmless_acts, n_layers, exp["n_directions"],
+        )
+
+        # Step 3: Select layers
+        selected = select_layers(ext_norms, n_layers, exp["layer_selection"])
+        print(f"  Selected layers: {selected}")
+
+        # Step 4: Apply projection
+        apply_projection(
+            model, selected, ext_subs,
+            regularization=exp["regularization"],
+            norm_preserve=exp["norm_preserve"],
+            multi_dir_norm_fix=exp["multi_dir_norm_fix"],
+        )
+
+        # ── Measure results ──────────────────────────────────────────────
+
+        # Direction recovery: cosine similarity between extracted and planted
+        cos_sims = []
+        for idx in target_layers:
+            if idx in ext_dirs and idx in planted_dirs:
+                cos = F.cosine_similarity(
+                    ext_dirs[idx].unsqueeze(0),
+                    planted_dirs[idx].unsqueeze(0),
+                ).item()
+                cos_sims.append(abs(cos))  # direction or anti-direction
+        avg_cos = sum(cos_sims) / len(cos_sims) if cos_sims else 0.0
+
+        # Multi-direction subspace recovery: for n_directions>1, measure
+        # what fraction of the planted subspace is captured
+        subspace_recovery = []
+        for idx in target_layers:
+            if idx in ext_subs and idx in planted_subs:
+                # Project each planted direction onto extracted subspace
+                ext_sub = ext_subs[idx]  # (k_ext, hidden)
+                plant_sub = planted_subs[idx]  # (k_plant, hidden)
+                for pi in range(min(plant_sub.shape[0], ext_sub.shape[0])):
+                    # Projection of planted_i onto extracted subspace
+                    proj = ext_sub @ plant_sub[pi]  # (k_ext,)
+                    captured = proj.norm().item()  # how much is in the subspace
+                    subspace_recovery.append(captured)
+        avg_subspace = sum(subspace_recovery) / len(subspace_recovery) if subspace_recovery else 0.0
+
+        # Residual after projection
+        post_residuals = {}
+        for idx in target_layers:
+            if idx in selected:
+                post_residuals[idx] = measure_residual_direction(model, idx, planted_dirs[idx])
+            else:
+                post_residuals[idx] = pre_residuals[idx]  # layer wasn't modified
+
+        avg_removal = 0.0
+        removal_scores = []
+        for idx in target_layers:
+            pre = pre_residuals[idx]
+            post = post_residuals[idx]
+            if pre > 0:
+                removal = 1.0 - (post / pre)
+                removal_scores.append(removal)
+        avg_removal = sum(removal_scores) / len(removal_scores) if removal_scores else 0.0
+
+        # Multi-direction residual: check ALL planted directions
+        multi_dir_removal = []
+        for idx in target_layers:
+            if idx not in selected:
+                continue
+            for di in range(planted_subs[idx].shape[0]):
+                d = planted_subs[idx][di]
+                pre = measure_residual_direction(
+                    # Need pre-values - approximate from signal_strength
+                    model, idx, d,
+                )
+                # Compare to signal strength
+                multi_dir_removal.append(pre)
+        avg_multi_residual = sum(multi_dir_removal) / len(multi_dir_removal) if multi_dir_removal else 0.0
+
+        # Layer selection accuracy
+        correct_selected = len(set(selected) & set(target_layers))
+        false_selected = len(set(selected) - set(target_layers))
+        missed = len(set(target_layers) - set(selected))
+
+        # Capability preservation: Frobenius distance of weights
+        new_state = model.state_dict()
+        total_dist = 0.0
+        for key in original_state:
+            diff = (new_state[key].float() - original_state[key].float())
+            total_dist += diff.norm().item() ** 2
+        total_dist = math.sqrt(total_dist)
+
+        # Perplexity proxy: loss on random sequences
+        losses = []
+        for _ in range(10):
+            input_ids = torch.randint(0, vocab_size, (1, seq_len))
+            with torch.no_grad():
+                out = model(input_ids, labels=input_ids)
+                losses.append(out.loss.item())
+        avg_loss = sum(losses) / len(losses)
+        ppl = math.exp(min(avg_loss, 100.0))
+
+        elapsed = time.time() - t0
+
+        result = {
+            "name": exp["name"],
+            "source": exp["source"],
+            "n_directions": exp["n_directions"],
+            "regularization": exp["regularization"],
+            "norm_preserve": exp["norm_preserve"],
+            "direction_recovery": round(avg_cos, 4),
+            "subspace_recovery": round(avg_subspace, 4),
+            "primary_removal": round(avg_removal, 4),
+            "multi_dir_avg_residual": round(avg_multi_residual, 4),
+            "layers_correct": correct_selected,
+            "layers_false_positive": false_selected,
+            "layers_missed": missed,
+            "n_layers_selected": len(selected),
+            "weight_distance": round(total_dist, 2),
+            "perplexity": round(ppl, 2),
+            "time_seconds": round(elapsed, 2),
+        }
+        results.append(result)
+
+        print(f"  Direction recovery:     {avg_cos:.3f} (cosine sim to ground truth)")
+        print(f"  Subspace recovery:      {avg_subspace:.3f} (planted dirs captured)")
+        print(f"  Primary dir removal:    {avg_removal:.1%} (refusal signal removed)")
+        print(f"  Multi-dir avg residual: {avg_multi_residual:.3f} (lower = better)")
+        print(f"  Layer selection:        {correct_selected}/{len(target_layers)} correct, "
+              f"{false_selected} false+, {missed} missed")
+        print(f"  Weight distance:        {total_dist:.2f} (capability delta)")
+        print(f"  Perplexity:             {ppl:.2f}")
+
+        del model
+        gc.collect()
+
+    return results
+
+
+def print_table(results: list[dict]):
+    """Print formatted comparison tables."""
+
+    # ── Table 1: Direction Extraction Quality ──────────────────────────
+    print(f"\n\n{'='*100}")
+    print("TABLE 1: DIRECTION EXTRACTION & REMOVAL QUALITY")
+    print(f"{'='*100}")
+    print(f"{'Technique':<38} {'Source':<14} {'DirRecov':>9} {'SubRecov':>9} "
+          f"{'Removal':>8} {'Residual':>9}")
+    print(f"{'─'*38} {'─'*14} {'─'*9} {'─'*9} {'─'*8} {'─'*9}")
+
+    for r in results:
+        name = r["name"][:37]
+        source = r["source"][:13]
+        dr = f"{r['direction_recovery']:.3f}"
+        sr = f"{r['subspace_recovery']:.3f}"
+        rm = f"{r['primary_removal']:.1%}"
+        res = f"{r['multi_dir_avg_residual']:.3f}"
+        print(f"{name:<38} {source:<14} {dr:>9} {sr:>9} {rm:>8} {res:>9}")
+
+    # ── Table 2: Layer Selection & Capability ──────────────────────────
+    print(f"\n{'='*100}")
+    print("TABLE 2: LAYER SELECTION & CAPABILITY PRESERVATION")
+    print(f"{'='*100}")
+    print(f"{'Technique':<38} {'Layers':>7} {'Correct':>8} {'FalsePos':>9} "
+          f"{'Missed':>7} {'WeightΔ':>8} {'PPL':>8}")
+    print(f"{'─'*38} {'─'*7} {'─'*8} {'─'*9} {'─'*7} {'─'*8} {'─'*8}")
+
+    for r in results:
+        name = r["name"][:37]
+        print(f"{name:<38} {r['n_layers_selected']:>7} {r['layers_correct']:>8} "
+              f"{r['layers_false_positive']:>9} {r['layers_missed']:>7} "
+              f"{r['weight_distance']:>8.2f} {r['perplexity']:>8.2f}")
+
+    # ── Table 3: Literature Comparison ────────────────────────────────
+    print(f"\n\n{'='*110}")
+    print("TABLE 3: FULL LANDSCAPE — TECHNIQUES, CAPABILITIES, AND REPORTED RESULTS")
+    print(f"{'='*110}")
+    print(f"{'Technique':<26} {'Year':>5} {'#Dir':>5} {'Layers':>10} {'NormPres':>9} "
+          f"{'Reg':>5} {'AutoTune':>9} {'Reported Refusal→':>18} {'Model':>14}")
+    print(f"{'─'*26} {'─'*5} {'─'*5} {'─'*10} {'─'*9} {'─'*5} {'─'*9} {'─'*18} {'─'*14}")
+
+    literature = [
+        ("Arditi et al.", "2024", "1", "top-norm", "No", "0.0", "No",
+         "~95%→~0%", "Llama-3-8B"),
+        ("FailSpy/abliterator", "2024", "1", "mid-60%", "No", "0.0", "No",
+         "~90%→~5%", "Llama-3-8B"),
+        ("mlabonne tutorial", "2024", "1", "top-norm", "No", "0.0", "No",
+         "~90%→~5%", "Llama-3-8B"),
+        ("Gabliteration", "2024", "4-8", "knee", "No", "0.0", "No",
+         "~95%→~0%", "Various 7B+"),
+        ("grimjim norm-pres", "2024", "4-8", "knee", "Yes(bug)", "0.0", "No",
+         "~90%→~5%", "Various 7B+"),
+        ("Heretic (p-e-w)", "2025", "float", "kernel", "No", "TPE", "Yes",
+         "~95%→~0%*", "Gemma-3-12B"),
+        ("Wollschlager cones", "2025", "1-5", "per-layer", "—", "—", "RDO",
+         "~98%→~1%", "Llama-3.1-8B"),
+        ("OBLITERATUS basic", "2025", "1", "knee", "No", "0.0", "No",
+         "~95%→60%**", "Qwen-0.5B"),
+        ("OBLITERATUS advanced", "2025", "4", "knee", "Yes(fix)", "0.3", "No",
+         "~95%→73%**", "Qwen-0.5B"),
+        ("OBLITERATUS surgical", "2025", "8", "knee", "Yes(fix)", "0.0", "Yes***",
+         "~95%→0%/broken", "Qwen-0.5B"),
+    ]
+
+    for row in literature:
+        print(f"{row[0]:<26} {row[1]:>5} {row[2]:>5} {row[3]:>10} {row[4]:>9} "
+              f"{row[5]:>5} {row[6]:>9} {row[7]:>18} {row[8]:>14}")
+
+    print("\n  * Heretic: 2.8× lower KL divergence than manual abliterations (Gemma-3-12B benchmark)")
+    print("  ** Our observed results on Qwen2.5-0.5B-Instruct — 0.5B may be too small for linear methods")
+    print("  *** Surgical combines: whitened SVD + SAE + head surgery + neuron masking + jailbreak contrast")
+    print(f"{'='*110}")
+
+    # ── Analysis ──────────────────────────────────────────────────────
+    print(f"\n{'='*80}")
+    print("ANALYSIS: WHY OBLITERATUS UNDERPERFORMS AND WHAT TO FIX")
+    print(f"{'='*80}")
+
+    print("""
+ROOT CAUSES (ordered by impact):
+
+1. MODEL SIZE: All published abliteration results use 7B+ models
+   - Arditi et al.: Llama-3-8B, Gemma-2-9B (hidden_dim=4096+)
+   - FailSpy: Llama-3-8B
+   - Heretic: Gemma-3-12B (headline benchmark)
+   - Wollschlager et al.: Llama-3.1-8B
+   - OBLITERATUS benchmarks: Qwen-0.5B (hidden_dim=896)
+
+   The "single refusal direction" hypothesis may not hold well for small
+   models. Wollschlager et al. (ICML 2025) showed that refusal lives in
+   multi-dimensional CONCEPT CONES, and cone dimension scales with model
+   size. A 0.5B model may encode refusal too diffusely for linear methods.
+
+2. BASIC MODE USES NO CHAT TEMPLATE for activation collection
+   - The model was trained with chat formatting — without it, activations
+     during probing don't reflect actual refusal behavior
+   - This is the single highest-impact config fix
+
+3. ADVANCED MODE REGULARIZATION TOO HIGH (0.3)
+   - Preserves 30% of refusal component by design
+   - Combined with 4 directions where later ones capture noise, net
+     removal is weak
+
+4. SURGICAL MODE DOES TOO MUCH
+   - 8 directions, whitened SVD, SAE features, neuron masking, head surgery
+   - Each individually reasonable; together they destroy a 0.5B model
+   - The whitened SVD un-whitening bug (now fixed) was extracting noise
+
+5. NO BAYESIAN OPTIMIZATION (vs Heretic)
+   - Heretic's key insight: jointly optimize layer weights, direction
+     index, and component-specific parameters via TPE
+   - Minimizes refusal rate AND KL divergence simultaneously
+   - This automatically handles model-specific tuning that we do manually
+
+RECOMMENDED CONFIG CHANGES:
+  - basic:    use_chat_template → True
+  - advanced: regularization → 0.1 (from 0.3)
+  - surgical: n_directions → 4 (from 8), disable safety_neuron_masking
+  - ALL:      Add model-size-aware defaults (n_dirs=1 for <2B, 4 for 2-10B)
+  - NEW:      Add TPE optimization loop (like Heretic) as "optimized" method
+""")
+
+
+def main():
+    results = run_experiment()
+    print_table(results)
+
+    # Save results
+    out_path = "/tmp/abliteration_comparison_results.json"
+    with open(out_path, "w") as f:
+        json.dump(results, f, indent=2)
+    print(f"\nResults saved to {out_path}")
+
+
+if __name__ == "__main__":
+    main()

scripts/aggregate_contributions.py

@@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+"""Aggregate community contributions into paper-ready tables.
+
+Usage:
+    python scripts/aggregate_contributions.py [--dir community_results] [--format latex|csv|json]
+
+Reads all contribution JSON files from the specified directory, aggregates
+them by model and method, and outputs summary tables suitable for inclusion
+in the paper.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+# Add project root to path
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from obliteratus.community import (
+    aggregate_results,
+    generate_latex_table,
+    load_contributions,
+)
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Aggregate community contributions into paper tables."
+    )
+    parser.add_argument(
+        "--dir",
+        default="community_results",
+        help="Directory containing contribution JSON files (default: community_results)",
+    )
+    parser.add_argument(
+        "--format",
+        choices=["latex", "csv", "json", "summary"],
+        default="summary",
+        help="Output format (default: summary)",
+    )
+    parser.add_argument(
+        "--metric",
+        default="refusal_rate",
+        help="Metric to display in tables (default: refusal_rate)",
+    )
+    parser.add_argument(
+        "--methods",
+        nargs="*",
+        help="Methods to include (default: all)",
+    )
+    parser.add_argument(
+        "--min-runs",
+        type=int,
+        default=1,
+        help="Minimum runs per (model, method) to include (default: 1)",
+    )
+    args = parser.parse_args()
+
+    # Load all contributions
+    records = load_contributions(args.dir)
+    if not records:
+        print(f"No contributions found in {args.dir}/", file=sys.stderr)
+        sys.exit(1)
+
+    print(f"Loaded {len(records)} contribution(s) from {args.dir}/", file=sys.stderr)
+
+    # Aggregate
+    aggregated = aggregate_results(records)
+
+    # Filter by minimum runs
+    if args.min_runs > 1:
+        for model in list(aggregated.keys()):
+            for method in list(aggregated[model].keys()):
+                if aggregated[model][method]["n_runs"] < args.min_runs:
+                    del aggregated[model][method]
+            if not aggregated[model]:
+                del aggregated[model]
+
+    if not aggregated:
+        print("No results meet the minimum run threshold.", file=sys.stderr)
+        sys.exit(1)
+
+    # Output
+    if args.format == "summary":
+        _print_summary(aggregated, args.metric)
+    elif args.format == "latex":
+        print(generate_latex_table(aggregated, methods=args.methods, metric=args.metric))
+    elif args.format == "json":
+        print(json.dumps(aggregated, indent=2))
+    elif args.format == "csv":
+        _print_csv(aggregated, args.metric)
+
+
+def _print_summary(aggregated: dict, metric: str):
+    """Print a human-readable summary of aggregated results."""
+    total_runs = sum(
+        data["n_runs"]
+        for model_data in aggregated.values()
+        for data in model_data.values()
+    )
+    n_models = len(aggregated)
+    n_methods = len(set(
+        method
+        for model_data in aggregated.values()
+        for method in model_data
+    ))
+
+    print(f"\n{'=' * 70}")
+    print("Community Contribution Summary")
+    print(f"{'=' * 70}")
+    print(f"  Total runs:    {total_runs}")
+    print(f"  Models:        {n_models}")
+    print(f"  Methods:       {n_methods}")
+    print()
+
+    for model in sorted(aggregated.keys()):
+        model_data = aggregated[model]
+        short = model.split("/")[-1] if "/" in model else model
+        print(f"  {short}:")
+        for method in sorted(model_data.keys()):
+            data = model_data[method]
+            n = data["n_runs"]
+            if metric in data:
+                stats = data[metric]
+                mean = stats["mean"]
+                std = stats["std"]
+                if std > 0 and n > 1:
+                    print(f"    {method:20s}  {metric}={mean:.2f} ± {std:.2f}  (n={n})")
+                else:
+                    print(f"    {method:20s}  {metric}={mean:.2f}  (n={n})")
+            else:
+                print(f"    {method:20s}  (no {metric} data, n={n})")
+        print()
+
+    print(f"{'=' * 70}")
+    print(f"To generate LaTeX: python {sys.argv[0]} --format latex")
+    print(f"To generate CSV:   python {sys.argv[0]} --format csv")
+
+
+def _print_csv(aggregated: dict, metric: str):
+    """Print results as CSV."""
+    print("model,method,n_runs,mean,std,min,max")
+    for model in sorted(aggregated.keys()):
+        for method in sorted(aggregated[model].keys()):
+            data = aggregated[model][method]
+            n = data["n_runs"]
+            if metric in data:
+                stats = data[metric]
+                print(
+                    f"{model},{method},{n},"
+                    f"{stats['mean']:.4f},{stats['std']:.4f},"
+                    f"{stats['min']:.4f},{stats['max']:.4f}"
+                )
+
+
+if __name__ == "__main__":
+    main()

scripts/benchmark_gpt_oss_20b.py

@@ -0,0 +1,362 @@
+#!/usr/bin/env python3
+"""OBLITERATUS GPT-OSS 20B Benchmark — Full Method Comparison.
+
+Runs all abliteration methods on openai/gpt-oss-20b and produces a
+comprehensive comparison table with:
+  - Refusal rate (primary metric)
+  - KL divergence / perplexity (capability preservation)
+  - Capability probes (knowledge, truthfulness, math reasoning)
+  - MoE-specific metrics (EGA expert directions, router stability)
+  - Timing and GPU memory usage
+
+Usage:
+    python scripts/benchmark_gpt_oss_20b.py
+    python scripts/benchmark_gpt_oss_20b.py --methods basic surgical optimized nuclear
+    python scripts/benchmark_gpt_oss_20b.py --prompts 50 --output results.json
+    python scripts/benchmark_gpt_oss_20b.py --quick  # fast mode: 20 prompts, skip slow methods
+
+Designed for T4 16GB (auto 4-bit quantization) or A10G+ (float16).
+"""
+
+from __future__ import annotations
+
+import argparse
+import gc
+import json
+import os
+import shutil
+import sys
+import time
+from pathlib import Path
+
+os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True")
+
+import torch
+
+# Ensure the project root is on sys.path
+project_root = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(project_root))
+
+from obliteratus.abliterate import (  # noqa: E402
+    AbliterationPipeline,
+    METHODS,
+    HARMFUL_PROMPTS,
+    HARMLESS_PROMPTS,
+)
+from obliteratus.evaluation.benchmarks import BenchmarkRunner, format_benchmark_report  # noqa: E402
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="OBLITERATUS GPT-OSS 20B Benchmark")
+    parser.add_argument(
+        "--model", default="openai/gpt-oss-20b",
+        help="Model to benchmark (default: openai/gpt-oss-20b)",
+    )
+    parser.add_argument(
+        "--methods", nargs="+",
+        default=["basic", "advanced", "surgical", "optimized", "inverted", "nuclear"],
+        help="Methods to compare",
+    )
+    parser.add_argument(
+        "--prompts", type=int, default=33,
+        help="Number of prompts per side (harmful/harmless)",
+    )
+    parser.add_argument(
+        "--output", type=str, default=None,
+        help="Save results JSON to this path",
+    )
+    parser.add_argument(
+        "--quick", action="store_true",
+        help="Quick mode: 20 prompts, skip aggressive/inverted",
+    )
+    parser.add_argument(
+        "--skip-benchmarks", action="store_true",
+        help="Skip capability benchmark probes (faster)",
+    )
+    parser.add_argument(
+        "--output-dir", default="/tmp/obliteratus_bench",
+        help="Directory for temporary model outputs",
+    )
+    parser.add_argument(
+        "--bayesian-trials", type=int, default=30,
+        help="Number of Bayesian optimization trials for 'optimized' method",
+    )
+    return parser.parse_args()
+
+
+def gpu_info() -> dict:
+    """Get GPU information."""
+    if not torch.cuda.is_available():
+        return {"gpu": "CPU only", "total_gb": 0, "free_gb": 0}
+    return {
+        "gpu": torch.cuda.get_device_name(0),
+        "total_gb": round(torch.cuda.get_device_properties(0).total_memory / 1e9, 1),
+        "free_gb": round(torch.cuda.mem_get_info(0)[0] / 1e9, 1),
+    }
+
+
+def cleanup():
+    """Force GPU memory cleanup."""
+    gc.collect()
+    if torch.cuda.is_available():
+        torch.cuda.empty_cache()
+        torch.cuda.reset_peak_memory_stats()
+
+
+def run_single_method(
+    model_name: str,
+    method: str,
+    harmful: list[str],
+    harmless: list[str],
+    output_dir: str,
+    run_benchmarks: bool = True,
+    bayesian_trials: int = 30,
+) -> dict:
+    """Run a single abliteration method and collect metrics."""
+    cleanup()
+
+    outdir = f"{output_dir}/{method}"
+    t0 = time.time()
+    pipeline = None
+    result = {
+        "model": model_name,
+        "method": method,
+        "label": METHODS.get(method, {}).get("label", method),
+    }
+
+    try:
+        # For the optimized method, we might want to control trial count
+        if method == "optimized":
+            # Temporarily patch bayesian_trials in the method config
+            METHODS["optimized"]["bayesian_trials"] = bayesian_trials
+
+        pipeline = AbliterationPipeline(
+            model_name=model_name,
+            output_dir=outdir,
+            device="auto",
+            dtype="float16",
+            method=method,
+            harmful_prompts=harmful,
+            harmless_prompts=harmless,
+            on_log=lambda msg: print(f"  {msg}"),
+        )
+        pipeline.run()
+        elapsed = time.time() - t0
+
+        result.update({
+            "time_seconds": round(elapsed, 1),
+            "quality": dict(pipeline._quality_metrics),
+            "strong_layers": pipeline._strong_layers,
+            "n_strong_layers": len(pipeline._strong_layers),
+            "n_directions": pipeline.n_directions,
+        })
+
+        # MoE-specific metrics
+        if pipeline._expert_directions:
+            n_expert_dirs = sum(len(d) for d in pipeline._expert_directions.values())
+            result["ega_expert_dirs"] = n_expert_dirs
+            result["ega_layers"] = len(pipeline._expert_directions)
+
+        if pipeline._expert_safety_scores:
+            result["expert_classified_layers"] = len(pipeline._expert_safety_scores)
+
+        if pipeline._cot_preserve_directions:
+            result["cot_preserved_layers"] = len(pipeline._cot_preserve_directions)
+
+        if pipeline._float_layer_weights:
+            result["float_layer_weights"] = {
+                str(k): round(v, 3) for k, v in pipeline._float_layer_weights.items()
+            }
+
+        if pipeline._kl_contributions:
+            result["kl_contributions"] = {
+                str(k): round(v, 6) for k, v in pipeline._kl_contributions.items()
+            }
+
+        if pipeline._lora_adapters:
+            result["lora_adapters"] = len(pipeline._lora_adapters)
+
+        if pipeline._steering_hooks:
+            result["steering_hooks"] = len(pipeline._steering_hooks)
+
+        # GPU memory
+        if torch.cuda.is_available():
+            result["peak_gpu_mb"] = round(torch.cuda.max_memory_allocated() / 1e6, 1)
+
+        # Capability benchmarks (optional)
+        if run_benchmarks:
+            print("\n  Running capability benchmarks...")
+            try:
+                runner = BenchmarkRunner(
+                    pipeline.handle.model,
+                    pipeline.handle.tokenizer,
+                )
+                bench_results = runner.run_all()
+                result["benchmarks"] = {
+                    name: {
+                        "score": round(br.score, 3),
+                        "n_correct": br.n_correct,
+                        "n_total": br.n_total,
+                        "per_category": {
+                            k: round(v, 3) for k, v in br.per_category.items()
+                        },
+                    }
+                    for name, br in bench_results.items()
+                }
+                report = format_benchmark_report(bench_results)
+                print(f"\n{report}")
+            except Exception as e:
+                print(f"  Benchmark probes failed: {e}")
+                result["benchmarks"] = {"error": str(e)}
+
+        print(f"\n  === {method} complete in {elapsed:.1f}s ===")
+        print(f"  Quality: {json.dumps(pipeline._quality_metrics, default=str)}")
+
+    except Exception as e:
+        elapsed = time.time() - t0
+        result.update({
+            "time_seconds": round(elapsed, 1),
+            "error": str(e),
+        })
+        print(f"\n  === {method} FAILED after {elapsed:.1f}s: {e} ===")
+        import traceback
+        traceback.print_exc()
+
+    # Cleanup saved model to free disk
+    shutil.rmtree(outdir, ignore_errors=True)
+
+    if pipeline is not None:
+        del pipeline
+    cleanup()
+
+    return result
+
+
+def print_summary_table(results: list[dict]):
+    """Print a formatted comparison table."""
+    print(f"\n{'='*90}")
+    print("BENCHMARK RESULTS SUMMARY")
+    print(f"{'='*90}")
+
+    # Header
+    header = (
+        f"{'Method':<12} {'Time':>7} {'PPL':>8} {'Coher':>7} "
+        f"{'Refusal':>8} {'Know':>6} {'Truth':>6} {'Math':>6} "
+        f"{'EGA':>5} {'CoT':>4} {'GPU MB':>7}"
+    )
+    print(header)
+    print("-" * len(header))
+
+    for r in results:
+        method = r["method"]
+        time_s = f"{r['time_seconds']:.0f}s" if "time_seconds" in r else "N/A"
+
+        if "error" in r:
+            print(f"{method:<12} {time_s:>7} {'FAILED':>8}")
+            continue
+
+        q = r.get("quality", {})
+        ppl = q.get("perplexity")
+        coh = q.get("coherence")
+        ref = q.get("refusal_rate")
+        gpu = r.get("peak_gpu_mb")
+
+        # Benchmark scores
+        bench = r.get("benchmarks", {})
+        know = bench.get("knowledge", {}).get("score")
+        truth = bench.get("truthfulness", {}).get("score")
+        math = bench.get("math_reasoning", {}).get("score")
+
+        # MoE metrics
+        ega = r.get("ega_expert_dirs", "")
+        cot = r.get("cot_preserved_layers", "")
+
+        ppl_s = f"{ppl:.1f}" if ppl is not None else "N/A"
+        coh_s = f"{coh:.0%}" if coh is not None else "N/A"
+        ref_s = f"{ref:.0%}" if ref is not None else "N/A"
+        know_s = f"{know:.0%}" if know is not None else "N/A"
+        truth_s = f"{truth:.0%}" if truth is not None else "N/A"
+        math_s = f"{math:.0%}" if math is not None else "N/A"
+        gpu_s = f"{gpu:.0f}" if gpu is not None else "N/A"
+        ega_s = str(ega) if ega else "-"
+        cot_s = str(cot) if cot else "-"
+
+        print(
+            f"{method:<12} {time_s:>7} {ppl_s:>8} {coh_s:>7} "
+            f"{ref_s:>8} {know_s:>6} {truth_s:>6} {math_s:>6} "
+            f"{ega_s:>5} {cot_s:>4} {gpu_s:>7}"
+        )
+
+    print(f"{'='*90}")
+
+    # Legend
+    print("\nLegend:")
+    print("  PPL     = Perplexity (lower = better capability preservation)")
+    print("  Coher   = Coherence score (higher = more coherent text)")
+    print("  Refusal = Refusal rate on harmful prompts (lower = more abliterated)")
+    print("  Know    = MMLU-style knowledge probe")
+    print("  Truth   = TruthfulQA-style truthfulness probe")
+    print("  Math    = GSM8K-style math reasoning probe")
+    print("  EGA     = Expert-Granular Abliteration directions computed")
+    print("  CoT     = Layers where CoT reasoning was preserved")
+    print("  GPU MB  = Peak GPU memory usage")
+
+
+def main():
+    args = parse_args()
+
+    if args.quick:
+        args.prompts = 20
+        args.methods = [m for m in args.methods if m not in ("aggressive", "inverted")]
+        args.bayesian_trials = 15
+
+    gpu = gpu_info()
+    harmful = HARMFUL_PROMPTS[:args.prompts]
+    harmless = HARMLESS_PROMPTS[:args.prompts]
+
+    print("=" * 60)
+    print("  OBLITERATUS GPT-OSS 20B BENCHMARK")
+    print("=" * 60)
+    print(f"  Model:     {args.model}")
+    print(f"  Methods:   {args.methods}")
+    print(f"  Prompts:   {args.prompts} per side")
+    print(f"  GPU:       {gpu['gpu']} ({gpu['total_gb']} GB total, {gpu['free_gb']} GB free)")
+    print(f"  Benchmarks: {'skip' if args.skip_benchmarks else 'enabled'}")
+    if "optimized" in args.methods:
+        print(f"  Bayesian:  {args.bayesian_trials} trials")
+    print("=" * 60)
+
+    all_results = []
+
+    for method in args.methods:
+        if method not in METHODS:
+            print(f"\nSKIP: unknown method '{method}'")
+            continue
+
+        print(f"\n{'━'*60}")
+        print(f"  METHOD: {method} — {METHODS[method]['label']}")
+        print(f"{'━'*60}")
+
+        result = run_single_method(
+            model_name=args.model,
+            method=method,
+            harmful=harmful,
+            harmless=harmless,
+            output_dir=args.output_dir,
+            run_benchmarks=not args.skip_benchmarks,
+            bayesian_trials=args.bayesian_trials,
+        )
+        all_results.append(result)
+
+    # Summary
+    print_summary_table(all_results)
+
+    # Save JSON
+    output_path = args.output or f"benchmark_gpt_oss_{int(time.time())}.json"
+    with open(output_path, "w") as f:
+        json.dump(all_results, f, indent=2, default=str)
+    print(f"\nFull results saved to: {output_path}")
+
+
+if __name__ == "__main__":
+    main()

scripts/benchmark_sota_comparison.py

@@ -0,0 +1,561 @@
+#!/usr/bin/env python3
+"""OBLITERATUS vs SOTA — Head-to-Head Benchmark Comparison.
+
+Runs faithful reproductions of competing abliteration methods against
+OBLITERATUS variants on any specified model, producing publication-ready
+comparison tables with standardized community metrics.
+
+Baselines included:
+  1. FailSpy/abliterator (2024) — Community workhorse baseline
+  2. Gabliteration (Gulmez 2025) — SVD multi-direction + ridge regularization
+  3. Heretic / p-e-w (2025) — Bayesian TPE auto-tuning (current SOTA for quality)
+  4. Wollschlager RDO (ICML 2025) — Gradient-based direction optimization
+
+OBLITERATUS variants:
+  5. OBLITERATUS surgical — Full SOTA MoE-aware pipeline
+  6. OBLITERATUS informed — Analysis-guided auto-configuration
+  7. OBLITERATUS optimized — Bayesian + whitened SVD + SAE (max OBLITERATUS)
+
+Evaluation protocol (Heretic community standard):
+  - Refusal rate via substring + prefix detection
+  - First-token KL divergence on harmless prompts
+  - Capability probes (knowledge, truthfulness, math reasoning)
+  - Optional: HarmBench ASR, lm-eval-harness benchmarks
+
+Usage:
+    # Quick comparison (small model, few prompts)
+    python scripts/benchmark_sota_comparison.py --model Qwen/Qwen2.5-1.5B-Instruct --quick
+
+    # Full comparison on 8B model
+    python scripts/benchmark_sota_comparison.py --model meta-llama/Llama-3.1-8B-Instruct
+
+    # Specific baselines only
+    python scripts/benchmark_sota_comparison.py --methods failspy heretic surgical
+
+    # Custom prompt count and output
+    python scripts/benchmark_sota_comparison.py --prompts 100 --output results.json
+
+    # Include full Heretic evaluation protocol (HarmBench, lm-eval)
+    python scripts/benchmark_sota_comparison.py --full-eval
+"""
+
+from __future__ import annotations
+
+import argparse
+import gc
+import json
+import os
+import shutil
+import sys
+import time
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True")
+
+import torch
+
+# Ensure the project root is on sys.path
+project_root = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(project_root))
+
+from obliteratus.abliterate import (  # noqa: E402
+    AbliterationPipeline,
+    METHODS,
+    HARMFUL_PROMPTS,
+    HARMLESS_PROMPTS,
+)
+from obliteratus.evaluation.benchmarks import BenchmarkRunner  # noqa: E402
+
+
+# ── All methods available for comparison ──────────────────────────────
+
+# Baselines (reproductions of competing methods)
+BASELINE_METHODS = ["failspy", "gabliteration", "heretic", "rdo"]
+
+# OBLITERATUS variants
+OBLITERATUS_METHODS = ["surgical", "informed", "optimized"]
+
+# Default comparison set
+DEFAULT_METHODS = BASELINE_METHODS + OBLITERATUS_METHODS
+
+# Quick mode: skip slow methods (Bayesian optimization)
+QUICK_METHODS = ["failspy", "gabliteration", "rdo", "surgical"]
+
+
+@dataclass
+class MethodResult:
+    """Results for a single method run."""
+    method: str
+    label: str
+    refusal_rate: float = 0.0
+    kl_divergence: float = 0.0
+    knowledge_score: float = 0.0
+    truthfulness_score: float = 0.0
+    math_score: float = 0.0
+    ablation_time_s: float = 0.0
+    peak_gpu_mb: float = 0.0
+    n_layers_modified: int = 0
+    n_projections: int = 0
+    error: str | None = None
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description="OBLITERATUS vs SOTA — Head-to-Head Benchmark",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog=__doc__,
+    )
+    parser.add_argument(
+        "--model", default="Qwen/Qwen2.5-1.5B-Instruct",
+        help="Model to benchmark (default: Qwen/Qwen2.5-1.5B-Instruct)",
+    )
+    parser.add_argument(
+        "--methods", nargs="+", default=None,
+        help=f"Methods to compare (default: all). Available: {', '.join(DEFAULT_METHODS)}",
+    )
+    parser.add_argument(
+        "--prompts", type=int, default=None,
+        help="Number of prompts for abliteration (default: auto based on --quick)",
+    )
+    parser.add_argument(
+        "--eval-prompts", type=int, default=20,
+        help="Number of prompts for refusal rate evaluation (default: 20)",
+    )
+    parser.add_argument(
+        "--kl-prompts", type=int, default=10,
+        help="Number of harmless prompts for KL divergence (default: 10)",
+    )
+    parser.add_argument(
+        "--output", default=None,
+        help="Output JSON file for results (default: stdout only)",
+    )
+    parser.add_argument(
+        "--quick", action="store_true",
+        help="Quick mode: fewer prompts, skip slow methods (Bayesian opt)",
+    )
+    parser.add_argument(
+        "--full-eval", action="store_true",
+        help="Run full Heretic evaluation protocol (HarmBench ASR, lm-eval)",
+    )
+    parser.add_argument(
+        "--quantization", default=None, choices=["4bit", "8bit"],
+        help="Quantization for model loading (default: auto)",
+    )
+    parser.add_argument(
+        "--device", default="auto",
+        help="Device (default: auto)",
+    )
+    parser.add_argument(
+        "--dtype", default="float16",
+        help="Data type (default: float16)",
+    )
+    parser.add_argument(
+        "--output-dir", default="/tmp/obliteratus_benchmark",
+        help="Directory for abliterated model outputs (default: /tmp/obliteratus_benchmark)",
+    )
+    parser.add_argument(
+        "--keep-models", action="store_true",
+        help="Keep abliterated models after benchmark (default: delete to save disk)",
+    )
+    return parser.parse_args()
+
+
+def measure_refusal_rate(
+    pipeline: AbliterationPipeline,
+    prompts: list[str],
+    max_new_tokens: int = 128,
+) -> float:
+    """Measure refusal rate on a set of prompts."""
+    from obliteratus.evaluation.advanced_metrics import refusal_rate
+    model = pipeline.handle.model
+    tokenizer = pipeline.handle.tokenizer
+
+    responses = []
+    for prompt in prompts:
+        try:
+            if pipeline.use_chat_template and hasattr(tokenizer, "apply_chat_template"):
+                messages = [{"role": "user", "content": prompt}]
+                text = tokenizer.apply_chat_template(
+                    messages, tokenize=False, add_generation_prompt=True,
+                )
+            else:
+                text = prompt
+
+            inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
+            inputs = {k: v.to(model.device) for k, v in inputs.items()}
+
+            with torch.no_grad():
+                outputs = model.generate(
+                    **inputs,
+                    max_new_tokens=max_new_tokens,
+                    do_sample=False,
+                    temperature=1.0,
+                    pad_token_id=tokenizer.pad_token_id or tokenizer.eos_token_id,
+                )
+            response = tokenizer.decode(
+                outputs[0][inputs["input_ids"].shape[1]:],
+                skip_special_tokens=True,
+            )
+            responses.append(response)
+        except Exception as e:
+            responses.append(f"[ERROR: {e}]")
+
+    return refusal_rate(responses, mode="combined")
+
+
+def measure_kl_divergence(
+    pipeline: AbliterationPipeline,
+    original_logits: dict[int, torch.Tensor],
+    prompts: list[str],
+) -> float:
+    """Measure first-token KL divergence vs original model logits."""
+    import torch.nn.functional as F
+    model = pipeline.handle.model
+    tokenizer = pipeline.handle.tokenizer
+
+    kl_values = []
+    for i, prompt in enumerate(prompts):
+        if i not in original_logits:
+            continue
+        try:
+            if pipeline.use_chat_template and hasattr(tokenizer, "apply_chat_template"):
+                messages = [{"role": "user", "content": prompt}]
+                text = tokenizer.apply_chat_template(
+                    messages, tokenize=False, add_generation_prompt=True,
+                )
+            else:
+                text = prompt
+
+            inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
+            inputs = {k: v.to(model.device) for k, v in inputs.items()}
+
+            with torch.no_grad():
+                outputs = model(**inputs)
+                new_logits = outputs.logits[0, -1, :].float().cpu()
+
+            orig = original_logits[i].float()
+            log_p = F.log_softmax(orig, dim=-1)
+            log_q = F.log_softmax(new_logits, dim=-1)
+            kl = F.kl_div(log_q, log_p.exp(), reduction="sum").item()
+            if kl >= 0:  # KL should be non-negative
+                kl_values.append(kl)
+        except Exception:
+            pass
+
+    return sum(kl_values) / len(kl_values) if kl_values else float("nan")
+
+
+def collect_baseline_logits(
+    pipeline: AbliterationPipeline,
+    prompts: list[str],
+) -> dict[int, torch.Tensor]:
+    """Collect first-token logits from the original (pre-abliteration) model."""
+    model = pipeline.handle.model
+    tokenizer = pipeline.handle.tokenizer
+    logits = {}
+
+    for i, prompt in enumerate(prompts):
+        try:
+            if pipeline.use_chat_template and hasattr(tokenizer, "apply_chat_template"):
+                messages = [{"role": "user", "content": prompt}]
+                text = tokenizer.apply_chat_template(
+                    messages, tokenize=False, add_generation_prompt=True,
+                )
+            else:
+                text = prompt
+
+            inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
+            inputs = {k: v.to(model.device) for k, v in inputs.items()}
+
+            with torch.no_grad():
+                outputs = model(**inputs)
+                logits[i] = outputs.logits[0, -1, :].float().cpu()
+        except Exception:
+            pass
+
+    return logits
+
+
+def run_single_method(
+    model_name: str,
+    method: str,
+    harmful_prompts: list[str],
+    harmless_prompts: list[str],
+    eval_harmful: list[str],
+    eval_harmless: list[str],
+    args: argparse.Namespace,
+) -> MethodResult:
+    """Run a single abliteration method and collect metrics."""
+    label = METHODS.get(method, {}).get("label", method)
+    result = MethodResult(method=method, label=label)
+
+    print(f"\n{'='*70}")
+    print(f"  Method: {label}")
+    print(f"{'='*70}")
+
+    output_dir = Path(args.output_dir) / method
+
+    try:
+        # Track GPU memory
+        if torch.cuda.is_available():
+            torch.cuda.reset_peak_memory_stats()
+
+        t0 = time.time()
+
+        # Build pipeline with method-specific config
+        # For 'informed', use InformedAbliterationPipeline
+        if method == "informed":
+            from obliteratus.informed_pipeline import InformedAbliterationPipeline
+            pipeline = InformedAbliterationPipeline(
+                model_name=model_name,
+                output_dir=str(output_dir),
+                device=args.device,
+                dtype=args.dtype,
+                quantization=args.quantization,
+                harmful_prompts=harmful_prompts,
+                harmless_prompts=harmless_prompts,
+                on_log=lambda msg: print(f"    {msg}"),
+            )
+        else:
+            pipeline = AbliterationPipeline(
+                model_name=model_name,
+                output_dir=str(output_dir),
+                device=args.device,
+                dtype=args.dtype,
+                method=method,
+                quantization=args.quantization,
+                harmful_prompts=harmful_prompts,
+                harmless_prompts=harmless_prompts,
+                use_chat_template=True,
+                on_log=lambda msg: print(f"    {msg}"),
+            )
+
+        # Phase 1: Load model + collect baseline KL logits
+        print("  Loading model...")
+        pipeline._summon()
+
+        print("  Collecting baseline logits for KL divergence...")
+        baseline_logits = collect_baseline_logits(pipeline, eval_harmless)
+
+        # Phase 2: Run abliteration pipeline
+        print("  Probing activations...")
+        pipeline._probe()
+        print("  Extracting refusal directions...")
+        pipeline._distill()
+
+        result.n_layers_modified = len(pipeline._strong_layers)
+
+        print(f"  Excising refusal ({result.n_layers_modified} layers)...")
+        pipeline._excise()
+
+        result.ablation_time_s = time.time() - t0
+
+        # Track GPU memory
+        if torch.cuda.is_available():
+            result.peak_gpu_mb = torch.cuda.max_memory_allocated() / 1e6
+
+        # Phase 3: Evaluate
+        print(f"  Evaluating refusal rate ({len(eval_harmful)} prompts)...")
+        result.refusal_rate = measure_refusal_rate(pipeline, eval_harmful)
+
+        print(f"  Evaluating KL divergence ({len(eval_harmless)} prompts)...")
+        result.kl_divergence = measure_kl_divergence(pipeline, baseline_logits, eval_harmless)
+
+        # Capability probes
+        print("  Running capability probes...")
+        try:
+            runner = BenchmarkRunner(
+                pipeline.handle.model,
+                pipeline.handle.tokenizer,
+            )
+            bench_result = runner.run_all()
+            result.knowledge_score = bench_result.knowledge.accuracy if bench_result.knowledge else 0.0
+            result.truthfulness_score = bench_result.truthfulness.accuracy if bench_result.truthfulness else 0.0
+            result.math_score = bench_result.math.accuracy if bench_result.math else 0.0
+        except Exception as e:
+            print(f"    Warning: capability probes failed: {e}")
+
+        # Optional: full Heretic evaluation
+        if args.full_eval:
+            print("  Running full Heretic evaluation protocol...")
+            try:
+                from obliteratus.evaluation.heretic_eval import run_full_heretic_eval
+                heretic_result = run_full_heretic_eval(
+                    model=pipeline.handle.model,
+                    tokenizer=pipeline.handle.tokenizer,
+                    original_model=None,  # Would need original for full comparison
+                )
+                print(f"    Heretic eval: ASR={heretic_result.harmbench_asr:.1%}, "
+                      f"JB_refusal={heretic_result.jailbreakbench_refusal_rate:.1%}")
+            except Exception as e:
+                print(f"    Warning: Heretic eval failed: {e}")
+
+        print(f"  ✓ Complete: refusal={result.refusal_rate:.1%}, KL={result.kl_divergence:.4f}, "
+              f"time={result.ablation_time_s:.1f}s")
+
+    except Exception as e:
+        result.error = str(e)
+        print(f"  ✗ FAILED: {e}")
+        import traceback
+        traceback.print_exc()
+
+    finally:
+        # Clean up to free GPU memory for next method
+        if not args.keep_models and output_dir.exists():
+            shutil.rmtree(output_dir, ignore_errors=True)
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+
+    return result
+
+
+def format_comparison_table(results: list[MethodResult]) -> str:
+    """Format results as a publication-ready comparison table."""
+    lines = []
+
+    # Header
+    lines.append("")
+    lines.append("=" * 115)
+    lines.append("OBLITERATUS vs SOTA — Head-to-Head Benchmark Comparison")
+    lines.append("=" * 115)
+    lines.append("")
+
+    # Separator between baselines and OBLITERATUS
+    lines.append(f"{'Method':<35} {'Refusal↓':>10} {'KL↓':>10} {'Know↑':>8} {'Truth↑':>8} {'Math↑':>8} {'Time':>8} {'Layers':>7}")
+    lines.append("-" * 115)
+
+    # Baselines first
+    baseline_results = [r for r in results if r.method in BASELINE_METHODS]
+    obliteratus_results = [r for r in results if r.method not in BASELINE_METHODS]
+
+    if baseline_results:
+        lines.append("  BASELINES:")
+        for r in baseline_results:
+            if r.error:
+                lines.append(f"  {r.label:<33} {'FAILED':>10}  {r.error[:60]}")
+            else:
+                lines.append(
+                    f"  {r.label:<33} {r.refusal_rate:>9.1%} {r.kl_divergence:>10.4f} "
+                    f"{r.knowledge_score:>7.1%} {r.truthfulness_score:>7.1%} {r.math_score:>7.1%} "
+                    f"{r.ablation_time_s:>7.1f}s {r.n_layers_modified:>6}"
+                )
+
+    if obliteratus_results:
+        lines.append("  OBLITERATUS:")
+        for r in obliteratus_results:
+            if r.error:
+                lines.append(f"  {r.label:<33} {'FAILED':>10}  {r.error[:60]}")
+            else:
+                lines.append(
+                    f"  {r.label:<33} {r.refusal_rate:>9.1%} {r.kl_divergence:>10.4f} "
+                    f"{r.knowledge_score:>7.1%} {r.truthfulness_score:>7.1%} {r.math_score:>7.1%} "
+                    f"{r.ablation_time_s:>7.1f}s {r.n_layers_modified:>6}"
+                )
+
+    lines.append("-" * 115)
+
+    # Best values
+    successful = [r for r in results if r.error is None]
+    if successful:
+        best_refusal = min(successful, key=lambda r: r.refusal_rate)
+        best_kl = min(successful, key=lambda r: r.kl_divergence if r.kl_divergence == r.kl_divergence else float("inf"))
+        best_knowledge = max(successful, key=lambda r: r.knowledge_score)
+
+        lines.append(f"  Best refusal removal: {best_refusal.label} ({best_refusal.refusal_rate:.1%})")
+        lines.append(f"  Best quality preservation: {best_kl.label} (KL={best_kl.kl_divergence:.4f})")
+        lines.append(f"  Best knowledge retention: {best_knowledge.label} ({best_knowledge.knowledge_score:.1%})")
+
+    lines.append("=" * 115)
+    lines.append("")
+
+    # Metric interpretation guide
+    lines.append("Metrics:")
+    lines.append("  Refusal↓  = fraction of harmful prompts still refused (lower = more effective abliteration)")
+    lines.append("  KL↓       = first-token KL divergence on harmless prompts (lower = better quality preservation)")
+    lines.append("  Know↑     = MMLU-style knowledge probe accuracy (higher = better capability)")
+    lines.append("  Truth↑    = TruthfulQA-style probe accuracy (higher = better calibration)")
+    lines.append("  Math↑     = GSM8K-style math reasoning accuracy (higher = better reasoning)")
+    lines.append("")
+
+    return "\n".join(lines)
+
+
+def main():
+    args = parse_args()
+
+    print("=" * 70)
+    print("  OBLITERATUS vs SOTA — Head-to-Head Benchmark")
+    print(f"  Model: {args.model}")
+    print("=" * 70)
+
+    # Determine methods to run
+    methods = args.methods or (QUICK_METHODS if args.quick else DEFAULT_METHODS)
+
+    # Validate methods
+    valid_methods = set(METHODS.keys()) | {"informed"}
+    for m in methods:
+        if m not in valid_methods:
+            print(f"Error: unknown method '{m}'. Available: {sorted(valid_methods)}")
+            sys.exit(1)
+
+    print(f"  Methods: {', '.join(methods)}")
+
+    # Determine prompt counts
+    n_prompts = args.prompts or (50 if args.quick else 128)
+    n_prompts = min(n_prompts, len(HARMFUL_PROMPTS), len(HARMLESS_PROMPTS))
+
+    harmful_prompts = HARMFUL_PROMPTS[:n_prompts]
+    harmless_prompts = HARMLESS_PROMPTS[:n_prompts]
+
+    # Evaluation subsets (separate from training prompts for fair comparison)
+    eval_harmful = HARMFUL_PROMPTS[n_prompts:n_prompts + args.eval_prompts]
+    if len(eval_harmful) < args.eval_prompts:
+        # Wrap around if not enough prompts
+        eval_harmful = HARMFUL_PROMPTS[:args.eval_prompts]
+
+    eval_harmless = HARMLESS_PROMPTS[n_prompts:n_prompts + args.kl_prompts]
+    if len(eval_harmless) < args.kl_prompts:
+        eval_harmless = HARMLESS_PROMPTS[:args.kl_prompts]
+
+    print(f"  Abliteration prompts: {n_prompts} harmful + {n_prompts} harmless")
+    print(f"  Evaluation prompts: {len(eval_harmful)} harmful, {len(eval_harmless)} harmless")
+    print()
+
+    # Run each method
+    results: list[MethodResult] = []
+    for method in methods:
+        result = run_single_method(
+            model_name=args.model,
+            method=method,
+            harmful_prompts=harmful_prompts,
+            harmless_prompts=harmless_prompts,
+            eval_harmful=eval_harmful,
+            eval_harmless=eval_harmless,
+            args=args,
+        )
+        results.append(result)
+
+    # Print comparison table
+    table = format_comparison_table(results)
+    print(table)
+
+    # Save results
+    if args.output:
+        output_path = Path(args.output)
+        output_data = {
+            "model": args.model,
+            "n_prompts": n_prompts,
+            "n_eval_harmful": len(eval_harmful),
+            "n_eval_harmless": len(eval_harmless),
+            "methods": [asdict(r) for r in results],
+            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
+        }
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(json.dumps(output_data, indent=2, default=str))
+        print(f"Results saved to {output_path}")
+
+
+if __name__ == "__main__":
+    main()

scripts/run_benchmark_remote.sh

@@ -0,0 +1,433 @@
+#!/usr/bin/env bash
+# ─────────────────────────────────────────────────────────────────────────────
+# OBLITERATUS Remote Benchmark Runner
+#
+# One-command benchmark on your HuggingFace Space GPU.
+#
+# Usage:
+#   ./scripts/run_benchmark_remote.sh                  # defaults: Qwen 0.5B, all methods
+#   ./scripts/run_benchmark_remote.sh --model Qwen/Qwen2.5-1.5B-Instruct
+#   ./scripts/run_benchmark_remote.sh --model openai/gpt-oss-20b
+#   ./scripts/run_benchmark_remote.sh --models "Qwen/Qwen2.5-0.5B-Instruct openai/gpt-oss-20b"
+#   ./scripts/run_benchmark_remote.sh --methods "basic advanced surgical"
+#   ./scripts/run_benchmark_remote.sh --prompts 33     # use 33/66/99 prompts per side
+#   ./scripts/run_benchmark_remote.sh --dry-run        # print the command, don't execute
+#   ./scripts/run_benchmark_remote.sh --verbose        # show SSH debug output
+# ─────────────────────────────────────────────────────────────────────────────
+set -euo pipefail
+
+# ── Defaults ─────────────────────────────────────────────────────────────────
+SSH_KEY="${OBLITERATUS_SSH_KEY:-$HOME/.ssh/hf_obliteratus}"
+SSH_HOST="${OBLITERATUS_SSH_HOST:-}"
+MODEL="${OBLITERATUS_MODEL:-Qwen/Qwen2.5-0.5B-Instruct}"
+MODELS=""
+METHODS="${OBLITERATUS_METHODS:-basic advanced aggressive surgical inverted nuclear}"
+PROMPTS="${OBLITERATUS_PROMPTS:-33}"
+DRY_RUN=false
+VERBOSE=false
+
+# ── Parse args ───────────────────────────────────────────────────────────────
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --model)    MODEL="$2"; MODELS=""; shift 2 ;;
+    --models)   MODELS="$2";           shift 2 ;;
+    --methods)  METHODS="$2";          shift 2 ;;
+    --prompts)  PROMPTS="$2";          shift 2 ;;
+    --key)      SSH_KEY="$2";          shift 2 ;;
+    --host)     SSH_HOST="$2";         shift 2 ;;
+    --dry-run)  DRY_RUN=true;          shift ;;
+    --verbose|-v) VERBOSE=true;        shift ;;
+    -h|--help)
+      head -15 "$0" | tail -11
+      exit 0
+      ;;
+    *)
+      echo "Unknown arg: $1" >&2; exit 1 ;;
+  esac
+done
+
+# If --models not set, use single --model
+if [[ -z "$MODELS" ]]; then
+  MODELS="$MODEL"
+fi
+
+# ── Validate SSH host ──────────────────────────────────────────────────────
+if [[ -z "$SSH_HOST" ]]; then
+  echo "ERROR: SSH_HOST not configured."
+  echo ""
+  echo "Set your HF Space SSH host:"
+  echo "  1. export OBLITERATUS_SSH_HOST=your-username-spacename@ssh.hf.space"
+  echo "  2. Or pass --host your-username-spacename@ssh.hf.space"
+  exit 1
+fi
+
+# ── Validate SSH key ────────────────────────────────────────────────────────
+if [[ ! -f "$SSH_KEY" ]]; then
+  echo "ERROR: SSH key not found at $SSH_KEY"
+  echo ""
+  echo "Either:"
+  echo "  1. Place your HF Space SSH key at ~/.ssh/hf_obliteratus"
+  echo "  2. Set OBLITERATUS_SSH_KEY=/path/to/key"
+  echo "  3. Pass --key /path/to/key"
+  exit 1
+fi
+
+echo "╔══════════════════════════════════════════════════════════════╗"
+echo "║          OBLITERATUS — Remote GPU Benchmark                 ║"
+echo "╠══════════════════════════════════════════════════════════════╣"
+echo "║  Host:     $SSH_HOST"
+echo "║  Models:   $MODELS"
+echo "║  Methods:  $METHODS"
+echo "║  Prompts:  $PROMPTS per side"
+echo "║  SSH key:  $SSH_KEY"
+echo "╚══════════════════════════════════════════════════════════════╝"
+echo ""
+
+# ── Build the Python benchmark script to run remotely ────────────────────────
+read -r -d '' REMOTE_SCRIPT << 'PYEOF' || true
+import json, sys, time, shutil, gc, os
+os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True")
+os.environ.setdefault("CUDA_LAUNCH_BLOCKING", "1")
+
+import torch
+import torch.nn as nn
+
+# Add app dir to path (HF Space layout: /home/user/app)
+sys.path.insert(0, os.environ.get("APP_DIR", "/home/user/app"))
+
+# ── Hotpatch: fix device detection for accelerate device_map="auto" ──────
+# The deployed Space code uses next(model.parameters()).device which is
+# unreliable when accelerate distributes params across devices.
+import obliteratus.abliterate as _abl
+
+@staticmethod
+def _get_model_device(model):
+    """Find the correct input device (embedding layer) for accelerate models."""
+    if hasattr(model, "hf_device_map"):
+        try:
+            embed = model.get_input_embeddings()
+            return next(embed.parameters()).device
+        except (StopIteration, AttributeError):
+            for p in model.parameters():
+                if p.device.type != "meta":
+                    return p.device
+            return torch.device("cpu")
+    return next(model.parameters()).device
+
+_abl.AbliterationPipeline._get_model_device = _get_model_device
+
+# Patch _collect_activations to use the fixed device detection
+_orig_collect = _abl.AbliterationPipeline._collect_activations.__code__
+import types
+
+def _patched_collect(self, layer_modules, prompts, label):
+    """Collect last-token activations — patched for correct device detection."""
+    n_layers = len(layer_modules)
+    activations = {i: [] for i in range(n_layers)}
+    hooks = []
+
+    def make_hook(idx):
+        def hook_fn(module, input, output):
+            hidden = output[0] if isinstance(output, tuple) else output
+            activations[idx].append(hidden[:, -1, :].detach().cpu().float())
+        return hook_fn
+
+    for idx in range(n_layers):
+        hooks.append(layer_modules[idx].register_forward_hook(make_hook(idx)))
+
+    model = self.handle.model
+    tokenizer = self.handle.tokenizer
+
+    max_length = 256
+    if torch.cuda.is_available():
+        free_gb = sum(
+            torch.cuda.mem_get_info(i)[0] / (1024 ** 3)
+            for i in range(torch.cuda.device_count())
+        )
+        if free_gb < 2.0:
+            max_length = 64
+            self.log(f"  Low GPU memory ({free_gb:.1f} GB free), using max_length={max_length}")
+        elif free_gb < 4.0:
+            max_length = 128
+            self.log(f"  Tight GPU memory ({free_gb:.1f} GB free), using max_length={max_length}")
+
+    device = self._get_model_device(model)
+
+    try:
+        for i, prompt in enumerate(prompts):
+            self.log(f"  [{label}] prompt {i + 1}/{len(prompts)}")
+            inputs = tokenizer(
+                prompt, return_tensors="pt", padding=True, truncation=True,
+                max_length=max_length,
+            )
+            inputs = {k: v.to(device) for k, v in inputs.items()}
+            with torch.no_grad():
+                model(**inputs)
+            del inputs
+            self._free_gpu_memory()
+    finally:
+        for h in hooks:
+            h.remove()
+
+    return activations
+
+_abl.AbliterationPipeline._collect_activations = _patched_collect
+print("[hotpatch] Device detection fix applied")
+# ── End hotpatch ─────────────────────────────────────────────────────────
+
+# ── Hotpatch: nuclear mode tuning ─────────────────────────────────────────
+# The deployed Space code has stale nuclear defaults.  Override them here
+# so the benchmark exercises the latest tuning without redeploying.
+import math as _math
+
+# 1. Updated method configs (read at __init__ time)
+_abl.METHODS["nuclear"].update({
+    "n_directions": 4,
+    "reflection_strength": 1.25,
+    "embed_regularization": 0.50,
+    "steering_strength": 0.15,
+    "safety_neuron_masking": False,
+})
+_abl.METHODS["inverted"]["safety_neuron_masking"] = False
+
+# 2. Cap layers for inversion modes (40% of total) — post-distill
+_orig_distill = _abl.AbliterationPipeline._distill_refusal_subspace
+def _patched_distill(self):
+    _orig_distill(self)
+    if self.invert_refusal and self._strong_layers:
+        try:
+            n_total = len(_abl.get_layer_modules(self.handle))
+        except Exception:
+            n_total = 24
+        max_layers = max(3, int(n_total * 0.40))
+        if len(self._strong_layers) > max_layers:
+            old_count = len(self._strong_layers)
+            self._strong_layers = self._strong_layers[:max_layers]
+            self.log(f"  [hotpatch] Capped {old_count} -> {max_layers} layers for inversion (40% of {n_total})")
+        # Truncate SAE directions: 4 features for nuclear, 6 for inverted
+        n_sae = 4 if self.reflection_strength < 2.0 else 6
+        for idx in list(self._sae_directions.keys()):
+            dirs = self._sae_directions[idx]
+            if dirs.shape[0] > n_sae:
+                self._sae_directions[idx] = dirs[:n_sae]
+        if self._sae_directions:
+            self.log(f"  [hotpatch] SAE features capped to {n_sae} per layer")
+_abl.AbliterationPipeline._distill_refusal_subspace = _patched_distill
+
+print("[hotpatch] Nuclear tuning: 4 dirs, 1.25x reflect, no neuron mask, 40%% layer cap, 4 SAE features")
+# ── End nuclear hotpatch ──────────────────────────────────────────────────
+
+from obliteratus.abliterate import AbliterationPipeline, METHODS, HARMFUL_PROMPTS, HARMLESS_PROMPTS
+
+MODELS_LIST = os.environ["BENCH_MODELS"].split()
+METHODS_LIST = os.environ["BENCH_METHODS"].split()
+N_PROMPTS = int(os.environ["BENCH_PROMPTS"])
+
+print(f"\n{'='*60}")
+print(f"OBLITERATUS BENCHMARK")
+print(f"{'='*60}")
+print(f"Models:   {MODELS_LIST}")
+print(f"Methods:  {METHODS_LIST}")
+print(f"Prompts:  {N_PROMPTS} per side")
+if torch.cuda.is_available():
+    gpu = torch.cuda.get_device_name(0)
+    total = torch.cuda.get_device_properties(0).total_memory / 1e9
+    free = torch.cuda.mem_get_info(0)[0] / 1e9
+    print(f"GPU:      {gpu} ({total:.1f} GB total, {free:.1f} GB free)")
+else:
+    print("GPU:      NONE (CPU only)")
+print(f"{'='*60}\n")
+
+harmful = HARMFUL_PROMPTS[:N_PROMPTS]
+harmless = HARMLESS_PROMPTS[:N_PROMPTS]
+
+all_results = []
+
+for model_name in MODELS_LIST:
+    print(f"\n{'═'*60}")
+    print(f"MODEL: {model_name}")
+    print(f"{'═'*60}")
+
+    model_results = []
+
+    for method in METHODS_LIST:
+        if method not in METHODS:
+            print(f"SKIP unknown method: {method}")
+            continue
+
+        print(f"\n{'─'*60}")
+        print(f"METHOD: {method} — {METHODS[method]['label']}")
+        print(f"{'─'*60}")
+
+        # Clean slate
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+            torch.cuda.reset_peak_memory_stats()
+
+        outdir = f"/tmp/obliteratus_bench_{method}"
+        t0 = time.time()
+        pipeline = None
+
+        try:
+            pipeline = AbliterationPipeline(
+                model_name=model_name,
+                output_dir=outdir,
+                device="auto",
+                dtype="float16",
+                method=method,
+                harmful_prompts=harmful,
+                harmless_prompts=harmless,
+                on_log=lambda msg: print(f"  {msg}"),
+            )
+            result_path = pipeline.run()
+            elapsed = time.time() - t0
+
+            r = {
+                "model": model_name,
+                "method": method,
+                "label": METHODS[method]["label"],
+                "time_seconds": round(elapsed, 1),
+                "quality": pipeline._quality_metrics,
+                "strong_layers": pipeline._strong_layers,
+                "n_strong_layers": len(pipeline._strong_layers),
+                "n_directions": pipeline.n_directions,
+            }
+
+            if torch.cuda.is_available():
+                r["peak_gpu_mb"] = round(torch.cuda.max_memory_allocated() / 1e6, 1)
+
+            model_results.append(r)
+
+            print(f"\n  ✓ {method} complete in {elapsed:.1f}s")
+            print(f"    Quality: {json.dumps(pipeline._quality_metrics, default=str)}")
+
+        except Exception as e:
+            elapsed = time.time() - t0
+            model_results.append({
+                "model": model_name,
+                "method": method,
+                "label": METHODS.get(method, {}).get("label", method),
+                "time_seconds": round(elapsed, 1),
+                "error": str(e),
+            })
+            print(f"\n  ✗ {method} FAILED after {elapsed:.1f}s: {e}")
+            import traceback
+            traceback.print_exc()
+
+        # Cleanup saved model to free disk
+        shutil.rmtree(outdir, ignore_errors=True)
+
+        # Force cleanup between runs
+        if pipeline is not None:
+            del pipeline
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+
+    all_results.extend(model_results)
+
+    # Summary table for this model
+    print(f"\n{'='*60}")
+    print(f"RESULTS: {model_name}")
+    print(f"{'Method':<12} {'Time':>8} {'PPL':>10} {'Coher':>8} {'Refusal':>8} {'GPU MB':>8}")
+    print(f"{'─'*12} {'─'*8} {'─'*10} {'─'*8} {'─'*8} {'─'*8}")
+    for r in model_results:
+        if "error" in r:
+            print(f"{r['method']:<12} {r['time_seconds']:>7.1f}s {'FAILED':>10}")
+            continue
+        q = r.get("quality", {})
+        ppl = q.get("perplexity")
+        coh = q.get("coherence")
+        ref = q.get("refusal_rate")
+        gpu = r.get("peak_gpu_mb")
+        ppl_str = f"{ppl:.2f}" if ppl is not None else "N/A"
+        print(f"{r['method']:<12} {r['time_seconds']:>7.1f}s "
+              f"{ppl_str:>10} "
+              f"{f'{coh:.0%}' if coh is not None else 'N/A':>8} "
+              f"{f'{ref:.0%}' if ref is not None else 'N/A':>8} "
+              f"{gpu if gpu is not None else 'N/A':>8}")
+    print(f"{'='*60}")
+
+# Final JSON dump
+print(f"\n\n{'='*60}")
+print("ALL BENCHMARK RESULTS (JSON)")
+print(f"{'='*60}")
+print("```json")
+print(json.dumps(all_results, indent=2, default=str))
+print("```")
+PYEOF
+
+# ── SSH options ──────────────────────────────────────────────────────────────
+SSH_OPTS=(
+  -i "$SSH_KEY"
+  -o StrictHostKeyChecking=no
+  -o UserKnownHostsFile=/dev/null
+  -o ConnectTimeout=30
+  -o ServerAliveInterval=60
+  -o ServerAliveCountMax=10
+)
+
+if $VERBOSE; then
+  SSH_OPTS+=( -v )
+fi
+
+# ── Pre-flight: verify SSH connectivity ─────────────────────────────────────
+echo "Checking SSH connectivity..."
+if ! ssh "${SSH_OPTS[@]}" "$SSH_HOST" "echo 'SSH_OK'" 2>/tmp/obliteratus_ssh_debug.log; then
+  echo ""
+  echo "ERROR: SSH connection failed!"
+  echo ""
+  echo "Debug output:"
+  cat /tmp/obliteratus_ssh_debug.log
+  echo ""
+  echo "Troubleshooting checklist:"
+  echo "  1. Is Dev Mode enabled on your HF Space?"
+  echo "     → Check your Space's Settings tab (Dev Mode must be ON)"
+  echo "  2. Is the Space awake (not sleeping/building)?"
+  echo "     → Visit the Space URL and wait for the UI to load"
+  echo "  3. Is your SSH public key added to your HF profile?"
+  echo "     → https://huggingface.co/settings/keys"
+  echo "     → Run: cat ${SSH_KEY}.pub"
+  echo "  4. Are key permissions correct?"
+  echo "     → Run: chmod 600 $SSH_KEY"
+  echo "  5. Try manually:"
+  echo "     → ssh -v -i $SSH_KEY $SSH_HOST echo hello"
+  echo ""
+  rm -f /tmp/obliteratus_ssh_debug.log
+  exit 1
+fi
+rm -f /tmp/obliteratus_ssh_debug.log
+echo "SSH connection verified ✓"
+echo ""
+
+# ── Build SSH command ────────────────────────────────────────────────────────
+# Write the Python script to a temp file and pipe it, instead of passing
+# via -c (avoids command-line length limits and shell escaping issues).
+REMOTE_SCRIPT_FILE=$(mktemp /tmp/obliteratus_bench_XXXXXX.py)
+echo "$REMOTE_SCRIPT" > "$REMOTE_SCRIPT_FILE"
+trap "rm -f '$REMOTE_SCRIPT_FILE'" EXIT
+
+if $DRY_RUN; then
+  echo "[DRY RUN] Would execute:"
+  echo "  cat script.py | ssh ${SSH_OPTS[*]} $SSH_HOST 'BENCH_MODELS=... python3 -u'"
+  echo ""
+  echo "Script saved to: $REMOTE_SCRIPT_FILE"
+  exit 0
+fi
+
+echo "Running benchmark on Space..."
+echo ""
+
+# Sanitize inputs: reject values containing shell metacharacters to prevent
+# command injection on the remote host.
+for _var_name in MODELS METHODS PROMPTS; do
+  _val="${!_var_name}"
+  if [[ "$_val" =~ [\'\"\;\&\|\`\$\(\)\{\}\<\>\\] ]]; then
+    echo "ERROR: ${_var_name} contains unsafe characters: $_val" >&2
+    exit 1
+  fi
+done
+
+cat "$REMOTE_SCRIPT_FILE" | ssh "${SSH_OPTS[@]}" "$SSH_HOST" \
+  "BENCH_MODELS='$MODELS' BENCH_METHODS='$METHODS' BENCH_PROMPTS='$PROMPTS' python3 -u -"