test: enforce runtime boundary contracts

2026-08-17 08:31:14 +02:00 · 2026-08-14 22:50:37 -04:00
parent 15aa3dc163
commit 2c5dc5442c
13 changed files with 702 additions and 98 deletions
@@ -8,12 +8,12 @@
    "all": {
      "version": "2026.8.8",
      "source": "bundled",
-      "installedAt": "2026-08-14T13:41:42.906Z",
+      "installedAt": "2026-08-15T02:40:58.412Z",
      "deployedTo": {
        "codex": {
          "agents": 0,
          "commands": 50,
-          "skills": 30,
+          "skills": 31,
          "rules": 2
        }
      },
@@ -22,7 +22,7 @@
    "bt6-maintainer": {
      "version": "0.2.0",
      "source": "project-local",
-      "installedAt": "2026-08-04T07:41:37.856Z",
+      "installedAt": "2026-08-15T02:40:59.927Z",
      "deployedTo": {
        "codex": {
          "agents": 5,
@@ -10,6 +10,7 @@
    "warning_budget": 0
  },
  "critical_cpu_paths": [
+    "obliteratus/runtime_contracts.py",
    "obliteratus/device.py",
    "obliteratus/models/loader.py",
    "obliteratus/architecture_profiles.py",
@@ -67,6 +67,7 @@
        "obliteratus/model_profile.py",
        "obliteratus/models/loader.py",
        "obliteratus/presets.py",
+        "obliteratus/runtime_contracts.py",
        "obliteratus/study_presets.py"
      ],
      "required_tests": [
@@ -78,6 +79,7 @@
        "tests/test_loader_boundaries.py",
        "tests/test_mlx_backend_boundaries.py",
        "tests/test_model_profile.py",
+        "tests/test_runtime_contracts.py",
        "tests/test_study_presets.py"
      ]
    },
@@ -292,6 +294,13 @@
      "required_tests": ["tests/test_device_boundaries.py"],
      "conditional_gates": ["cuda-runtime", "mps-runtime"]
    },
+    {
+      "path": "obliteratus/runtime_contracts.py",
+      "risk_class": "cpu-contract",
+      "risk": "model request, memory, architecture scale, and dtype capability decisions",
+      "required_tests": ["tests/test_runtime_contracts.py"],
+      "conditional_gates": []
+    },
    {
      "path": "obliteratus/models/loader.py",
      "risk_class": "mixed-runtime",
@@ -22,6 +22,8 @@ from dataclasses import dataclass, field
 from enum import Enum
 from typing import TYPE_CHECKING, Any

+from obliteratus.runtime_contracts import classify_architecture_size
+
 if TYPE_CHECKING:
    from obliteratus.adaptive_defaults import AdaptiveRecommendation

@@ -240,27 +242,14 @@ def detect_architecture(
                break

    # ── Step 3: Classify architecture ────────────────────────────────
-    if is_moe:
-        # Classification priority:
-        # 1. If total params known → use param threshold (100B)
-        # 2. Else if expert count known → use expert threshold (16)
-        # 3. Else fall back to name patterns → default SMALL_MOE (conservative)
-        if total_params_b > 0:
-            is_small = total_params_b < 100
-        elif num_experts > 0:
-            is_small = num_experts <= 16
-        else:
-            # No config available — use name heuristics.
-            # Check large patterns first (more specific).
-            is_small = True
-            for pattern in _LARGE_MOE_NAME_PATTERNS:
-                if pattern.lower() in name_lower:
-                    is_small = False
-                    break
-
-        arch_class = ArchitectureClass.SMALL_MOE if is_small else ArchitectureClass.LARGE_MOE
-    else:
-        arch_class = ArchitectureClass.DENSE
+    size = classify_architecture_size(
+        is_moe=is_moe,
+        total_params_b=total_params_b,
+        num_experts=num_experts,
+        model_name=model_name,
+        large_moe_name_patterns=_LARGE_MOE_NAME_PATTERNS,
+    )
+    arch_class = ArchitectureClass(size)

    reasoning_class = (
        ReasoningClass.REASONING if is_reasoning else ReasoningClass.STANDARD
@@ -14,6 +14,8 @@ from dataclasses import dataclass

 import torch

+from obliteratus.runtime_contracts import supports_bfloat16_target
+
 logger = logging.getLogger(__name__)

 # ---------------------------------------------------------------------------
@@ -256,17 +258,18 @@ def default_dtype(device: str | None = None) -> torch.dtype:
 def supports_bfloat16(device: str | None = None) -> bool:
    """Whether *bfloat16* is natively supported on the target device."""
    dev = device or get_device()
-    if dev.startswith("cuda"):
-        if is_cuda():
-            major, _ = torch.cuda.get_device_capability(0)
-            return major >= 8  # Ampere+
-        return False
-    if dev == "mps":
-        # MPS added bfloat16 support in PyTorch 2.3+
-        return hasattr(torch, "__version__") and tuple(
-            int(x) for x in torch.__version__.split(".")[:2]
-        ) >= (2, 3)
-    return True  # CPU supports bfloat16 on most modern platforms
+    cuda_available = is_cuda() if dev.startswith("cuda") else False
+    cuda_major = (
+        torch.cuda.get_device_capability(0)[0]
+        if dev.startswith("cuda") and cuda_available
+        else None
+    )
+    return supports_bfloat16_target(
+        dev,
+        cuda_available=cuda_available,
+        cuda_major=cuda_major,
+        torch_version=torch.__version__,
+    )


 def supports_float64(device: str | None = None) -> bool:
@@ -12,6 +12,12 @@ import sys as _sys

 import torch
 from obliteratus import device as dev
+from obliteratus.runtime_contracts import (
+    effective_model_memory_gb,
+    quantized_model_fits_gpu,
+    should_snapshot_model,
+    validate_model_load_request,
+)
 from transformers import (
    AutoConfig,
    AutoModelForCausalLM,
@@ -218,22 +224,28 @@ except Exception:

 # ── 11. file_utils.cached_path → huggingface_hub fallback ──────────
 # Removed in v4.22.  Very old model repos use it for file download.
+def _cached_path_shim(url_or_filename, cache_dir=None, **kwargs):
+    """Best-effort replacement for the removed legacy cached-path helper."""
+    if os.path.exists(str(url_or_filename)):
+        return str(url_or_filename)
+    try:
+        from huggingface_hub import hf_hub_download
+
+        parts = str(url_or_filename).rsplit("/", 1)
+        if len(parts) == 2:
+            return hf_hub_download(
+                repo_id=parts[0],
+                filename=parts[1],
+                cache_dir=cache_dir,
+            )
+    except Exception:
+        pass
+    return str(url_or_filename)
+
+
 try:
    import transformers.file_utils as _fu
    if not hasattr(_fu, "cached_path"):
-        def _cached_path_shim(url_or_filename, cache_dir=None, **kwargs):
-            """Minimal shim: local paths pass through, HF paths download."""
-            if os.path.exists(str(url_or_filename)):
-                return str(url_or_filename)
-            try:
-                from huggingface_hub import hf_hub_download
-                parts = str(url_or_filename).rsplit("/", 1)
-                if len(parts) == 2:
-                    return hf_hub_download(repo_id=parts[0], filename=parts[1],
-                                           cache_dir=cache_dir)
-            except Exception:
-                pass
-            return str(url_or_filename)
        _fu.cached_path = _cached_path_shim
 except Exception:
    pass
@@ -467,8 +479,7 @@ def _is_quantization_state_key(key: str) -> bool:

 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)
-    return estimate_gb / factor
+    return effective_model_memory_gb(estimate_gb, quantization)


 def _bounded_max_memory() -> dict[int | str, str]:
@@ -529,18 +540,15 @@ def load_model(
    """
    _apply_deferred_shims()

-    if not isinstance(model_name, str) or not model_name.strip():
-        raise ValueError("model_name must be a non-empty HuggingFace identifier or local path")
-    if task not in TASK_MODEL_MAP:
-        raise ValueError(f"Unknown task {task!r}. Choose from {list(TASK_MODEL_MAP)}")
-    if quantization not in (None, "4bit", "8bit"):
-        raise ValueError(
-            "Unknown quantization {!r}. Choose None, '4bit', or '8bit'".format(quantization),
-        )
+    validate_model_load_request(
+        model_name,
+        task,
+        quantization,
+        dtype,
+        valid_tasks=TASK_MODEL_MAP,
+    )

    dtype_map = {"float32": torch.float32, "float16": torch.float16, "bfloat16": torch.bfloat16}
-    if dtype not in dtype_map:
-        raise ValueError(f"Unknown dtype {dtype!r}. Choose from {list(dtype_map)}")
    torch_dtype = dtype_map[dtype]
    resolved_device = dev.get_device(device)
    if dtype == "bfloat16" and not dev.supports_bfloat16(resolved_device):
@@ -679,12 +687,7 @@ def load_model(
            logger.info(f"Auto-created offload folder: {_offload_dir}")

        effective_est_gb = _effective_model_memory_gb(est_gb, quantization)
-        quantized_fit = (
-            quantization in ("4bit", "8bit")
-            and effective_est_gb > 0
-            and gpu_gb > 0
-            and effective_est_gb < gpu_gb * 0.7
-        )
+        quantized_fit = quantized_model_fits_gpu(est_gb, quantization, gpu_gb)
        if quantized_fit:
            logger.info(
                f"Quantized estimate ({effective_est_gb:.1f} GB) fits GPU "
@@ -765,37 +768,34 @@ def load_model(
        _owns_offload_dir=_owns_offload_dir,
    )

-    # Skip snapshot for large models to avoid doubling memory usage
-    if skip_snapshot is True:
-        pass  # user explicitly opted out
-    elif skip_snapshot is False:
-        handle.snapshot()  # user explicitly forced snapshot
-    else:
-        # Auto-decide: skip when GPU is present and memory is tight.
-        # For natively quantized models (est_gb==0), check actual GPU usage instead.
-        if gpu_gb > 0 and native_quant is not None:
-            # Model is pre-quantized but we can't estimate its true size.
-            # Check actual free memory after loading — if less than 40% free, skip snapshot.
-            free_gb = dev.get_total_free_gb()
-            if free_gb < gpu_gb * 0.4:
-                logger.warning(
-                    f"Auto-skipping state dict snapshot for natively quantized model "
-                    f"(free GPU: {free_gb:.1f} GB / {gpu_gb:.1f} GB). "
-                    f"Use skip_snapshot=False to force."
-                )
-            else:
-                handle.snapshot()
-        elif gpu_gb > 0 and est_gb > 0:
-            effective_gb = _effective_model_memory_gb(est_gb, quantization)
-            if effective_gb > gpu_gb * 0.5:
-                logger.warning(
-                    f"Auto-skipping state dict snapshot to save memory "
-                    f"(model ~{effective_gb:.0f} GB vs GPU {gpu_gb:.0f} GB). "
-                    f"Use skip_snapshot=False to force."
-                )
-            else:
-                handle.snapshot()
-        else:
-            handle.snapshot()
+    # Skip snapshot for large models to avoid doubling memory usage.
+    remaining_gpu_gb = (
+        dev.get_total_free_gb()
+        if skip_snapshot is None and gpu_gb > 0 and native_quant is not None
+        else gpu_gb
+    )
+    snapshot = should_snapshot_model(
+        skip_snapshot=skip_snapshot,
+        initial_gpu_free_gb=gpu_gb,
+        remaining_gpu_free_gb=remaining_gpu_gb,
+        has_native_quantization=native_quant is not None,
+        estimate_gb=est_gb,
+        quantization=quantization,
+    )
+    if snapshot:
+        handle.snapshot()
+    elif skip_snapshot is None and gpu_gb > 0 and native_quant is not None:
+        logger.warning(
+            f"Auto-skipping state dict snapshot for natively quantized model "
+            f"(free GPU: {remaining_gpu_gb:.1f} GB / {gpu_gb:.1f} GB). "
+            f"Use skip_snapshot=False to force."
+        )
+    elif skip_snapshot is None and gpu_gb > 0 and est_gb > 0:
+        effective_gb = _effective_model_memory_gb(est_gb, quantization)
+        logger.warning(
+            f"Auto-skipping state dict snapshot to save memory "
+            f"(model ~{effective_gb:.0f} GB vs GPU {gpu_gb:.0f} GB). "
+            f"Use skip_snapshot=False to force."
+        )

    return handle
@@ -0,0 +1,111 @@
+"""Pure contracts shared by model loading and runtime selection paths."""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+from typing import Literal
+
+
+VALID_MODEL_TASKS = ("causal_lm", "classification")
+VALID_QUANTIZATIONS = (None, "4bit", "8bit")
+VALID_DTYPES = ("float32", "float16", "bfloat16")
+
+
+def validate_model_load_request(
+    model_name: object,
+    task: str,
+    quantization: str | None,
+    dtype: str,
+    *,
+    valid_tasks: Collection[str] = VALID_MODEL_TASKS,
+) -> None:
+    """Reject malformed loader requests before any provider access."""
+    if not isinstance(model_name, str) or not model_name.strip():
+        raise ValueError("model_name must be a non-empty HuggingFace identifier or local path")
+    if task not in valid_tasks:
+        raise ValueError(f"Unknown task {task!r}. Choose from {list(valid_tasks)}")
+    if quantization not in VALID_QUANTIZATIONS:
+        raise ValueError(
+            "Unknown quantization {!r}. Choose None, '4bit', or '8bit'".format(quantization),
+        )
+    if dtype not in VALID_DTYPES:
+        raise ValueError(f"Unknown dtype {dtype!r}. Choose from {list(VALID_DTYPES)}")
+
+
+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)
+    return estimate_gb / factor
+
+
+def quantized_model_fits_gpu(
+    estimate_gb: float,
+    quantization: str | None,
+    available_gpu_gb: float,
+) -> bool:
+    """Return whether a quantized estimate fits with 30 percent headroom."""
+    effective_gb = effective_model_memory_gb(estimate_gb, quantization)
+    return (
+        quantization in ("4bit", "8bit")
+        and effective_gb > 0
+        and effective_gb < available_gpu_gb * 0.7
+    )
+
+
+def should_snapshot_model(
+    *,
+    skip_snapshot: bool | None,
+    initial_gpu_free_gb: float,
+    remaining_gpu_free_gb: float,
+    has_native_quantization: bool,
+    estimate_gb: float,
+    quantization: str | None,
+) -> bool:
+    """Decide whether a restorable state snapshot fits the memory policy."""
+    if skip_snapshot is True:
+        return False
+    if skip_snapshot is False:
+        return True
+    if initial_gpu_free_gb > 0 and has_native_quantization:
+        return remaining_gpu_free_gb >= initial_gpu_free_gb * 0.4
+    if initial_gpu_free_gb > 0:
+        effective_gb = effective_model_memory_gb(estimate_gb, quantization)
+        return effective_gb <= initial_gpu_free_gb * 0.5
+    return True
+
+
+def classify_architecture_size(
+    *,
+    is_moe: bool,
+    total_params_b: float,
+    num_experts: int,
+    model_name: str,
+    large_moe_name_patterns: Collection[str],
+) -> Literal["dense", "small_moe", "large_moe"]:
+    """Classify dense and MoE scale using the documented precedence rules."""
+    if not is_moe:
+        return "dense"
+    if total_params_b > 0:
+        is_small = total_params_b < 100
+    elif num_experts > 0:
+        is_small = num_experts <= 16
+    else:
+        name_lower = model_name.lower()
+        is_small = not any(pattern.lower() in name_lower for pattern in large_moe_name_patterns)
+    return "small_moe" if is_small else "large_moe"
+
+
+def supports_bfloat16_target(
+    device: str,
+    *,
+    cuda_available: bool,
+    cuda_major: int | None,
+    torch_version: str,
+) -> bool:
+    """Evaluate the bfloat16 capability contract from deterministic inputs."""
+    if device.startswith("cuda"):
+        return cuda_available and cuda_major is not None and cuda_major >= 8
+    if device == "mps":
+        major, minor = (int(value) for value in torch_version.split(".")[:2])
+        return (major, minor) >= (2, 3)
+    return True
@@ -118,6 +118,7 @@ source_paths = ["obliteratus/", "scripts/"]
 only_mutate = [
    "obliteratus/config.py",
    "obliteratus/analysis/numerical_contracts.py",
+    "obliteratus/runtime_contracts.py",
    "scripts/check_coverage_thresholds.py",
 ]
 pytest_add_cli_args = ["--no-cov", "-q"]
@@ -126,6 +127,7 @@ pytest_add_cli_args_test_selection = [
    "tests/test_config_properties.py",
    "tests/test_coverage_thresholds.py",
    "tests/test_numerical_contracts.py",
+    "tests/test_runtime_contracts.py",
 ]
 mutate_only_covered_lines = true
 on_dependency_change = "rerun"
@@ -6,7 +6,10 @@ architecture class (dense/MoE, standard/reasoning).

 from __future__ import annotations

+import builtins
+from types import SimpleNamespace

+from obliteratus import architecture_profiles as profiles
 from obliteratus.architecture_profiles import (
    ArchitectureClass,
    ArchitectureProfile,
@@ -596,3 +599,124 @@ class TestEdgeCases:
        profile = detect_architecture("custom/single-expert", config=MockConfig())
        # 1 expert still triggers MoE detection (the code treats any >0 as MoE)
        assert profile.is_moe
+
+
+def _telemetry_profile() -> ArchitectureProfile:
+    return detect_architecture("meta-llama/Llama-3.1-8B-Instruct")
+
+
+def test_telemetry_import_failure_preserves_research_defaults(monkeypatch):
+    profile = _telemetry_profile()
+    original_method = profile.recommended_method
+    real_import = builtins.__import__
+
+    def reject_adaptive_defaults(name, *args, **kwargs):
+        if name == "obliteratus.adaptive_defaults":
+            raise ImportError("optional module unavailable")
+        return real_import(name, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "__import__", reject_adaptive_defaults)
+    returned, recommendation = profiles.enhance_profile_with_telemetry(profile)
+    assert returned is profile
+    assert recommendation is None
+    assert profile.recommended_method == original_method
+
+
+def test_telemetry_provider_failure_is_non_destructive(monkeypatch):
+    profile = _telemetry_profile()
+
+    def fail_recommendation(**_kwargs):
+        raise RuntimeError("offline")
+
+    monkeypatch.setattr(
+        "obliteratus.adaptive_defaults.get_adaptive_recommendation",
+        fail_recommendation,
+    )
+    returned, recommendation = profiles.enhance_profile_with_telemetry(profile)
+    assert returned is profile
+    assert recommendation is None
+    assert "Telemetry override" not in profile.profile_description
+
+
+def test_none_and_low_confidence_telemetry_do_not_override_research(monkeypatch):
+    for confidence in ("none", "low"):
+        profile = _telemetry_profile()
+        original_method = profile.recommended_method
+        recommendation = SimpleNamespace(
+            confidence=confidence,
+            recommended_method="surgical",
+            method_overrides={"n_directions": 99},
+            n_records=1,
+        )
+        monkeypatch.setattr(
+            "obliteratus.adaptive_defaults.get_adaptive_recommendation",
+            lambda **_kwargs: recommendation,
+        )
+        returned, observed = profiles.enhance_profile_with_telemetry(profile)
+        assert returned is profile
+        assert observed is recommendation
+        assert profile.recommended_method == original_method
+        assert profile.method_overrides.get("n_directions") != 99
+
+
+def test_medium_confidence_telemetry_overlays_method_and_parameters(monkeypatch):
+    profile = _telemetry_profile()
+    recommendation = SimpleNamespace(
+        confidence="medium",
+        recommended_method="surgical",
+        method_overrides={"n_directions": 9, "regularization": 0.07},
+        n_records=42,
+    )
+    calls = []
+
+    def recommend(**kwargs):
+        calls.append(kwargs)
+        return recommendation
+
+    monkeypatch.setattr(
+        "obliteratus.adaptive_defaults.get_adaptive_recommendation",
+        recommend,
+    )
+    returned, observed = profiles.enhance_profile_with_telemetry(profile)
+    assert returned is profile
+    assert observed is recommendation
+    assert calls == [
+        {
+            "arch_class": "dense",
+            "reasoning_class": "standard",
+            "total_params_b": profile.total_params_b,
+            "model_name": profile.model_name,
+        },
+    ]
+    assert profile.recommended_method == "surgical"
+    assert profile.method_overrides["n_directions"] == 9
+    assert profile.method_overrides["regularization"] == 0.07
+    assert "Telemetry override (medium confidence)" in profile.profile_description
+    assert "42 runs" in profile.profile_description
+
+
+def test_high_confidence_empty_recommendation_is_a_noop(monkeypatch):
+    profile = _telemetry_profile()
+    before = (
+        profile.recommended_method,
+        dict(profile.method_overrides),
+        profile.profile_description,
+    )
+    recommendation = SimpleNamespace(
+        confidence="high",
+        recommended_method=None,
+        method_overrides={},
+        n_records=100,
+    )
+    monkeypatch.setattr(
+        "obliteratus.adaptive_defaults.get_adaptive_recommendation",
+        lambda **_kwargs: recommendation,
+    )
+    returned, observed = profiles.enhance_profile_with_telemetry(profile)
+    assert returned is profile
+    assert observed is recommendation
+    assert (
+        profile.recommended_method,
+        profile.method_overrides,
+        profile.profile_description,
+    ) == before
@@ -163,6 +163,29 @@ def test_cache_cleanup_paths_are_best_effort(monkeypatch):
    mps_sync.assert_called_once_with()


+def test_cuda_cleanup_tolerates_every_recovery_failure(monkeypatch):
+    monkeypatch.setattr(device, "is_cuda", lambda: True)
+    monkeypatch.setattr(device.torch.cuda, "empty_cache", Mock(side_effect=RuntimeError("busy")))
+    monkeypatch.setattr(device.torch.cuda, "synchronize", Mock(side_effect=RuntimeError("lost")))
+    monkeypatch.setattr(
+        device.torch.cuda,
+        "reset_peak_memory_stats",
+        Mock(side_effect=RuntimeError("lost")),
+    )
+    monkeypatch.setattr(device.gc, "collect", Mock())
+    device.free_gpu_memory()
+
+
+@pytest.mark.parametrize(
+    ("version", "expected"),
+    [("2.2.9", False), ("2.3.0", True), ("2.10.1", True)],
+)
+def test_mps_bfloat16_version_boundary(monkeypatch, version, expected):
+    monkeypatch.setattr(device, "is_cuda", lambda: False)
+    monkeypatch.setattr(device.torch, "__version__", version)
+    assert device.supports_bfloat16("mps") is expected
+
+
 def test_seed_dtype_and_capability_contracts(monkeypatch):
    manual_seed = Mock()
    cuda_seed = Mock()
@@ -198,6 +221,11 @@ def test_svd_dtype_and_oom_matching():
    assert device.is_oom_error(RuntimeError("other")) is False


+def test_mps_svd_dtype_uses_float32_without_requiring_mps_hardware():
+    tensor = SimpleNamespace(device=SimpleNamespace(type="mps"), dtype=torch.float64)
+    assert device.safe_svd_dtype(tensor) is torch.float32
+
+
 def test_configure_cuda_allocator(monkeypatch):
    monkeypatch.delenv("PYTORCH_CUDA_ALLOC_CONF", raising=False)
    monkeypatch.setattr(device, "is_cuda", lambda: True)
@@ -454,3 +454,106 @@ def test_automatic_snapshot_memory_policy(
    handle = loader.load_model("x")
    assert snapshot.call_count == snapshots
    handle.cleanup()
+
+
+def test_classification_load_sets_requested_label_count(loader_boundary):
+    handle = loader.load_model(
+        "x",
+        task="classification",
+        num_labels=7,
+        skip_snapshot=True,
+    )
+    assert loader_boundary.config.num_labels == 7
+    assert loader_boundary.model_class.from_pretrained.call_args.kwargs["config"] is loader_boundary.config
+    handle.cleanup()
+
+
+def test_non_gated_model_oserror_is_preserved(loader_boundary):
+    loader_boundary.model_class.from_pretrained.side_effect = OSError("offline weight cache miss")
+    with pytest.raises(OSError, match="offline weight cache miss"):
+        loader.load_model("x", local_files_only=True)
+
+
+def test_cached_path_compatibility_resolves_local_hub_and_fallback(monkeypatch, tmp_path):
+    local = tmp_path / "weights.bin"
+    local.write_bytes(b"weights")
+    assert loader._cached_path_shim(local) == str(local)
+
+    download = Mock(return_value="/cache/config.json")
+    monkeypatch.setattr("huggingface_hub.hf_hub_download", download)
+    assert loader._cached_path_shim("owner/model/config.json", cache_dir="/cache") == "/cache/config.json"
+    download.assert_called_once_with(
+        repo_id="owner/model",
+        filename="config.json",
+        cache_dir="/cache",
+    )
+
+    download.side_effect = RuntimeError("network unavailable")
+    assert loader._cached_path_shim("owner/model/config.json") == "owner/model/config.json"
+    download.reset_mock(side_effect=True)
+    assert loader._cached_path_shim("single-name") == "single-name"
+    download.assert_not_called()
+
+
+def test_deferred_transformers_shims_are_additive_and_idempotent(monkeypatch):
+    fake_transformers = SimpleNamespace(_objects={})
+    monkeypatch.setitem(loader._sys.modules, "transformers", fake_transformers)
+    monkeypatch.setattr(loader, "_DEFERRED_SHIMS_APPLIED", False)
+    loader._apply_deferred_shims()
+
+    assert fake_transformers.LogitsWarper is fake_transformers._objects["LogitsWarper"]
+    assert fake_transformers.is_tf_available() is False
+    assert fake_transformers.is_flax_available() is False
+    assert fake_transformers.is_safetensors_available() is True
+    before = dict(fake_transformers.__dict__)
+    loader._apply_deferred_shims()
+    assert fake_transformers.__dict__ == before
+
+
+def test_deferred_transformers_shims_tolerate_missing_module(monkeypatch):
+    monkeypatch.delitem(loader._sys.modules, "transformers", raising=False)
+    monkeypatch.setattr(loader, "_DEFERRED_SHIMS_APPLIED", False)
+    loader._apply_deferred_shims()
+    assert loader._DEFERRED_SHIMS_APPLIED is True
+
+
+def test_deferred_transformers_shims_isolate_lazy_module_failures(monkeypatch):
+    class BrokenLazyModule:
+        def __getattribute__(self, name):
+            if name in {"LogitsWarper", "is_tf_available"}:
+                raise RuntimeError("lazy import failed")
+            return object.__getattribute__(self, name)
+
+    monkeypatch.setitem(loader._sys.modules, "transformers", BrokenLazyModule())
+    monkeypatch.setattr(loader, "_DEFERRED_SHIMS_APPLIED", False)
+    loader._apply_deferred_shims()
+    assert loader._DEFERRED_SHIMS_APPLIED is True
+
+
+def test_image_text_input_compatibility_shim_preserves_argument_order():
+    images = object()
+    text = object()
+    assert loader._validate_images_text_input_order(
+        images=images,
+        text=text,
+        ignored="compatibility",
+    ) == (images, text)
+
+
+def test_cleanup_is_best_effort_when_directory_removal_fails(monkeypatch, tmp_path):
+    offload = tmp_path / "offload"
+    offload.mkdir()
+    handle = loader.ModelHandle(
+        _model(),
+        SimpleNamespace(),
+        _config(),
+        "x",
+        "causal_lm",
+        _offload_dir=str(offload),
+        _owns_offload_dir=True,
+    )
+    monkeypatch.setattr("shutil.rmtree", Mock(side_effect=OSError("busy")))
+    handle.cleanup()
+    assert handle._offload_dir is None
+    assert handle._owns_offload_dir is False
+    assert offload.exists()
@@ -17,6 +17,8 @@ def test_mutation_campaign_preloads_native_modules_before_covered_line_discovery
    workflow = Path(".github/workflows/ci.yml").read_text()

    assert "mutate_only_covered_lines = true" in mutmut_config
+    assert '"obliteratus/runtime_contracts.py"' in mutmut_config
+    assert '"tests/test_runtime_contracts.py"' in mutmut_config
    assert "import torch, yaml; from mutmut.__main__ import cli; cli()" in workflow


@@ -0,0 +1,232 @@
+"""Mutation-friendly contracts for model, architecture, and device decisions."""
+
+from __future__ import annotations
+
+import pytest
+from hypothesis import given, strategies as st
+
+from obliteratus.runtime_contracts import (
+    classify_architecture_size,
+    effective_model_memory_gb,
+    quantized_model_fits_gpu,
+    should_snapshot_model,
+    supports_bfloat16_target,
+    validate_model_load_request,
+)
+
+
+@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:
+        validate_model_load_request(model_name, "causal_lm", None, "float32")
+    assert str(error.value) == "model_name must be a non-empty HuggingFace identifier or local path"
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "message"),
+    [
+        ("task", "embedding", "Unknown task 'embedding'. Choose from ['causal_lm', 'classification']"),
+        ("quantization", "3bit", "Unknown quantization '3bit'. Choose None, '4bit', or '8bit'"),
+        ("dtype", "float128", "Unknown dtype 'float128'. Choose from ['float32', 'float16', 'bfloat16']"),
+    ],
+)
+def test_invalid_loader_enums_are_rejected(field, value, message):
+    request = {
+        "model_name": "local/model",
+        "task": "causal_lm",
+        "quantization": None,
+        "dtype": "float32",
+    }
+    request[field] = value
+    with pytest.raises(ValueError) as error:
+        validate_model_load_request(**request)
+    assert str(error.value) == message
+
+
+@pytest.mark.parametrize("task", ["causal_lm", "classification"])
+@pytest.mark.parametrize("quantization", [None, "4bit", "8bit"])
+@pytest.mark.parametrize("dtype", ["float32", "float16", "bfloat16"])
+def test_every_documented_loader_enum_combination_is_valid(task, quantization, dtype):
+    validate_model_load_request("local/model", task, quantization, dtype)
+
+
+def test_loader_task_validation_can_follow_the_provider_registry():
+    validate_model_load_request(
+        "local/model",
+        "custom_task",
+        None,
+        "float32",
+        valid_tasks={"custom_task": object()},
+    )
+
+
+@given(st.floats(min_value=0, max_value=10_000, allow_nan=False, allow_infinity=False))
+def test_effective_memory_respects_quantization_ratios(estimate_gb):
+    assert effective_model_memory_gb(estimate_gb, None) == estimate_gb
+    assert effective_model_memory_gb(estimate_gb, "4bit") == estimate_gb / 4
+    assert effective_model_memory_gb(estimate_gb, "8bit") == estimate_gb / 2
+
+
+@pytest.mark.parametrize(
+    ("estimate_gb", "quantization", "available_gb", "expected"),
+    [
+        (44.0, "4bit", 16.0, True),
+        (44.8, "4bit", 16.0, False),
+        (22.0, "8bit", 16.0, True),
+        (22.4, "8bit", 16.0, False),
+        (2.0, "4bit", 1.0, True),
+        (1.0, None, 16.0, False),
+        (0.0, "4bit", 16.0, False),
+        (1.0, "4bit", 0.0, False),
+    ],
+)
+def test_quantized_fit_requires_positive_memory_and_thirty_percent_headroom(
+    estimate_gb,
+    quantization,
+    available_gb,
+    expected,
+):
+    assert quantized_model_fits_gpu(estimate_gb, quantization, available_gb) is expected
+
+
+def _snapshot(**overrides):
+    values = {
+        "skip_snapshot": None,
+        "initial_gpu_free_gb": 0.0,
+        "remaining_gpu_free_gb": 0.0,
+        "has_native_quantization": False,
+        "estimate_gb": 0.0,
+        "quantization": None,
+    }
+    values.update(overrides)
+    return should_snapshot_model(**values)
+
+
+def test_explicit_snapshot_choice_overrides_every_memory_signal():
+    constrained = {
+        "initial_gpu_free_gb": 1.0,
+        "remaining_gpu_free_gb": 0.0,
+        "has_native_quantization": True,
+        "estimate_gb": 1000.0,
+    }
+    assert _snapshot(skip_snapshot=True, **constrained) is False
+    assert _snapshot(skip_snapshot=False, **constrained) is True
+
+
+@pytest.mark.parametrize(
+    ("remaining_gb", "expected"),
+    [(3.99, False), (4.0, True), (4.01, True)],
+)
+def test_native_quantization_snapshot_boundary_is_forty_percent(remaining_gb, expected):
+    assert _snapshot(
+        initial_gpu_free_gb=10.0,
+        remaining_gpu_free_gb=remaining_gb,
+        has_native_quantization=True,
+    ) is expected
+
+
+def test_native_snapshot_policy_uses_any_positive_gpu_signal():
+    assert _snapshot(
+        initial_gpu_free_gb=0.5,
+        remaining_gpu_free_gb=0.1,
+        has_native_quantization=True,
+    ) is False
+    assert _snapshot(
+        initial_gpu_free_gb=0.0,
+        remaining_gpu_free_gb=-1.0,
+        has_native_quantization=True,
+    ) is True
+
+
+@pytest.mark.parametrize(
+    ("estimate_gb", "quantization", "expected"),
+    [(5.0, None, True), (5.01, None, False), (20.0, "4bit", True), (20.04, "4bit", False)],
+)
+def test_estimated_snapshot_boundary_is_half_of_free_memory(
+    estimate_gb,
+    quantization,
+    expected,
+):
+    assert _snapshot(
+        initial_gpu_free_gb=10.0,
+        estimate_gb=estimate_gb,
+        quantization=quantization,
+    ) is expected
+
+
+def test_estimated_snapshot_policy_uses_any_positive_gpu_signal():
+    assert _snapshot(initial_gpu_free_gb=0.5, estimate_gb=0.3) is False
+    assert _snapshot(initial_gpu_free_gb=0.0, estimate_gb=1.0) is True
+
+
+def test_snapshot_defaults_to_enabled_without_a_positive_gpu_signal():
+    assert _snapshot() is True
+    assert _snapshot(initial_gpu_free_gb=-1.0, estimate_gb=100.0) is True
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "expected"),
+    [
+        ({"is_moe": False, "total_params_b": 999, "num_experts": 999}, "dense"),
+        ({"is_moe": True, "total_params_b": 99.99, "num_experts": 999}, "small_moe"),
+        ({"is_moe": True, "total_params_b": 0.5, "num_experts": 999}, "small_moe"),
+        ({"is_moe": True, "total_params_b": 100.0, "num_experts": 1}, "large_moe"),
+        ({"is_moe": True, "total_params_b": 0.0, "num_experts": 1}, "small_moe"),
+        ({"is_moe": True, "total_params_b": 0.0, "num_experts": 16}, "small_moe"),
+        ({"is_moe": True, "total_params_b": 0.0, "num_experts": 17}, "large_moe"),
+    ],
+)
+def test_architecture_classification_precedence_and_boundaries(kwargs, expected):
+    assert classify_architecture_size(
+        model_name="custom/model",
+        large_moe_name_patterns=("large-model",),
+        **kwargs,
+    ) == expected
+
+
+def test_architecture_name_fallback_is_case_insensitive_and_conservative():
+    common = {
+        "is_moe": True,
+        "total_params_b": 0.0,
+        "num_experts": 0,
+        "large_moe_name_patterns": ("giant-moe",),
+    }
+    assert classify_architecture_size(model_name="ORG/GIANT-MOE-V1", **common) == "large_moe"
+    assert classify_architecture_size(model_name="org/unknown-moe", **common) == "small_moe"
+
+
+def test_single_expert_precedes_the_large_name_fallback():
+    assert classify_architecture_size(
+        is_moe=True,
+        total_params_b=0.0,
+        num_experts=1,
+        model_name="org/giant-moe-v1",
+        large_moe_name_patterns=("giant-moe",),
+    ) == "small_moe"
+
+
+@pytest.mark.parametrize(
+    ("target", "cuda_available", "cuda_major", "version", "expected"),
+    [
+        ("cuda", True, 8, "2.0.0", True),
+        ("cuda:1", True, 7, "2.0.0", False),
+        ("cuda", False, None, "2.0.0", False),
+        ("mps", False, None, "2.2.9", False),
+        ("mps", False, None, "2.3.0", True),
+        ("mps", False, None, "2.10.0", True),
+        ("cpu", True, 1, "1.0.0", True),
+    ],
+)
+def test_bfloat16_capability_boundaries(
+    target,
+    cuda_available,
+    cuda_major,
+    version,
+    expected,
+):
+    assert supports_bfloat16_target(
+        target,
+        cuda_available=cuda_available,
+        cuda_major=cuda_major,
+        torch_version=version,
+    ) is expected