From acc6b3b254e6d358ddabe495b1ca79295dd36d70 Mon Sep 17 00:00:00 2001 From: Joseph Magly <1159087+jmagly@users.noreply.github.com> Date: Sun, 16 Aug 2026 07:53:00 -0400 Subject: [PATCH] test: close Gate 3 tiny-runtime semantics --- ci/test-quality-policy.json | 4 +- ci/test-risk-map.json | 1 + obliteratus/abliterate.py | 119 ++++++++++++----- obliteratus/evaluation/advanced_metrics.py | 2 - obliteratus/runtime_contracts.py | 27 ++++ scripts/check_quality_policy.py | 4 +- tests/test_adaptive_defaults.py | 73 ++++++++++ tests/test_advanced_metrics.py | 105 +++++++++++++++ tests/test_loader_boundaries.py | 98 +++++++++++++- tests/test_offline_integration.py | 44 ++++++ tests/test_projection_math_contracts.py | 148 +++++++++++++++++++++ tests/test_quality_policy.py | 19 ++- tests/test_runtime_contracts.py | 37 ++++++ 13 files changed, 638 insertions(+), 43 deletions(-) diff --git a/ci/test-quality-policy.json b/ci/test-quality-policy.json index 94cf419..115587c 100644 --- a/ci/test-quality-policy.json +++ b/ci/test-quality-policy.json @@ -4,8 +4,8 @@ "repository_statement": 75.0, "repository_branch": 60.0, "changed_line": 95.0, - "mature_cpu_statement": 92.0, - "mature_cpu_branch": 80.0, + "mature_cpu_statement": 94.0, + "mature_cpu_branch": 84.0, "mutation_score": 85.0, "warning_budget": 0 }, diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json index ef0aa2c..31d0bb6 100644 --- a/ci/test-risk-map.json +++ b/ci/test-risk-map.json @@ -378,6 +378,7 @@ "tests/test_checkpoint_atomicity.py", "tests/test_persistence_contracts.py", "tests/test_persistence_pipeline.py", + "tests/test_projection_math_contracts.py", "tests/test_offline_integration.py" ], "conditional_gates": ["model-download-runtime"] diff --git a/obliteratus/abliterate.py b/obliteratus/abliterate.py index fa5fa20..3a6a212 100644 --- a/obliteratus/abliterate.py +++ b/obliteratus/abliterate.py @@ -59,6 +59,7 @@ from obliteratus.runtime_contracts import ( # noqa: E402 attention_projection_names, classify_weight_storage, is_quantized_parameter, + norm_restoration_ratio, ) from obliteratus.strategies.utils import ( # noqa: E402 get_attention_module, @@ -4686,19 +4687,23 @@ class AbliterationPipeline: return except (AttributeError, RuntimeError, TypeError): pass - # Fallback: store as float weight (loses quantization benefits - # but preserves correctness) + if weight is None: + raise RuntimeError( + f"Packed quantized module {module_cls} cannot be re-packed or " + "materialized because it exposes neither pack() nor a writable weight." + ) + # Fallback: store as float weight (loses quantization benefits but + # preserves the projected values and makes the loss explicit). warnings.warn( f"Cannot re-pack {module_cls} after projection. Storing as " f"float weight — inference will use more memory but remain " f"correct. Save and re-quantize the model for efficient serving.", stacklevel=3, ) - if hasattr(proj_module, "weight"): - proj_module.weight = nn.Parameter( - W_modified.to(device=proj_module.qweight.device), - requires_grad=False, - ) + proj_module.weight = nn.Parameter( + W_modified.to(device=proj_module.qweight.device), + requires_grad=False, + ) return # ── Non-float weight (e.g. uint8 from custom quantization) ───── @@ -4747,13 +4752,30 @@ class AbliterationPipeline: including the zero'd-out direction). Works recursively, covering attention, FFN, MoE experts, routers, - and shared experts uniformly. + and shared experts uniformly. Packed weights are measured through their + logical dequantized values, and tied parameters are captured once. """ norms: dict[str, float] = {} - for param_name, param in layer.named_parameters(): - if param_name.endswith(".weight"): - data = param.data.float() if not param.data.is_floating_point() else param.data - norms[param_name] = data.norm().item() + seen: set[int] = set() + for module_name, module in layer.named_modules(): + weight = getattr(module, "weight", None) + storage_kind = classify_weight_storage( + module_class_name=module.__class__.__name__, + parameter_class_name=weight.__class__.__name__ if weight is not None else "", + has_quant_state=hasattr(weight, "quant_state"), + data_is_floating_point=( + weight.data.is_floating_point() if weight is not None else False + ), + ) + if weight is None and storage_kind != "packed_module": + continue + identity = id(weight) if weight is not None else id(module) + if identity in seen: + continue + param_name = f"{module_name}.weight" if module_name else "weight" + data, _requires_replacement = AbliterationPipeline._dequantize_weight(module) + norms[param_name] = data.float().norm().item() + seen.add(identity) return norms @staticmethod @@ -4767,28 +4789,63 @@ class AbliterationPipeline: out, ensuring the norm-preservation rescaling doesn't reintroduce previously removed directional components. """ - for param_name, param in layer.named_parameters(): + seen: set[int] = set() + module_entries = list(layer.named_modules()) + aliases: dict[int, list[nn.Module]] = {} + for _module_name, module in module_entries: + weight = getattr(module, "weight", None) + if weight is not None: + aliases.setdefault(id(weight), []).append(module) + + for module_name, module in module_entries: + weight = getattr(module, "weight", None) + storage_kind = classify_weight_storage( + module_class_name=module.__class__.__name__, + parameter_class_name=weight.__class__.__name__ if weight is not None else "", + has_quant_state=hasattr(weight, "quant_state"), + data_is_floating_point=( + weight.data.is_floating_point() if weight is not None else False + ), + ) + if weight is None and storage_kind != "packed_module": + continue + identity = id(weight) if weight is not None else id(module) + if identity in seen: + continue + param_name = f"{module_name}.weight" if module_name else "weight" if param_name not in saved_norms: continue + original_norm = saved_norms[param_name] - if original_norm > 0: - needs_cast = not param.data.is_floating_point() - data = param.data.float() if needs_cast else param.data - new_norm = data.norm().item() - if math.isnan(new_norm) or math.isinf(new_norm) or new_norm == 0: - continue # Skip — weight is degenerate after projection - if abs(new_norm - original_norm) > 1e-6: - ratio = original_norm / new_norm - # Cap amplification to prevent compound norm drift across - # layers. Uncapped amplification destroys coherence. - if ratio > _MAX_NORM_RATIO: - ratio = _MAX_NORM_RATIO - if needs_cast: - # Non-float dtypes (e.g. uint8) can't mul_ by a float - # scalar in-place — rescale in float then cast back. - param.data.copy_(data.mul_(ratio).to(param.data.dtype)) - else: - param.data.mul_(ratio) + data, requires_replacement = AbliterationPipeline._dequantize_weight(module) + ratio = norm_restoration_ratio( + original_norm, + data.float().norm().item(), + max_ratio=_MAX_NORM_RATIO, + ) + if ratio is None: + seen.add(identity) + continue + + with torch.no_grad(): + if storage_kind == "integer": + # Integer storage has no scale/zero-point contract. Casting + # back can erase the update, so materialize logical float + # values while preserving Parameter identity and ties. + weight.data = data.mul(ratio).to(device=weight.device) + elif requires_replacement: + AbliterationPipeline._replace_quantized_weight( + module, + data.mul(ratio), + ) + replacement = getattr(module, "weight", None) + if replacement is not None: + for alias in aliases.get(identity, []): + if alias is not module: + alias.weight = replacement + else: + data.mul_(ratio) + seen.add(identity) @staticmethod def _select_projection_coefficients( diff --git a/obliteratus/evaluation/advanced_metrics.py b/obliteratus/evaluation/advanced_metrics.py index 9e9f8fe..a834df5 100644 --- a/obliteratus/evaluation/advanced_metrics.py +++ b/obliteratus/evaluation/advanced_metrics.py @@ -535,8 +535,6 @@ def first_token_kl_divergence( name="first-token KL", dimensions=(3,), ) - if logits_original.shape[1] == 0: - raise ValueError("first-token KL requires at least one sequence position") # Take logits at the last input position (predicting first generated token) first_logits_orig = logits_original[:, -1, :] # (batch, vocab) first_logits_mod = logits_modified[:, -1, :] diff --git a/obliteratus/runtime_contracts.py b/obliteratus/runtime_contracts.py index d7c1db8..f4fc8e6 100644 --- a/obliteratus/runtime_contracts.py +++ b/obliteratus/runtime_contracts.py @@ -2,6 +2,7 @@ from __future__ import annotations +import math from collections.abc import Collection from dataclasses import dataclass from typing import Literal @@ -147,6 +148,32 @@ def classify_weight_storage( return "float" +def norm_restoration_ratio( + original_norm: float, + current_norm: float, + *, + max_ratio: float, + tolerance: float = 1e-6, +) -> float | None: + """Return a bounded rescale ratio or ``None`` for a no-op restoration. + + Non-finite or degenerate norms fail closed. Restoring a smaller norm is + allowed exactly, while amplification is capped by the caller-owned policy. + """ + if not math.isfinite(max_ratio) or max_ratio <= 0: + raise ValueError("max norm ratio must be a positive finite number") + if ( + not math.isfinite(original_norm) + or not math.isfinite(current_norm) + or original_norm <= 0 + or current_norm <= 0 + ): + return None + if abs(current_norm - original_norm) <= tolerance: + return None + return min(original_norm / current_norm, max_ratio) + + def effective_model_memory_gb(estimate_gb: float, quantization: str | None) -> float: """Adjust a full-precision weight estimate for runtime quantization.""" factor = {"4bit": 4, "8bit": 2}.get(quantization, 1) diff --git a/scripts/check_quality_policy.py b/scripts/check_quality_policy.py index 1648f67..479e374 100644 --- a/scripts/check_quality_policy.py +++ b/scripts/check_quality_policy.py @@ -16,8 +16,8 @@ BASELINE_FLOORS = { "repository_statement": 75.0, "repository_branch": 60.0, "changed_line": 95.0, - "mature_cpu_statement": 92.0, - "mature_cpu_branch": 80.0, + "mature_cpu_statement": 94.0, + "mature_cpu_branch": 84.0, "mutation_score": 85.0, "warning_budget": 0.0, } diff --git a/tests/test_adaptive_defaults.py b/tests/test_adaptive_defaults.py index 7857abc..99df732 100644 --- a/tests/test_adaptive_defaults.py +++ b/tests/test_adaptive_defaults.py @@ -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"] diff --git a/tests/test_advanced_metrics.py b/tests/test_advanced_metrics.py index 329f73d..1fc3b2f 100644 --- a/tests/test_advanced_metrics.py +++ b/tests/test_advanced_metrics.py @@ -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(" 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(" ", mode="combined") is False + assert _is_refusal("!!!!!!!!", mode="combined") is False + assert _is_refusal_detailed(" ") == (False, "") + assert _is_refusal_detailed(" ") == (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, diff --git a/tests/test_loader_boundaries.py b/tests/test_loader_boundaries.py index c2b0507..d60acdf 100644 --- a/tests/test_loader_boundaries.py +++ b/tests/test_loader_boundaries.py @@ -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", diff --git a/tests/test_offline_integration.py b/tests/test_offline_integration.py index c50acf4..cdc0c3b 100644 --- a/tests/test_offline_integration.py +++ b/tests/test_offline_integration.py @@ -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" diff --git a/tests/test_projection_math_contracts.py b/tests/test_projection_math_contracts.py index e8433ad..9933dd0 100644 --- a/tests/test_projection_math_contracts.py +++ b/tests/test_projection_math_contracts.py @@ -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 diff --git a/tests/test_quality_policy.py b/tests/test_quality_policy.py index 9437976..d8e4454 100644 --- a/tests/test_quality_policy.py +++ b/tests/test_quality_policy.py @@ -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) diff --git a/tests/test_runtime_contracts.py b/tests/test_runtime_contracts.py index 3ff0e89..53382d6 100644 --- a/tests/test_runtime_contracts.py +++ b/tests/test_runtime_contracts.py @@ -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: