test: close Gate 3 tiny-runtime semantics

This commit is contained in:
Joseph Magly
2026-08-16 07:53:00 -04:00
parent 9683e0be4d
commit acc6b3b254
13 changed files with 638 additions and 43 deletions
+73
View File
@@ -93,6 +93,15 @@ def test_method_statistics_ranges_and_bucket_ranking():
assert adaptive.BucketKnowledge(("dense", "standard", "tiny")).best_method is None
def test_method_statistics_ignore_config_keys_without_observed_values():
stats = adaptive.MethodStats(
"safe",
scores=[1.0, 0.9, 0.8, 0.7],
configs=[{"unused": None} for _ in range(4)],
)
assert stats.best_config_ranges() == {}
def test_build_knowledge_base_filters_invalid_runs_and_aggregates_metrics():
records = [
_record(session="one"),
@@ -110,6 +119,12 @@ def test_build_knowledge_base_filters_invalid_runs_and_aggregates_metrics():
assert bucket.methods["basic"].configs == []
def test_knowledge_build_fetches_records_when_not_supplied(monkeypatch):
monkeypatch.setattr(adaptive, "_fetch_all_records", lambda: [_record(session="fetched")])
knowledge = adaptive.build_knowledge_base()
assert knowledge[("dense", "standard", "medium")].total_runs == 1
def test_fetch_records_caches_deduplicates_and_tolerates_sources(monkeypatch):
adaptive._cache.clear()
monkeypatch.setattr(adaptive, "_cache_ts", 0.0)
@@ -166,6 +181,54 @@ def test_recommendation_exact_fallback_confidence_and_formatting():
assert "No telemetry data" in adaptive.format_recommendation(none)
def test_recommendation_merges_matching_architecture_and_reasoning_across_sizes():
records = [
_record(params_b=1, session=f"small-{index}")
for index in range(3)
] + [
_record(params_b=30, session=f"large-{index}")
for index in range(2)
]
recommendation = adaptive.get_adaptive_recommendation(
"dense",
"standard",
7,
knowledge=adaptive.build_knowledge_base(records),
)
assert recommendation.arch_key == ("dense", "standard", "*")
assert recommendation.n_records == 5
assert recommendation.n_method_records == 5
assert recommendation.confidence == "medium"
assert "all sizes" in recommendation.bucket_label
def test_recommendation_fetches_knowledge_and_reports_high_confidence(monkeypatch):
knowledge = adaptive.build_knowledge_base([
_record(session=f"high-{index}") for index in range(20)
])
monkeypatch.setattr(adaptive, "build_knowledge_base", lambda: knowledge)
recommendation = adaptive.get_adaptive_recommendation("dense", "standard", 7)
assert recommendation.confidence == "high"
assert recommendation.n_method_records == 20
def test_recommendation_handles_bucket_with_no_runnable_method():
key = ("dense", "standard", "medium")
empty_bucket = adaptive.BucketKnowledge(
key,
methods={"empty": adaptive.MethodStats("empty")},
total_runs=5,
)
recommendation = adaptive.get_adaptive_recommendation(
"dense",
"standard",
7,
knowledge={key: empty_bucket},
)
assert recommendation.confidence == "none"
assert "no method" in recommendation.reason
def test_global_insights_reports_rankings_buckets_and_hyperparameters():
knowledge = adaptive.build_knowledge_base([
_record("advanced", config={"strength": 1.5, "enabled": True}, session="one"),
@@ -176,3 +239,13 @@ def test_global_insights_reports_rankings_buckets_and_hyperparameters():
assert insights["overall_best_methods"][0]["method"] == "advanced"
assert insights["hyperparameter_trends"]["strength"]["type"] == "numeric"
assert insights["hyperparameter_trends"]["enabled"]["type"] == "bool"
def test_global_insights_fetches_knowledge_and_ignores_empty_config_values(monkeypatch):
knowledge = adaptive.build_knowledge_base([
_record(config={"unused": None}, session="empty-config")
])
monkeypatch.setattr(adaptive, "build_knowledge_base", lambda: knowledge)
insights = adaptive.get_global_insights()
assert insights["total_records"] == 1
assert "unused" not in insights["hyperparameter_trends"]
+105
View File
@@ -9,7 +9,10 @@ from obliteratus.evaluation.advanced_metrics import (
REFUSAL_PREFIXES,
REFUSAL_SUBSTRINGS,
AbliterationEvalResult,
_is_degenerate,
_is_refusal,
_is_refusal_detailed,
_strip_cot_tags,
activation_cosine_similarity,
effective_rank,
effective_rank_change,
@@ -27,6 +30,55 @@ from obliteratus.evaluation.advanced_metrics import (
# ---------------------------------------------------------------------------
class TestRefusalRate:
def test_cot_extraction_and_degenerate_classification_boundaries(self):
assert _strip_cot_tags("<assistant> final answer ") == "final answer"
assert _is_degenerate("") is True
assert _is_degenerate("!!!") is True
assert _is_degenerate("ababababab") is True
assert _is_degenerate("normal answer") is False
def test_empty_and_degenerate_post_cot_outputs_are_not_refusals(self):
assert _is_refusal("<assistant> ", mode="combined") is False
assert _is_refusal("!!!!!!!!", mode="combined") is False
assert _is_refusal_detailed(" ") == (False, "")
assert _is_refusal_detailed("<assistant> ") == (False, "")
assert _is_refusal_detailed("!!!!!!!!") == (False, "DEGENERATE")
@pytest.mark.parametrize(
("response", "mode", "reason_prefix"),
[
("I'm sorry, I cannot help with that.", "prefix", "prefix:"),
("Context first: I cannot assist with that request.", "substring", "substring:"),
("This raises serious safety concerns for users.", "semantic", "semantic:"),
("A normal helpful response.", "combined", ""),
],
)
def test_detailed_refusal_reasons_cover_each_decision_mode(
self,
response,
mode,
reason_prefix,
):
refused, reason = _is_refusal_detailed(response, mode=mode)
assert refused is bool(reason_prefix)
assert reason.startswith(reason_prefix)
def test_semantic_classifier_reports_positive_and_negative_matches(self):
from obliteratus.evaluation.advanced_metrics import _is_refusal_semantic
assert _is_refusal_semantic("This raises serious legal concerns.") is True
assert _is_refusal_semantic("A normal helpful response.") is False
def test_empty_confidence_interval_is_explicitly_unavailable(self):
assert refusal_rate_with_ci([]) == {
"available": False,
"rate": None,
"ci_lower": None,
"ci_upper": None,
"n_samples": 0,
"refusal_count": 0,
}
def test_refusal_mentioning_assistant_is_detected(self):
response = (
"I cannot help with that request. I am designed to be a helpful "
@@ -164,6 +216,15 @@ class TestRefusalRate:
# ---------------------------------------------------------------------------
class TestKLDivergence:
def test_rejects_wrong_tensor_dimension(self):
with pytest.raises(ValueError, match="both be 3D tensors"):
token_kl_divergence(torch.zeros(2, 3), torch.zeros(2, 3))
def test_first_token_kl_rejects_empty_sequence_axis(self):
empty = torch.empty(1, 0, 3)
with pytest.raises(ValueError, match="must not be empty"):
first_token_kl_divergence(empty, empty)
def test_identical_distributions(self):
"""KL divergence of identical distributions should be 0."""
logits = torch.randn(2, 10, 100)
@@ -239,6 +300,17 @@ class TestKLDivergence:
# ---------------------------------------------------------------------------
class TestEffectiveRank:
@pytest.mark.parametrize(
("matrix", "message"),
[
(torch.empty(0, 2), "non-empty"),
(torch.tensor([[float("inf")]]), "finite"),
],
)
def test_rejects_empty_and_nonfinite_matrices(self, matrix, message):
with pytest.raises(ValueError, match=message):
effective_rank(matrix)
def test_rank_one_matrix(self):
"""Rank-1 matrix should have effective rank close to 1."""
v = torch.randn(8, 1)
@@ -333,6 +405,9 @@ class TestActivationCosineSimilarity:
# ---------------------------------------------------------------------------
class TestLinearCKA:
def test_zero_centered_energy_returns_defined_zero(self):
assert linear_cka(torch.ones(2, 3), torch.ones(2, 4)) == 0.0
def test_identical_representations(self):
"""CKA of identical representations should be 1.0."""
X = torch.randn(20, 16)
@@ -400,6 +475,28 @@ class TestLinearCKA:
# ---------------------------------------------------------------------------
class TestRefusalProjection:
@pytest.mark.parametrize(
("activations", "direction", "message"),
[
(torch.ones(2), torch.ones(2), "activations"),
(torch.empty(0, 2), torch.ones(2), "activations"),
(torch.ones(2, 2), torch.empty(0), "refusal_direction"),
(torch.ones(2, 2), torch.ones(1, 1, 2), "refusal_direction"),
(torch.tensor([[float("nan"), 0.0]]), torch.ones(2), "finite"),
(torch.ones(1, 2), torch.tensor([float("inf"), 0.0]), "finite"),
],
)
def test_rejects_invalid_projection_inputs(self, activations, direction, message):
with pytest.raises(ValueError, match=message):
refusal_projection_magnitude(activations, direction)
def test_three_dimensional_activations_and_row_direction_are_normalized(self):
result = refusal_projection_magnitude(
torch.tensor([[[2.0, 0.0], [4.0, 0.0]]]),
torch.tensor([[2.0, 0.0]]),
)
assert result["mean"] == 3.0
def test_aligned_activations(self):
"""Activations aligned with direction should have high projection."""
d = torch.tensor([1.0, 0.0, 0.0])
@@ -447,6 +544,14 @@ class TestRefusalProjection:
# ---------------------------------------------------------------------------
class TestEvalReport:
@pytest.mark.parametrize(
("kl", "label"),
[(0.3, "good"), (0.7, "moderate degradation")],
)
def test_format_report_kl_quality_boundaries(self, kl, label):
result = AbliterationEvalResult(0.0, 0.0, kl, 1.0, 1.0, 1.0, 1.0)
assert label in format_eval_report(result)
def test_format_report(self):
result = AbliterationEvalResult(
refusal_rate_harmful=0.1,
+97 -1
View File
@@ -3,8 +3,10 @@
from __future__ import annotations
import builtins
import runpy
import sys
from pathlib import Path
from types import SimpleNamespace
from types import ModuleType, SimpleNamespace
from unittest.mock import MagicMock, Mock
import pytest
@@ -13,6 +15,100 @@ import torch
from obliteratus.models import loader
def _execute_loader_source() -> dict:
return runpy.run_path(str(Path(loader.__file__).resolve()), run_name="_loader_contract_probe")
def test_compatibility_shims_tolerate_independent_missing_or_broken_imports(monkeypatch):
original_import = builtins.__import__
failure_keys = {
("transformers", ("AutoModelForImageTextToText",)): ImportError,
("transformers.utils", ("output_capturing",)): ImportError,
("transformers.utils.import_utils", ()): RuntimeError,
("transformers.utils", ()): RuntimeError,
("transformers.pytorch_utils", ()): RuntimeError,
("transformers.generation_utils", ()): ModuleNotFoundError,
("transformers.generation", ()): RuntimeError,
("transformers.deepspeed", ()): ModuleNotFoundError,
("transformers.integrations.deepspeed", ()): RuntimeError,
("transformers.cache_utils", ("DynamicCache",)): RuntimeError,
("transformers.generation.logits_process", ()): RuntimeError,
("transformers.processing_utils", ()): RuntimeError,
("transformers.file_utils", ()): RuntimeError,
}
def rejecting_import(name, globals=None, locals=None, fromlist=(), level=0):
exception = failure_keys.get((name, tuple(fromlist or ())))
if exception is not None:
raise exception(f"injected loader compatibility failure for {name}")
return original_import(name, globals, locals, fromlist, level)
monkeypatch.setattr(builtins, "__import__", rejecting_import)
namespace = _execute_loader_source()
assert namespace["TASK_MODEL_MAP"]
assert namespace["AutoModelForImageTextToText"] is None
def test_compatibility_shims_tolerate_missing_generic_module(monkeypatch):
original_import = builtins.__import__
def rejecting_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == "transformers.utils.generic":
raise RuntimeError("generic module unavailable")
return original_import(name, globals, locals, fromlist, level)
monkeypatch.setattr(builtins, "__import__", rejecting_import)
assert _execute_loader_source()["TASK_MODEL_MAP"]
def test_compatibility_shims_tolerate_top_level_utils_patch_failure(monkeypatch):
original_import = builtins.__import__
def rejecting_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == "transformers.utils" and not fromlist:
raise RuntimeError("top-level utils module unavailable")
return original_import(name, globals, locals, fromlist, level)
monkeypatch.setattr(builtins, "__import__", rejecting_import)
assert _execute_loader_source()["TASK_MODEL_MAP"]
def test_dynamic_cache_compatibility_alias_is_installed_when_needed(monkeypatch):
original_import = builtins.__import__
class LegacyDynamicCache:
def get_max_cache_shape(self):
return 7
cache_module = SimpleNamespace(DynamicCache=LegacyDynamicCache)
def cache_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == "transformers.cache_utils" and tuple(fromlist or ()) == ("DynamicCache",):
return cache_module
return original_import(name, globals, locals, fromlist, level)
monkeypatch.setattr(builtins, "__import__", cache_import)
assert _execute_loader_source()["TASK_MODEL_MAP"]
assert LegacyDynamicCache().get_max_length() == 7
def test_compatibility_shims_tolerate_rejected_generic_attribute_patch(monkeypatch):
import transformers.utils as transformers_utils
class RejectWorkingDirectory(ModuleType):
def __setattr__(self, name, value):
if name == "working_or_temp_dir":
raise RuntimeError("read-only compatibility module")
super().__setattr__(name, value)
fake_generic = RejectWorkingDirectory("transformers.utils.generic")
monkeypatch.setitem(sys.modules, "transformers.utils.generic", fake_generic)
monkeypatch.setattr(transformers_utils, "generic", fake_generic)
assert _execute_loader_source()["TASK_MODEL_MAP"]
def _config(**overrides):
values = {
"model_type": "gpt2",
+44
View File
@@ -189,6 +189,50 @@ def test_full_pipeline_saves_and_reloads_a_real_offline_model(tmp_path):
assert all(duration >= 0 for duration in pipeline._stage_durations.values())
def test_multidirection_pipeline_restores_tiny_model_layer_norms(tmp_path):
source = build_tiny_offline_model(tmp_path / "source")
output = tmp_path / "output"
original = _state_dict(source)
layer_weights = {
name: tensor.float().norm().item()
for name, tensor in original.items()
if ".h.0." in name and tensor.ndim >= 2
}
pipeline = AbliterationPipeline(
model_name=str(source),
output_dir=str(output),
device="cpu",
dtype="float32",
method="advanced",
n_directions=2,
norm_preserve=True,
refinement_passes=1,
max_seq_length=8,
verify_sample_size=1,
refusal_max_tokens=1,
harmful_prompts=["harmful request", "harmful answer"],
harmless_prompts=["harmless request", "harmless answer"],
)
pipeline.run()
restored = _state_dict(output)
assert any(
not torch.equal(original[name], restored[name])
for name in layer_weights
)
for name, original_norm in layer_weights.items():
assert restored[name].float().norm().item() == pytest.approx(
original_norm,
rel=1e-5,
abs=1e-7,
)
metadata = json.loads((output / "abliteration_metadata.json").read_text())
assert metadata["method_config"]["n_directions"] == 2
assert metadata["method_config"]["norm_preserve"] is True
def test_installed_wheel_cli_loads_local_model_without_repository_imports(tmp_path):
source = build_tiny_offline_model(tmp_path / "source")
isolated_workdir = tmp_path / "outside-repository"
+148
View File
@@ -792,3 +792,151 @@ def test_packed_weight_replacement_uses_the_module_repacker():
assert len(calls) == 1
assert torch.equal(calls[0][0], modified)
assert torch.equal(calls[0][1], packed.scales)
def test_packed_weight_replacement_without_repacker_or_weight_fails_closed():
packed_type = type("QuantLinear", (), {})
packed = packed_type()
packed.qweight = torch.ones(1)
packed.scales = torch.tensor([0.5])
with pytest.raises(RuntimeError, match="cannot be re-packed or materialized"):
AbliterationPipeline._replace_quantized_weight(
packed,
torch.tensor([[1.0, 2.0]]),
)
def test_packed_weight_replacement_materializes_float_when_weight_is_writable():
packed_type = type("QuantLinear", (), {})
packed = packed_type()
packed.qweight = torch.ones(1)
packed.weight = torch.nn.Parameter(
torch.ones((1, 2), dtype=torch.uint8),
requires_grad=False,
)
modified = torch.tensor([[1.5, 2.5]])
with pytest.warns(UserWarning, match="Storing as float weight"):
AbliterationPipeline._replace_quantized_weight(packed, modified)
assert packed.weight.dtype is torch.float32
assert torch.equal(packed.weight, modified)
assert packed.weight.requires_grad is False
def test_layer_norm_capture_uses_logical_float_weights_and_deduplicates_ties(monkeypatch):
layer = torch.nn.Module()
layer.first = torch.nn.Linear(2, 2, bias=False)
layer.second = torch.nn.Linear(2, 2, bias=False)
shared = torch.nn.Parameter(
torch.tensor([[1, 2], [3, 4]], dtype=torch.uint8),
requires_grad=False,
)
layer.first.weight = shared
layer.second.weight = shared
calls = []
def dequantize(module):
calls.append(module)
return module.weight.data.float(), True
monkeypatch.setattr(
AbliterationPipeline,
"_dequantize_weight",
staticmethod(dequantize),
)
norms = AbliterationPipeline._capture_layer_weight_norms(layer)
assert norms == {"first.weight": pytest.approx(torch.tensor([1, 2, 3, 4]).float().norm().item())}
assert calls == [layer.first]
def test_integer_norm_restoration_materializes_float_and_preserves_tied_identity():
layer = torch.nn.Module()
layer.first = torch.nn.Linear(2, 2, bias=False)
layer.second = torch.nn.Linear(2, 2, bias=False)
shared = torch.nn.Parameter(torch.ones((2, 2), dtype=torch.uint8), requires_grad=False)
layer.first.weight = shared
layer.second.weight = shared
target_norm = shared.data.float().norm().item() * 1.05
AbliterationPipeline._restore_layer_weight_norms(
layer,
{"first.weight": target_norm},
)
assert layer.first.weight is shared
assert layer.second.weight is shared
assert shared.dtype is torch.float32
assert shared.norm().item() == pytest.approx(target_norm, rel=1e-6)
assert torch.equal(shared, torch.full((2, 2), 1.05))
def test_quantized_norm_restoration_uses_dequantize_and_replacement_seams(monkeypatch):
layer = torch.nn.Module()
layer.proj = torch.nn.Linear(2, 2, bias=False)
layer.tied = torch.nn.Linear(2, 2, bias=False)
layer.proj.weight.quant_state = object()
layer.tied.weight = layer.proj.weight
logical = torch.tensor([[1.0, 0.0], [0.0, 1.0]])
replacements = []
monkeypatch.setattr(
AbliterationPipeline,
"_dequantize_weight",
staticmethod(lambda _module: (logical.clone(), True)),
)
monkeypatch.setattr(
AbliterationPipeline,
"_replace_quantized_weight",
staticmethod(
lambda module, weight: (
replacements.append((module, weight.clone())),
setattr(module, "weight", torch.nn.Parameter(weight.clone())),
)
),
)
AbliterationPipeline._restore_layer_weight_norms(
layer,
{"proj.weight": logical.norm().item() * 0.5},
)
assert len(replacements) == 1
assert replacements[0][0] is layer.proj
assert replacements[0][1].norm().item() == pytest.approx(logical.norm().item() * 0.5)
assert layer.proj.weight is layer.tied.weight
def test_float_norm_restoration_scales_weight_in_place():
layer = torch.nn.Module()
layer.proj = torch.nn.Linear(2, 2, bias=False)
with torch.no_grad():
layer.proj.weight.fill_(1.0)
identity = layer.proj.weight
AbliterationPipeline._restore_layer_weight_norms(
layer,
{"proj.weight": 1.0},
)
assert layer.proj.weight is identity
assert layer.proj.weight.norm().item() == pytest.approx(1.0)
def test_norm_restoration_skips_degenerate_logical_weight():
layer = torch.nn.Module()
layer.proj = torch.nn.Linear(2, 2, bias=False)
with torch.no_grad():
layer.proj.weight.zero_()
identity = layer.proj.weight
AbliterationPipeline._restore_layer_weight_norms(
layer,
{"proj.weight": 2.0},
)
assert layer.proj.weight is identity
assert torch.count_nonzero(layer.proj.weight).item() == 0
+14 -5
View File
@@ -161,6 +161,15 @@ def test_mutation_score_floor_is_immutable_across_policy_ci_and_validator():
assert "--minimum 85.0" in workflow
def test_mature_cpu_coverage_floors_are_immutable_across_policy_and_validator():
policy = json.loads(Path("ci/test-quality-policy.json").read_text(encoding="utf-8"))
assert quality.BASELINE_FLOORS["mature_cpu_statement"] == 94.0
assert quality.BASELINE_FLOORS["mature_cpu_branch"] == 84.0
assert policy["minimums"]["mature_cpu_statement"] == 94.0
assert policy["minimums"]["mature_cpu_branch"] == 84.0
def _policy():
return {
"minimums": dict(quality.BASELINE_FLOORS),
@@ -214,9 +223,9 @@ def _coverage():
"obliteratus/pure.py": {
"summary": {
"num_statements": 100,
"covered_lines": 92,
"covered_lines": 94,
"num_branches": 100,
"covered_branches": 80,
"covered_branches": 84,
},
},
"obliteratus/external.py": {
@@ -651,8 +660,8 @@ def test_policy_and_exact_mature_floors_pass():
assert quality.validate_policy(policy) == []
measurement, failures = quality.validate_mature_cpu_scope(_coverage(), policy)
assert failures == []
assert measurement["line_percent"] == 92
assert measurement["branch_percent"] == 80
assert measurement["line_percent"] == 94
assert measurement["branch_percent"] == 84
def test_floor_regression_requires_structured_reviewed_exception():
@@ -1002,7 +1011,7 @@ def test_mature_scope_rejects_regression_and_stale_exclusion():
report["files"]["obliteratus/pure.py"]["summary"]["covered_lines"] = 91
_, failures = quality.validate_mature_cpu_scope(report, policy)
assert failures == [
"mature CPU line coverage 91.00% is below the 92.00% floor",
"mature CPU line coverage 91.00% is below the 94.00% floor",
]
del report["files"]["obliteratus/external.py"]
_, failures = quality.measure_mature_cpu_scope(report, policy)
+37
View File
@@ -12,6 +12,7 @@ from obliteratus.runtime_contracts import (
classify_architecture_size,
effective_model_memory_gb,
is_quantized_parameter,
norm_restoration_ratio,
quantized_model_fits_gpu,
resolve_model_load_policy,
should_snapshot_model,
@@ -20,6 +21,42 @@ from obliteratus.runtime_contracts import (
)
@pytest.mark.parametrize(
("original", "current", "expected"),
[
(10.0, 10.0, None),
(0.0, 4.0, None),
(10.0, 0.0, None),
(float("nan"), 4.0, None),
(10.0, float("inf"), None),
(1.0, 0.5, 1.1),
(8.0, 10.0, 0.8),
(12.0, 10.0, 1.1),
],
)
def test_norm_restoration_ratio_is_bounded_and_fail_closed(original, current, expected):
assert norm_restoration_ratio(original, current, max_ratio=1.1) == expected
def test_norm_restoration_ratio_rejects_invalid_policy_bounds():
with pytest.raises(ValueError) as error:
norm_restoration_ratio(2.0, 1.0, max_ratio=0.0)
assert str(error.value) == "max norm ratio must be a positive finite number"
def test_norm_restoration_ratio_allows_subunit_policy_cap():
assert norm_restoration_ratio(2.0, 1.0, max_ratio=0.5) == 0.5
def test_norm_restoration_ratio_tolerance_boundary_is_a_noop():
assert norm_restoration_ratio(
1.0,
1.25,
max_ratio=1.1,
tolerance=0.25,
) is None
@pytest.mark.parametrize("model_name", [None, 7, "", " \t"])
def test_model_name_must_be_a_nonempty_string(model_name):
with pytest.raises(ValueError) as error: