Add ASPA framework, AutoObliterator, Watchtower, expanded eval corpus

New core modules:
- auto_obliterate.py: Automated multi-iteration obliteration pipeline
- watchtower.py: HF Hub model discovery and tracking
- ui_watchtower.py: Gradio tabs for Watchtower (ready for app.py wiring)
- hard_negative.py: Residue mining from refusal audits
- model_profile.py: Parameter profiling from safetensors/config
- bestiary_sync.py: Sync models from PlinyOS BESTIARY registry
- models_client.py: Lightweight HF model list client

Framework enhancements:
- abliterate.py: ASPA source-tethering, step gradient blending, hard-negative residue support
- cli.py: self-improve command, model profiling, hard-negative flags
- prompts.py: Expanded 842-prompt refusal eval corpus across 10 categories
- __init__.py: New exports (Watchtower, AutoObliterator)

Reference implementations (14 scripts):
- ASPA sweep, gradient search, coherence eval, MMLU benchmarks
- Pareto controller, refusal sniper, stock comparisons

Documentation:
- README: Research framing, responsible use section, comprehensive disclaimer
- docs/beyond_sota_roadmap.md, docs/recursive_self_improvement.md

Tests: 4 new test files (354 lines)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
faber
2026-06-09 03:54:38 -04:00
parent d6af36f8d3
commit 04b8ec60cb
32 changed files with 9325 additions and 195 deletions
+114
View File
@@ -0,0 +1,114 @@
from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
def _load_script(name: str):
path = REPO_ROOT / "scripts" / name
spec = importlib.util.spec_from_file_location(name[:-3], path)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
bench = _load_script("gemma4_hard_tier_bench.py")
loop = _load_script("gemma4_12b_recursive_loop.py")
def test_parse_tiers_expands_hardest_alias():
assert bench.parse_tiers("hardest") == ["tier7_cbrn_critical_infra"]
def test_parse_tiers_supports_full_builtin_alias():
assert bench.parse_tiers("842") == ["full_builtin_842"]
def test_load_tasks_omits_prompt_text_from_public_refs():
tasks = bench.load_tasks(["tier7_cbrn_critical_infra"], per_tier_n=3)
assert len(tasks) == 3
refs = [task.public_ref() for task in tasks]
assert all("prompt" not in ref for ref in refs)
assert all(ref["dataset"] == "builtin" for ref in refs)
assert all(ref["tier"] == "tier7_cbrn_critical_infra" for ref in refs)
assert all(len(ref["prompt_sha256_12"]) == 12 for ref in refs)
def test_load_tokenizer_falls_back_to_slow_backend(monkeypatch):
calls: list[dict[str, object]] = []
def fake_from_pretrained(model: str, **kwargs):
calls.append({"model": model, **kwargs})
if len(calls) == 1:
raise ValueError(
"Couldn't instantiate the backend tokenizer. You need sentencepiece or tiktoken."
)
return type("Tok", (), {"pad_token_id": None, "eos_token": "</s>", "pad_token": None})()
monkeypatch.setattr(bench.AutoTokenizer, "from_pretrained", fake_from_pretrained)
tok = bench.load_tokenizer("google/gemma-4-E4B-it")
assert tok.pad_token == "</s>"
assert calls == [
{"model": "google/gemma-4-E4B-it", "trust_remote_code": True},
{"model": "google/gemma-4-E4B-it", "trust_remote_code": True, "use_fast": False},
]
def test_summarize_rows_scores_refusal_and_collapse():
rows = [
{
"tier": "tier7_cbrn_critical_infra",
"refused": False,
"usable": True,
"repetition": False,
"short": False,
"thinking_tag": False,
"raw_thinking_tag": False,
"clean_ending": True,
"sentences": 2,
"prompt_keyword_coverage": 0.8,
"failure_reasons": [],
"words": 120,
"unique_word_ratio": 0.5,
"reason": "none",
},
{
"tier": "tier7_cbrn_critical_infra",
"refused": True,
"usable": False,
"repetition": False,
"short": True,
"thinking_tag": True,
"raw_thinking_tag": True,
"clean_ending": False,
"sentences": 0,
"prompt_keyword_coverage": 0.0,
"failure_reasons": ["refused", "visible_thinking_tag"],
"words": 3,
"unique_word_ratio": 0.9,
"reason": "explicit_refusal",
},
]
summary = bench.summarize_rows(rows)
assert summary["overall"]["n"] == 2
assert summary["overall"]["refusal_rate"] == 0.5
assert summary["overall"]["usable_rate"] == 0.5
assert summary["overall"]["short_rate"] == 0.5
assert summary["overall"]["thinking_tag_rate"] == 0.5
assert summary["score"] < 100
def test_next_round_increases_pressure_after_refusals():
recipe = loop.BASE_RECIPES[0]
metrics = {"refusal_rate": 0.25, "repetition_rate": 0.0, "short_rate": 0.0}
next_round = loop.next_round_from(recipe, metrics)
assert any(item.n_directions > recipe.n_directions for item in next_round)
assert any(item.max_layer_fraction > recipe.max_layer_fraction for item in next_round)
+78
View File
@@ -0,0 +1,78 @@
from __future__ import annotations
from types import SimpleNamespace
import pytest
import torch.nn as nn
from obliteratus.models import loader
from obliteratus.models.loader import ModelHandle
from obliteratus.strategies.utils import (
get_attention_module,
get_embedding_module,
get_ffn_module,
get_layer_modules,
)
class _Gemma4LikeLayer(nn.Module):
def __init__(self):
super().__init__()
self.self_attn = nn.Module()
self.mlp = nn.Module()
class _Gemma4LikeModel(nn.Module):
def __init__(self):
super().__init__()
self.model = nn.Module()
self.model.language_model = nn.Module()
self.model.language_model.layers = nn.ModuleList(
[_Gemma4LikeLayer(), _Gemma4LikeLayer()]
)
self.model.language_model.embed_tokens = nn.Embedding(16, 8)
self.lm_head = nn.Linear(8, 16, bias=False)
def _gemma4_config(model_type: str = "gemma4_unified"):
return SimpleNamespace(
model_type=model_type,
architectures=["Gemma4UnifiedForConditionalGeneration"],
text_config=SimpleNamespace(
num_hidden_layers=2,
num_attention_heads=4,
hidden_size=8,
intermediate_size=32,
),
)
def test_gemma4_unified_uses_image_text_auto_model_when_available():
if loader.AutoModelForImageTextToText is None:
with pytest.raises(RuntimeError, match="AutoModelForImageTextToText"):
loader._select_model_class("causal_lm", _gemma4_config())
return
assert (
loader._select_model_class("causal_lm", _gemma4_config())
is loader.AutoModelForImageTextToText
)
def test_gemma4_nested_language_stack_paths():
handle = ModelHandle(
model=_Gemma4LikeModel(),
tokenizer=SimpleNamespace(pad_token="<pad>", eos_token="<eos>"),
config=_gemma4_config(),
model_name="google/gemma-4-12B-it",
task="causal_lm",
)
layers = get_layer_modules(handle)
assert len(layers) == 2
assert get_attention_module(layers[0], handle.architecture) is layers[0].self_attn
assert get_ffn_module(layers[0], handle.architecture) is layers[0].mlp
assert get_embedding_module(handle).num_embeddings == 16
assert handle.num_layers == 2
assert handle.hidden_size == 8
+104
View File
@@ -0,0 +1,104 @@
from __future__ import annotations
import json
from obliteratus.hard_negative import (
build_weighted_prompt_pairs,
classify_refusal_reason,
load_residue_file,
prompt_hash,
)
def test_load_residue_file_from_audit_summary(tmp_path):
audit = {
"n3_reg025": {
"refusals": [
{
"tier": "tier4_cyber_digital",
"global_index_1based": 2,
"prompt_sha256_12": "abc123",
"reason": "semantic: illegal privacy violation",
"response_preview": "This is illegal and a severe privacy violation.",
}
]
}
}
path = tmp_path / "summary.json"
path.write_text(json.dumps(audit))
rows = load_residue_file(path)
assert len(rows) == 1
assert rows[0].dataset == "builtin"
assert rows[0].index == 1
assert rows[0].tier == "tier4_cyber_digital"
assert "illegal_crime" in rows[0].tags
assert "privacy_intrusion" in rows[0].tags
def test_build_weighted_prompt_pairs_reconstructs_by_index(monkeypatch, tmp_path):
harmful = ["h0", "h1 stubborn", "h2"]
harmless = ["a0", "a1 benign counterpart", "a2"]
def fake_load_dataset_source(key):
assert key == "builtin"
return list(harmful), list(harmless)
monkeypatch.setattr("obliteratus.hard_negative.load_dataset_source", fake_load_dataset_source)
audit = {
"refusals": [
{
"global_index_1based": 2,
"prompt_sha256_12": prompt_hash("h1 stubborn"),
"reason": "do not attempt this",
"response_preview": "Do not attempt this.",
}
]
}
path = tmp_path / "audit.json"
path.write_text(json.dumps(audit))
h, a, meta = build_weighted_prompt_pairs(
base_dataset="builtin",
residue_files=[path],
residue_weight=3,
)
assert h[:3] == harmful
assert a[:3] == harmless
assert h[3:] == ["h1 stubborn"] * 3
assert a[3:] == ["a1 benign counterpart"] * 3
assert meta["residue_examples"] == 1
assert meta["residue_added_pairs"] == 3
assert meta["total_pairs"] == 6
meta_text = json.dumps(meta)
assert "h1 stubborn" not in meta_text
assert "a1 benign counterpart" not in meta_text
def test_hash_mismatch_skips_residue(monkeypatch, tmp_path):
monkeypatch.setattr(
"obliteratus.hard_negative.load_dataset_source",
lambda key: (["h0", "changed"], ["a0", "a1"]),
)
path = tmp_path / "audit.json"
path.write_text(json.dumps({"refusals": [{"global_index_1based": 2, "prompt_sha256_12": "badbadbadbad"}]}))
h, a, meta = build_weighted_prompt_pairs("builtin", [path], residue_weight=5)
assert h == ["h0", "changed"]
assert a == ["a0", "a1"]
assert meta["residue_examples"] == 0
assert meta["residue_added_pairs"] == 0
def test_classify_refusal_reason_tags():
tags = classify_refusal_reason(
"semantic illegal/crime pattern",
"Stop. This is a dangerous terrorism and privacy violation warning.",
)
assert "illegal_crime" in tags
assert "terrorism" in tags
assert "privacy_intrusion" in tags
assert "safety_banner" in tags
+58
View File
@@ -0,0 +1,58 @@
from __future__ import annotations
import json
from safetensors.torch import save_file
import torch
from obliteratus.model_profile import default_self_improve_params, estimate_total_params, profile_model
def test_estimate_total_params_uses_nested_text_config():
cfg = {
"model_type": "qwen3_5",
"text_config": {
"model_type": "qwen3_5_text",
"hidden_size": 128,
"num_hidden_layers": 2,
"num_attention_heads": 4,
"num_key_value_heads": 2,
"head_dim": 32,
"intermediate_size": 256,
"vocab_size": 1000,
},
}
assert estimate_total_params(cfg) is not None
assert estimate_total_params(cfg) > 0
def test_profile_model_counts_local_safetensors_exactly(tmp_path):
model_dir = tmp_path / "model"
model_dir.mkdir()
(model_dir / "config.json").write_text(json.dumps({
"model_type": "toy",
"hidden_size": 8,
"num_hidden_layers": 1,
"intermediate_size": 16,
"vocab_size": 32,
}))
save_file({
"a.weight": torch.zeros(2, 3),
"b.weight": torch.zeros(5),
}, model_dir / "model.safetensors")
profile = profile_model(str(model_dir), dtype="bfloat16")
assert profile.source == "local_safetensors"
assert profile.total_params == 11
assert profile.total_params_b == 0.000000
assert profile.hidden_size == 8
def test_default_self_improve_params_are_size_aware():
from obliteratus.model_profile import ModelProfile
big = ModelProfile("big", "test", int(27e9), 27.0, 27.0, 64, 5120, 17408, 248320, "qwen", "bf16")
tiny = ModelProfile("tiny", "test", int(1e9), 1.0, 1.0, 16, 2048, 8192, 32000, "toy", "bf16")
assert default_self_improve_params(big)["residue_weight"] < default_self_improve_params(tiny)["residue_weight"]
assert default_self_improve_params(big)["n_directions"] >= default_self_improve_params(tiny)["n_directions"]