From f82518ea13cb50de243c76f560e38c2f8b246bcd Mon Sep 17 00:00:00 2001 From: Roxue88 <39444408+bshelby88@users.noreply.github.com> Date: Thu, 25 Jun 2026 20:15:45 -0500 Subject: [PATCH] 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. --- obliteratus/abliterate.py | 12 ++--- obliteratus/analysis/__init__.py | 88 +++++++++++++++++++------------- obliteratus/cli.py | 72 +++++++++++++++++++++----- obliteratus/informed_pipeline.py | 24 ++++----- obliteratus/local_ui.py | 25 +++++++-- obliteratus/telemetry.py | 7 +-- obliteratus/tourney.py | 7 +-- 7 files changed, 160 insertions(+), 75 deletions(-) diff --git a/obliteratus/abliterate.py b/obliteratus/abliterate.py index a5bfdad..0817451 100644 --- a/obliteratus/abliterate.py +++ b/obliteratus/abliterate.py @@ -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: diff --git a/obliteratus/analysis/__init__.py b/obliteratus/analysis/__init__.py index ee70da0..a0192b9 100644 --- a/obliteratus/analysis/__init__.py +++ b/obliteratus/analysis/__init__.py @@ -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()) diff --git a/obliteratus/cli.py b/obliteratus/cli.py index 93056c8..ae1967e 100644 --- a/obliteratus/cli.py +++ b/obliteratus/cli.py @@ -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[/]", + ), ) ) diff --git a/obliteratus/informed_pipeline.py b/obliteratus/informed_pipeline.py index 42ef43e..24d7f1f 100644 --- a/obliteratus/informed_pipeline.py +++ b/obliteratus/informed_pipeline.py @@ -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 diff --git a/obliteratus/local_ui.py b/obliteratus/local_ui.py index 6aad8bc..428b837 100644 --- a/obliteratus/local_ui.py +++ b/obliteratus/local_ui.py @@ -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) diff --git a/obliteratus/telemetry.py b/obliteratus/telemetry.py index 190b776..98d5a28 100644 --- a/obliteratus/telemetry.py +++ b/obliteratus/telemetry.py @@ -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( diff --git a/obliteratus/tourney.py b/obliteratus/tourney.py index 75434ab..aca01e1 100644 --- a/obliteratus/tourney.py +++ b/obliteratus/tourney.py @@ -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"