diff --git a/.aiwg/aiwg.config b/.aiwg/aiwg.config index c587e93..423ef1e 100644 --- a/.aiwg/aiwg.config +++ b/.aiwg/aiwg.config @@ -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, diff --git a/ci/test-quality-policy.json b/ci/test-quality-policy.json index 9d6b3d9..6def675 100644 --- a/ci/test-quality-policy.json +++ b/ci/test-quality-policy.json @@ -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", diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json index 19a9f4a..3a1d82e 100644 --- a/ci/test-risk-map.json +++ b/ci/test-risk-map.json @@ -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", diff --git a/obliteratus/architecture_profiles.py b/obliteratus/architecture_profiles.py index 716bb0f..d835dba 100644 --- a/obliteratus/architecture_profiles.py +++ b/obliteratus/architecture_profiles.py @@ -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 diff --git a/obliteratus/device.py b/obliteratus/device.py index 44a6baa..8ea8c11 100644 --- a/obliteratus/device.py +++ b/obliteratus/device.py @@ -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: diff --git a/obliteratus/models/loader.py b/obliteratus/models/loader.py index 504c3bc..48ccbd5 100644 --- a/obliteratus/models/loader.py +++ b/obliteratus/models/loader.py @@ -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 diff --git a/obliteratus/runtime_contracts.py b/obliteratus/runtime_contracts.py new file mode 100644 index 0000000..0b86088 --- /dev/null +++ b/obliteratus/runtime_contracts.py @@ -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 diff --git a/pyproject.toml b/pyproject.toml index 48cd2ab..3c78897 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/tests/test_architecture_profiles.py b/tests/test_architecture_profiles.py index 00d1c13..c8c5cf5 100644 --- a/tests/test_architecture_profiles.py +++ b/tests/test_architecture_profiles.py @@ -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 diff --git a/tests/test_device_boundaries.py b/tests/test_device_boundaries.py index e4ea711..3492d4a 100644 --- a/tests/test_device_boundaries.py +++ b/tests/test_device_boundaries.py @@ -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) diff --git a/tests/test_loader_boundaries.py b/tests/test_loader_boundaries.py index 752d9ec..d578c1a 100644 --- a/tests/test_loader_boundaries.py +++ b/tests/test_loader_boundaries.py @@ -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() diff --git a/tests/test_quality_policy.py b/tests/test_quality_policy.py index 8826d38..9c44557 100644 --- a/tests/test_quality_policy.py +++ b/tests/test_quality_policy.py @@ -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 diff --git a/tests/test_runtime_contracts.py b/tests/test_runtime_contracts.py new file mode 100644 index 0000000..9e9b2d4 --- /dev/null +++ b/tests/test_runtime_contracts.py @@ -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