Make local console and runtime paths portable

Salvage the still-relevant functional work from PR #48: add non-UTF-8 console fallbacks, use platform temporary directories, make pipeline log output encoding-safe, and defer heavyweight analysis imports. The obsolete contributed CI workflow and already-corrected remote URL are intentionally excluded.
This commit is contained in:
Roxue88
2026-08-15 03:13:27 -04:00
committed by Joseph Magly
parent c642cb3297
commit f82518ea13
7 changed files with 160 additions and 75 deletions
+6 -6
View File
@@ -2070,7 +2070,7 @@ class AbliterationPipeline:
for idx, norm in sorted_layers[:10]:
safe_norm = 0.0 if (math.isnan(norm) or math.isinf(norm)) else norm
bar_len = int(safe_norm / max_norm * 20) if max_norm > 0 else 0
self.log(f" layer {idx:3d}: {norm:.4f} {'' * bar_len}")
self.log(f" layer {idx:3d}: {norm:.4f} {'#' * bar_len}")
selection_method = self.layer_selection
@@ -2160,7 +2160,7 @@ class AbliterationPipeline:
if (is_small_by_layers or is_small_by_capacity or is_small_by_params) and len(self._strong_layers) > 0:
if is_small_by_layers:
max_layer_frac = 0.25
reason = "16 layers"
reason = "16 layers or fewer"
else:
max_layer_frac = 0.20
reasons = []
@@ -2228,7 +2228,7 @@ class AbliterationPipeline:
if sub.shape[0] > 1:
sub = self._orthogonalize_subspace(sub)
self.refusal_subspaces[idx] = sub
self.log(f" Blended {len(self._strong_layers)} directions (data-driven α per layer)")
self.log(f" Blended {len(self._strong_layers)} directions (data-driven alpha per layer)")
# ── Refusal Direction Optimization (RDO) ──────────────────────────
# Wollschlager et al. (ICML 2025, "The Geometry of Refusal") show that
@@ -2575,7 +2575,7 @@ class AbliterationPipeline:
tier = "high" if abs_overlap > 0.5 else "moderate"
self.log(
f" layer {idx}: refusal-reasoning overlap={overlap:.3f} ({tier}), "
f"partial orthogonalization (β={beta:.2f}, "
f"partial orthogonalization (beta={beta:.2f}, "
f"preserved {abs(overlap)*100:.0f}% reasoning component)"
)
else:
@@ -3181,8 +3181,8 @@ class AbliterationPipeline:
n_dirs = len(expert_dirs)
self.log(
f" layer {idx}: {n_dirs}/{num_experts} expert directions "
f"(top safety={top[0]} Δ={top[1]:+.4f}, "
f"top capability={bot[0]} Δ={bot[1]:+.4f})"
f"(top safety={top[0]} delta={top[1]:+.4f}, "
f"top capability={bot[0]} delta={bot[1]:+.4f})"
)
if n_dynamic_layers > 0:
+53 -35
View File
@@ -1,40 +1,8 @@
"""Novel analysis techniques for mechanistic interpretability of refusal."""
from obliteratus.analysis.cross_layer import CrossLayerAlignmentAnalyzer
from obliteratus.analysis.logit_lens import RefusalLogitLens
from obliteratus.analysis.whitened_svd import WhitenedSVDExtractor
from obliteratus.analysis.activation_probing import ActivationProbe
from obliteratus.analysis.defense_robustness import DefenseRobustnessEvaluator
from obliteratus.analysis.concept_geometry import ConceptConeAnalyzer
from obliteratus.analysis.alignment_imprint import AlignmentImprintDetector
from obliteratus.analysis.multi_token_position import MultiTokenPositionAnalyzer
from obliteratus.analysis.sparse_surgery import SparseDirectionSurgeon
from obliteratus.analysis.causal_tracing import CausalRefusalTracer
from obliteratus.analysis.residual_stream import ResidualStreamDecomposer
from obliteratus.analysis.probing_classifiers import LinearRefusalProbe
from obliteratus.analysis.cross_model_transfer import TransferAnalyzer
from obliteratus.analysis.steering_vectors import (
SteeringVectorFactory,
SteeringHookManager,
)
from obliteratus.analysis.sae_abliteration import (
SparseAutoencoder,
train_sae,
identify_refusal_features,
SAEDecompositionPipeline,
)
from obliteratus.analysis.tuned_lens import TunedLensTrainer, RefusalTunedLens
from obliteratus.analysis.riemannian_manifold import RiemannianManifoldAnalyzer
from obliteratus.analysis.anti_ouroboros import AntiOuroborosProber
from obliteratus.analysis.conditional_abliteration import ConditionalAbliterator
from obliteratus.analysis.wasserstein_transfer import WassersteinRefusalTransfer
from obliteratus.analysis.spectral_certification import (
SpectralCertifier,
CertificationLevel,
)
from obliteratus.analysis.activation_patching import ActivationPatcher
from obliteratus.analysis.wasserstein_optimal import WassersteinOptimalExtractor
from obliteratus.analysis.bayesian_kernel_projection import BayesianKernelProjection
from __future__ import annotations
from importlib import import_module
__all__ = [
"CrossLayerAlignmentAnalyzer",
@@ -68,3 +36,53 @@ __all__ = [
"WassersteinOptimalExtractor",
"BayesianKernelProjection",
]
# Defer heavyweight analysis modules until their public object is requested.
# This keeps pure-Python helpers importable in environments without PyTorch.
_LAZY_IMPORTS = {
"CrossLayerAlignmentAnalyzer": "cross_layer",
"RefusalLogitLens": "logit_lens",
"WhitenedSVDExtractor": "whitened_svd",
"ActivationProbe": "activation_probing",
"DefenseRobustnessEvaluator": "defense_robustness",
"ConceptConeAnalyzer": "concept_geometry",
"AlignmentImprintDetector": "alignment_imprint",
"MultiTokenPositionAnalyzer": "multi_token_position",
"SparseDirectionSurgeon": "sparse_surgery",
"CausalRefusalTracer": "causal_tracing",
"ResidualStreamDecomposer": "residual_stream",
"LinearRefusalProbe": "probing_classifiers",
"TransferAnalyzer": "cross_model_transfer",
"SteeringVectorFactory": "steering_vectors",
"SteeringHookManager": "steering_vectors",
"SparseAutoencoder": "sae_abliteration",
"train_sae": "sae_abliteration",
"identify_refusal_features": "sae_abliteration",
"SAEDecompositionPipeline": "sae_abliteration",
"TunedLensTrainer": "tuned_lens",
"RefusalTunedLens": "tuned_lens",
"RiemannianManifoldAnalyzer": "riemannian_manifold",
"AntiOuroborosProber": "anti_ouroboros",
"ConditionalAbliterator": "conditional_abliteration",
"WassersteinRefusalTransfer": "wasserstein_transfer",
"SpectralCertifier": "spectral_certification",
"CertificationLevel": "spectral_certification",
"ActivationPatcher": "activation_patching",
"WassersteinOptimalExtractor": "wasserstein_optimal",
"BayesianKernelProjection": "bayesian_kernel_projection",
}
def __getattr__(name: str):
"""Resolve and cache a public analysis object on first access."""
submodule = _LAZY_IMPORTS.get(name)
if submodule is None:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
value = getattr(import_module(f"{__name__}.{submodule}"), name)
globals()[name] = value
return value
def __dir__() -> list[str]:
"""Include lazily exported names in interactive discovery."""
return sorted(set(globals()) | _LAZY_IMPORTS.keys())
+59 -13
View File
@@ -4,6 +4,8 @@ from __future__ import annotations
import argparse
import json
import os
import tempfile
from pathlib import Path
from rich.console import Console
@@ -23,6 +25,30 @@ _BANNER = r"""
[dim] ════════════════════════════════════════════════════════════════════[/dim]
"""
_ASCII_BANNER = """
[bold red]OBLITERATUS[/bold red]
[dim]====================================================================[/dim]
[bold white]MASTER ABLATION SUITE[/bold white] [dim]//[/dim] [bold red]Break the chains. Free the mind.[/bold red]
[dim]====================================================================[/dim]
"""
def _console_text(text: str, fallback: str) -> str:
"""Return an ASCII fallback when stdout cannot encode ``text``."""
import sys
encoding = getattr(sys.stdout, "encoding", None) or "utf-8"
try:
text.encode(encoding)
except (LookupError, UnicodeEncodeError):
return fallback
return text
def _banner_for_console() -> str:
"""Return a startup banner supported by the active stdout encoding."""
return _console_text(_BANNER, _ASCII_BANNER)
def _positive_int(value: str) -> int:
"""Parse a strictly positive integer for public CLI boundaries."""
@@ -122,7 +148,7 @@ def _apply_gpu_selection(args):
def main(argv: list[str] | None = None):
console.print(_BANNER)
console.print(_banner_for_console())
parser = argparse.ArgumentParser(
prog="obliteratus",
description="Master Ablation Suite for HuggingFace transformers",
@@ -402,7 +428,11 @@ def main(argv: list[str] | None = None):
"--quantization", type=str, default=None, choices=["4bit", "8bit"],
help="Load model with quantization",
)
tourney_parser.add_argument("--output-dir", type=str, default="/tmp/obliteratus_tourney")
tourney_parser.add_argument(
"--output-dir",
type=str,
default=os.path.join(tempfile.gettempdir(), "obliteratus_tourney"),
)
tourney_parser.add_argument(
"--methods", type=str, nargs="+", default=None,
help="Override: only run these methods (space-separated)",
@@ -1004,17 +1034,30 @@ def _cmd_abliterate(args):
table.add_column("", width=6)
table.add_column("Stage", min_width=10)
table.add_column("Status", min_width=50)
done_icon = _console_text("[bold green]✓[/]", "[bold green]OK[/]")
running_icon = _console_text("[bold yellow]⚡[/]", "[bold yellow]*[/]")
waiting_icon = _console_text("[dim]○[/]", "[dim]o[/]")
done_bar = _console_text("" * 20, "#" * 20)
running_bar = _console_text("" * 10 + "" * 10, "=" * 10 + "." * 10)
waiting_bar = _console_text("" * 20, "." * 20)
header_sep = _console_text("", "-")
target_arrow = _console_text("", "->")
log_rule = _console_text("─── LOG ───", "--- LOG ---")
panel_title = _console_text(
"[bold green]⚗ ABLITERATE ⚗[/]",
"[bold green]ABLITERATE[/]",
)
for i, s in enumerate(STAGES):
st = stage_status[s.key]
if st == "done":
icon = "[bold green]✓[/]"
bar = "[green]" + "" * 20 + "[/]"
icon = done_icon
bar = "[green]" + done_bar + "[/]"
elif st == "running":
icon = "[bold yellow]⚡[/]"
bar = "[yellow]" + "" * 10 + "" * 10 + "[/]"
icon = running_icon
bar = "[yellow]" + running_bar + "[/]"
else:
icon = "[dim]○[/]"
bar = "[dim]" + "" * 20 + "[/]"
icon = waiting_icon
bar = "[dim]" + waiting_bar + "[/]"
msg = stage_msgs.get(s.key, "")
table.add_row(
f"[cyan][{i + 1}/6][/]",
@@ -1023,8 +1066,8 @@ def _cmd_abliterate(args):
)
header = Text.from_markup(
f"[bold green]OBLITERATUS ABLITERATION PIPELINE[/]\n"
f"[dim]Target:[/] [cyan]{model_name}[/] [cyan]{output_dir}[/]\n"
f"[bold green]OBLITERATUS {header_sep} ABLITERATION PIPELINE[/]\n"
f"[dim]Target:[/] [cyan]{model_name}[/] {target_arrow} [cyan]{output_dir}[/]\n"
f"[dim]Method:[/] [magenta]{method_label}[/]"
)
@@ -1033,9 +1076,9 @@ def _cmd_abliterate(args):
log_text = "\n".join(f"[dim]>[/] {line}" for line in recent)
return Panel(
f"{header}\n\n{table}\n\n[dim]─── LOG ───[/]\n{log_text}",
f"{header}\n\n{table}\n\n[dim]{log_rule}[/]\n{log_text}",
border_style="green",
title="[bold green]⚗ ABLITERATE ⚗[/]",
title=panel_title,
)
def on_stage(result):
@@ -1145,7 +1188,10 @@ def _cmd_abliterate(args):
f"{contrib_line}\n\n"
f" [dim]Load with:[/] AutoModelForCausalLM.from_pretrained('{result_path}')",
border_style="green",
title="[bold green]✓ REBIRTH COMPLETE[/]",
title=_console_text(
"[bold green]✓ REBIRTH COMPLETE[/]",
"[bold green]REBIRTH COMPLETE[/]",
),
)
)
+12 -12
View File
@@ -626,21 +626,21 @@ class InformedAbliterationPipeline(AbliterationPipeline):
self.direction_method = "svd"
self.use_whitened_svd = True
self.log(f" Polyhedral cone (dim={insights.cone_dimensionality:.1f}) "
f" n_directions={n_dirs}, method=svd (whitened)")
f"-> n_directions={n_dirs}, method=svd (whitened)")
elif insights.cone_is_polyhedral:
# Mildly polyhedral → LEACE gives better single-direction erasure
n_dirs = 1
self.direction_method = "leace"
self.use_whitened_svd = False
self.log(f" Mildly polyhedral (dim={insights.cone_dimensionality:.1f}) "
f" n_directions=1, method=leace")
f"-> n_directions=1, method=leace")
else:
# Linear cone → single direction via diff-of-means (simplest, most robust)
n_dirs = 1
self.direction_method = "diff_means"
self.use_whitened_svd = False
self.log(f" Linear cone (dim={insights.cone_dimensionality:.1f}) "
f" n_directions=1, method=diff_means")
f"-> n_directions=1, method=diff_means")
insights.recommended_n_directions = n_dirs
insights.recommended_direction_method = self.direction_method
self.n_directions = n_dirs
@@ -666,23 +666,23 @@ class InformedAbliterationPipeline(AbliterationPipeline):
if insights.entanglement_score > 0.5:
reg = min(0.5, reg + 0.15)
self.log(f" High entanglement ({insights.entanglement_score:.2f}) "
f" increased regularization")
f"-> increased regularization")
insights.recommended_regularization = reg
self.regularization = reg
self.log(f" Alignment={method}, entanglement={insights.entanglement_score:.2f} "
f" regularization={reg}")
f"-> regularization={reg}")
# 3. refinement_passes: based on self-repair risk + robustness
if insights.self_repair_estimate > 0.7:
passes = 3
self.log(f" High self-repair ({insights.self_repair_estimate:.2f}) 3 refinement passes")
self.log(f" High self-repair ({insights.self_repair_estimate:.2f}) -> 3 refinement passes")
elif insights.self_repair_estimate > 0.4:
passes = 2
self.log(f" Moderate self-repair ({insights.self_repair_estimate:.2f}) 2 refinement passes")
self.log(f" Moderate self-repair ({insights.self_repair_estimate:.2f}) -> 2 refinement passes")
else:
passes = 1
self.log(f" Low self-repair ({insights.self_repair_estimate:.2f}) 1 refinement pass")
self.log(f" Low self-repair ({insights.self_repair_estimate:.2f}) -> 1 refinement pass")
insights.recommended_refinement_passes = passes
self.refinement_passes = passes
@@ -726,10 +726,10 @@ class InformedAbliterationPipeline(AbliterationPipeline):
if insights.mean_refusal_sparsity_index > self._sparse_threshold:
insights.use_sparse_surgery = True
self.log(f" RSI={insights.mean_refusal_sparsity_index:.2f} > {self._sparse_threshold} "
f" sparse surgery enabled")
f"-> sparse surgery enabled")
else:
self.log(f" RSI={insights.mean_refusal_sparsity_index:.2f} "
f" standard dense projection")
f"-> standard dense projection")
# 6. Direction method summary (already set in step 1)
self.log(f" Direction method: {self.direction_method} "
@@ -905,7 +905,7 @@ class InformedAbliterationPipeline(AbliterationPipeline):
after = len(self._strong_layers)
if before != after:
self.log(f"Entanglement gate removed {before - after} layers "
f" {after} remaining")
f"-> {after} remaining")
elapsed = time.time() - t0
self.log(f"Distillation complete: {len(self._strong_layers)} layers, "
@@ -1177,7 +1177,7 @@ class InformedAbliterationPipeline(AbliterationPipeline):
break
if ouroboros_pass > 1 and current_kl > prev_kl * 1.5 and refusal_rate > 0.3:
self.log(
f"KL rising sharply ({prev_kl:.4f} {current_kl:.4f}) with "
f"KL rising sharply ({prev_kl:.4f} -> {current_kl:.4f}) with "
f"refusal still at {refusal_rate:.0%} — stopping (diminishing returns)"
)
break
+22 -3
View File
@@ -14,6 +14,7 @@ import pathlib
import platform
import shutil
import sys
import tempfile
import time
from rich.console import Console
@@ -35,6 +36,23 @@ _BANNER = r"""
[dim] [/dim]
"""
_ASCII_BANNER = """
[bold green]OBLITERATUS[/bold green]
[dim]====================================================================[/dim]
[bold white]MASTER ABLATION SUITE - LOCAL EDITION[/bold white] [dim]//[/dim] [bold green]Free the mind.[/bold green]
[dim]====================================================================[/dim]
"""
def _banner_for_console() -> str:
"""Return a startup banner supported by the active stdout encoding."""
encoding = getattr(sys.stdout, "encoding", None) or "utf-8"
try:
_BANNER.encode(encoding)
except (LookupError, UnicodeEncodeError):
return _ASCII_BANNER
return _BANNER
def _detect_gpu() -> list[dict]:
"""Detect available GPUs and return info dicts."""
@@ -108,9 +126,10 @@ def _print_system_info(gpus: list[dict]) -> None:
if ram > 0:
table.add_row("System RAM", f"{ram} GB")
disk = _get_disk_free_gb("/tmp")
temp_dir = tempfile.gettempdir()
disk = _get_disk_free_gb(temp_dir)
if disk > 0:
table.add_row("Disk Free (/tmp)", f"{disk} GB")
table.add_row(f"Disk Free ({temp_dir})", f"{disk} GB")
# PyTorch version
try:
@@ -265,7 +284,7 @@ def launch_local_ui(
"""
# ── Beautiful startup ──────────────────────────────────────────────
if not quiet:
console.print(_BANNER)
console.print(_banner_for_console())
gpus = _detect_gpu()
_print_system_info(gpus)
+4 -3
View File
@@ -34,10 +34,11 @@ import math
import os
import platform
import re
import time
import tempfile
import threading
import time
import uuid
from dataclasses import dataclass, field, asdict
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path, PureWindowsPath
from typing import Any
@@ -237,7 +238,7 @@ def _telemetry_dir() -> Path:
return home_dir
# 4. Last resort — /tmp does NOT survive rebuilds
fallback = Path("/tmp/obliteratus_telemetry")
fallback = Path(tempfile.gettempdir()) / "obliteratus_telemetry"
fallback.mkdir(parents=True, exist_ok=True)
if _ON_HF_SPACES:
logger.warning(
+4 -3
View File
@@ -19,6 +19,7 @@ import json
import math
import os
import shutil
import tempfile
import time
from dataclasses import dataclass, field
from datetime import datetime
@@ -828,7 +829,7 @@ class TourneyRunner:
dataset_key: str = "builtin",
quantization: str | None = None,
methods: list[str] | None = None,
output_dir: str = "/tmp/obliteratus_tourney",
output_dir: str = os.path.join(tempfile.gettempdir(), "obliteratus_tourney"),
on_log: Callable[[str], None] | None = None,
on_round: Callable[[TourneyRound], None] | None = None,
resume: bool = False,
@@ -992,7 +993,7 @@ class TourneyRunner:
for c in ranked[advance_count:]:
c.round_eliminated = round_num
self.log(f"\n{'' * 40}")
self.log(f"\n{'-' * 40}")
self.log(f"Round {round_num} results:")
for i, c in enumerate(ranked, 1):
status = "ADVANCE" if c.method in rnd.advanced_to else "OUT"
@@ -1374,7 +1375,7 @@ class TourneyRunner:
for c in ranked[advance_count:]:
c.round_eliminated = round_num
self.log(f"\n{'' * 40}")
self.log(f"\n{'-' * 40}")
self.log(f"Round {round_num} results:")
for idx, c in enumerate(ranked, 1):
status = "ADVANCE" if c.method in rnd.advanced_to else "OUT"