diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json index d34ec30..ef0aa2c 100644 --- a/ci/test-risk-map.json +++ b/ci/test-risk-map.json @@ -43,7 +43,8 @@ "obliteratus/lora_ablation.py", "obliteratus/runner.py", "obliteratus/sweep.py", - "obliteratus/tourney.py" + "obliteratus/tourney.py", + "obliteratus/tourney_contracts.py" ], "required_tests": [ "tests/test_abliterate.py", @@ -55,6 +56,7 @@ "tests/test_offline_integration.py", "tests/test_runner_boundaries.py", "tests/test_sweep_contracts.py", + "tests/test_tourney_contracts.py", "tests/test_checkpoint_atomicity.py", "tests/test_persistence_contracts.py", "tests/test_persistence_pipeline.py" @@ -111,11 +113,14 @@ "required_tests": [ "tests/test_cli.py", "tests/test_cli_boundaries.py", + "tests/test_interactive_contracts.py", + "tests/test_local_ui_contracts.py", "tests/test_local_ui_portability.py", "tests/test_bestiary_contracts.py", "tests/test_models_client_contracts.py", "tests/test_remote_boundaries.py", "tests/test_remote_contracts.py", + "tests/test_ui_watchtower_contracts.py", "tests/test_watchtower_contracts.py", "tests/conditional/test_network_services.py", "tests/conditional/test_operator_ui.py", @@ -466,11 +471,18 @@ }, { "path": "obliteratus/tourney.py", - "risk_class": "conditional-runtime", + "risk_class": "mixed-runtime", "risk": "multi-model tournament mutation, comparison, and optional publication", - "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_external_evaluation_runtime.py"], + "required_tests": ["tests/test_tourney_contracts.py", "tests/test_module_imports.py", "tests/conditional/test_external_evaluation_runtime.py"], "conditional_gates": ["external-evaluation"] }, + { + "path": "obliteratus/tourney_contracts.py", + "risk_class": "cpu-contract", + "risk": "checkpoint document parsing and version validation", + "required_tests": ["tests/test_tourney_contracts.py"], + "conditional_gates": [] + }, { "path": "obliteratus/bestiary_sync.py", "risk_class": "conditional-runtime", @@ -501,9 +513,9 @@ }, { "path": "obliteratus/interactive.py", - "risk_class": "conditional-runtime", + "risk_class": "mixed-runtime", "risk": "interactive terminal prompts and operator decision flow", - "required_tests": ["tests/test_cli.py", "tests/conditional/test_operator_ui.py"], + "required_tests": ["tests/test_cli.py", "tests/test_interactive_contracts.py", "tests/conditional/test_operator_ui.py"], "conditional_gates": ["operator-ui"] }, { @@ -512,6 +524,7 @@ "risk": "Gradio construction, launch configuration, authentication, and signals", "required_tests": [ "tests/test_cli.py", + "tests/test_local_ui_contracts.py", "tests/test_local_ui_portability.py", "tests/conditional/test_operator_ui.py" ], @@ -519,9 +532,9 @@ }, { "path": "obliteratus/ui_watchtower.py", - "risk_class": "conditional-runtime", + "risk_class": "mixed-runtime", "risk": "service-backed UI tabs and scheduler controls", - "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_operator_ui.py"], + "required_tests": ["tests/test_ui_watchtower_contracts.py", "tests/test_module_imports.py", "tests/conditional/test_operator_ui.py"], "conditional_gates": ["operator-ui"] }, { diff --git a/obliteratus/interactive.py b/obliteratus/interactive.py index e30a755..3268da0 100644 --- a/obliteratus/interactive.py +++ b/obliteratus/interactive.py @@ -266,7 +266,7 @@ def run_interactive(): task="causal_lm", dtype=dtype, device=device, - trust_remote_code=True, + trust_remote_code=False, ) dataset_cfg = DatasetConfig( @@ -308,7 +308,7 @@ def run_interactive(): # Handle quantization by modifying the loader if quantization: - _run_quantized(config, quantization) + return _run_quantized(config, quantization) else: from obliteratus.runner import run_study return run_study(config) diff --git a/obliteratus/local_ui.py b/obliteratus/local_ui.py index c9d950d..b7fec89 100644 --- a/obliteratus/local_ui.py +++ b/obliteratus/local_ui.py @@ -80,7 +80,7 @@ def _detect_gpu() -> list[dict]: "compute": "mps", } ) - except ImportError: + except (AttributeError, ImportError, OSError, RuntimeError, ValueError): pass return gpus diff --git a/obliteratus/tourney.py b/obliteratus/tourney.py index aca01e1..1d8913e 100644 --- a/obliteratus/tourney.py +++ b/obliteratus/tourney.py @@ -26,6 +26,8 @@ from datetime import datetime from pathlib import Path from typing import Any, Callable +from obliteratus.tourney_contracts import parse_checkpoint_document + # --------------------------------------------------------------------------- # All tournament-eligible methods. # @@ -265,11 +267,8 @@ def _load_checkpoint(output_dir: Path) -> dict | None: if not path.exists(): return None try: - data = json.loads(path.read_text()) - if data.get("version") != 1: - return None - return data - except (json.JSONDecodeError, KeyError): + return parse_checkpoint_document(path.read_text()) + except (OSError, UnicodeError): return None @@ -331,6 +330,8 @@ def _restore_rounds(checkpoint: dict) -> tuple[TourneyResult, list[Contender], l time_s=c_data.get("time_s", 0.0), error=c_data.get("error"), round_eliminated=c_data.get("round_eliminated", 0), + direction_method=c_data.get("direction_method", ""), + spectral_cert=c_data.get("spectral_cert", ""), )) remaining = ir.get("remaining_methods", []) diff --git a/obliteratus/tourney_contracts.py b/obliteratus/tourney_contracts.py new file mode 100644 index 0000000..fc5bac9 --- /dev/null +++ b/obliteratus/tourney_contracts.py @@ -0,0 +1,17 @@ +"""Pure persistence contracts for tournament orchestration.""" + +from __future__ import annotations + +import json +from typing import Any + + +def parse_checkpoint_document(raw: str) -> dict[str, Any] | None: + """Return a supported checkpoint object, or ``None`` for invalid input.""" + try: + data = json.loads(raw) + except json.JSONDecodeError: + return None + if not isinstance(data, dict) or data.get("version") != 1: + return None + return data diff --git a/obliteratus/ui_watchtower.py b/obliteratus/ui_watchtower.py index 97b3811..1dffb05 100644 --- a/obliteratus/ui_watchtower.py +++ b/obliteratus/ui_watchtower.py @@ -16,9 +16,20 @@ Usage in app.py: from __future__ import annotations from datetime import datetime +from html import escape from pathlib import Path -import gradio as gr +try: + import gradio as gr +except ImportError: # Pure handlers remain available without the optional UI extra. + gr = None + + +def _component_update(**kwargs): + """Build a Gradio update when available, or its plain mapping contract.""" + if gr is None: + return kwargs + return gr.update(**kwargs) # ── Lazy imports to avoid circular deps and slow startup ────────────── @@ -67,7 +78,7 @@ def _run_one_click( model_id: str, max_iterations: int, target_refusal_pct: float, - progress=gr.Progress(), + progress=None, ): """Generator that runs auto-obliteration, yielding (status, log, metrics). @@ -78,7 +89,7 @@ def _run_one_click( "⚠️ Please enter or select a model ID.", "Error: No model selected.", "", - gr.update(interactive=False), + _component_update(interactive=False), ) return @@ -92,7 +103,7 @@ def _run_one_click( f"⚠️ Import error: {e}", f"Failed to import AutoObliterator: {e}", "", - gr.update(interactive=False), + _component_update(interactive=False), ) return @@ -118,7 +129,7 @@ def _run_one_click( status, log, metrics, - gr.update( + _component_update( interactive=has_output, visible=has_output, ), @@ -131,21 +142,23 @@ def _run_one_click( f"⚠️ Error: {e}", f"Fatal error during auto-obliteration:\n{e}", "", - gr.update(interactive=False), + _component_update(interactive=False), ) return # Final yield with download button if final_result and final_result.final_output_dir: output_path = final_result.final_output_dir + refusal_rate = final_result.final_refusal_rate + refusal_display = "unknown" if refusal_rate is None else refusal_rate yield ( "✅ Complete!" if final_result.success else "⚠️ Complete (target not fully met)", f"Auto-obliteration finished.\n" f"Output saved to: {output_path}\n\n" - f"Final refusal rate: {final_result.final_refusal_rate or 'unknown'}\n" + f"Final refusal rate: {refusal_display}\n" f"Total time: {final_result.total_time_seconds}s", obliterator._format_metrics(), - gr.update(interactive=True, visible=True), + _component_update(interactive=True, visible=True), ) @@ -158,9 +171,23 @@ def _download_result(model_id: str): # Find the latest iteration output if base.exists(): - iter_dirs = sorted(base.glob("iter_*"), reverse=True) + base_resolved = base.resolve() + + def iteration_number(path: Path) -> int: + try: + return int(path.name.removeprefix("iter_")) + except ValueError: + return -1 + + iter_dirs = sorted(base.glob("iter_*"), key=iteration_number, reverse=True) for d in iter_dirs: - if (d / "config.json").exists(): + try: + resolved = d.resolve(strict=True) + except OSError: + continue + if not resolved.is_relative_to(base_resolved): + continue + if d.is_dir() and (d / "config.json").is_file(): return str(d) return None @@ -185,10 +212,11 @@ def _scan_now(): return status_html, table_data, log_text except Exception as e: + message = escape(str(e)) return ( - f"
Error: {e}
", + f"
Error: {message}
", [], - f"Scan failed: {e}", + f"Scan failed: {message}", ) @@ -278,9 +306,12 @@ def _format_status_html(stats: dict, new_count: int = 0) -> str: except Exception: pass - obliterated = by_status.get("obliterated", 0) - queued = by_status.get("queued", 0) - new = by_status.get("new", 0) + total = escape(str(total)) + scan_count = escape(str(scan_count)) + last_scan = escape(str(last_scan)) + obliterated = escape(str(by_status.get("obliterated", 0))) + queued = escape(str(by_status.get("queued", 0))) + new = escape(str(by_status.get("new", 0))) new_badge = "" if new_count > 0: @@ -334,7 +365,7 @@ def _format_status_html(stats: dict, new_count: int = 0) -> str: def _refresh_one_click_dropdown(): """Refresh the one-click dropdown with latest trending models.""" choices = _get_trending_choices() - return gr.update(choices=choices, value=choices[0] if choices else "") + return _component_update(choices=choices, value=choices[0] if choices else "") # ── Build tabs ──────────────────────────────────────────────────────── @@ -346,6 +377,9 @@ def build_watchtower_tabs(): Creates and wires up all UI components. """ + if gr is None: + raise ImportError("Gradio is required to construct Watchtower tabs; install the spaces extra") + # ══════════════════════════════════════════════════════════════════ # ⚡ ONE-CLICK TAB # ══════════════════════════════════════════════════════════════════ diff --git a/pyproject.toml b/pyproject.toml index cdf3039..c044473 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -123,6 +123,7 @@ only_mutate = [ "obliteratus/persistence_contracts.py", "obliteratus/remote_contracts.py", "obliteratus/service_contracts.py", + "obliteratus/tourney_contracts.py", "obliteratus/evaluation/lm_eval_integration.py", "scripts/check_coverage_thresholds.py", ] @@ -132,6 +133,7 @@ required_mutation_targets = [ "obliteratus/persistence_contracts.py", "obliteratus/runtime_contracts.py", "obliteratus/service_contracts.py", + "obliteratus/tourney_contracts.py", ] pytest_add_cli_args = ["--no-cov", "-q"] pytest_add_cli_args_test_selection = [ @@ -148,6 +150,7 @@ pytest_add_cli_args_test_selection = [ "tests/test_persistence_pipeline.py", "tests/test_remote_contracts.py", "tests/test_runtime_contracts.py", + "tests/test_tourney_contracts.py", "tests/test_watchtower_contracts.py", "tests/test_whitened_svd_oracles.py", ] diff --git a/scripts/run_repeat_gate.py b/scripts/run_repeat_gate.py index 628421f..9c7060d 100644 --- a/scripts/run_repeat_gate.py +++ b/scripts/run_repeat_gate.py @@ -25,6 +25,8 @@ DEFAULT_TESTS = ( "tests/test_evaluation_reporting_contracts.py", "tests/test_lm_eval_reporting_contracts.py", "tests/test_informed_pipeline_contracts.py", + "tests/test_interactive_contracts.py", + "tests/test_local_ui_contracts.py", "tests/test_model_profile_contracts.py", "tests/test_models_client_contracts.py", "tests/test_numerical_contracts.py", @@ -40,6 +42,8 @@ DEFAULT_TESTS = ( "tests/test_strategy_navigation_contracts.py", "tests/test_sweep_contracts.py", "tests/test_telemetry_failure_contracts.py", + "tests/test_tourney_contracts.py", + "tests/test_ui_watchtower_contracts.py", "tests/test_watchtower_contracts.py", ) HASH_SEEDS = ("0", "1", "8675309") diff --git a/tests/test_interactive_contracts.py b/tests/test_interactive_contracts.py new file mode 100644 index 0000000..7a5fa2e --- /dev/null +++ b/tests/test_interactive_contracts.py @@ -0,0 +1,212 @@ +"""Scripted contracts for the guided interactive workflow.""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from obliteratus import interactive +from obliteratus.presets import ModelPreset + + +def _preset(*, quantization: str | None = None) -> ModelPreset: + return ModelPreset( + name="Safe model", + hf_id="org/safe-model", + description="fixture", + tier="tiny", + params="1B", + recommended_dtype="float32", + recommended_quantization=quantization, + ) + + +@pytest.mark.parametrize( + ("vram_gb", "expected"), + [(4, "small"), (8, "medium"), (19.9, "medium"), (20, "large")], +) +def test_compute_tier_maps_cuda_memory_boundaries(monkeypatch, vram_gb, expected): + from obliteratus import device + + monkeypatch.setattr(device, "is_cuda", lambda: True) + monkeypatch.setattr(device, "is_mps", lambda: False) + torch = SimpleNamespace( + cuda=SimpleNamespace( + get_device_properties=lambda _index: SimpleNamespace( + total_memory=vram_gb * 1024**3, + ) + ) + ) + monkeypatch.setitem(__import__("sys").modules, "torch", torch) + + assert interactive._detect_compute_tier() == expected + + +@pytest.mark.parametrize( + ("memory_gb", "expected"), + [(16, "small"), (24, "medium")], +) +def test_compute_tier_maps_mps_unified_memory(monkeypatch, memory_gb, expected): + from obliteratus import device + + monkeypatch.setattr(device, "is_cuda", lambda: False) + monkeypatch.setattr(device, "is_mps", lambda: True) + monkeypatch.setattr( + device, + "get_memory_info", + lambda: SimpleNamespace(total_gb=memory_gb), + ) + + assert interactive._detect_compute_tier() == expected + + +def test_compute_tier_falls_back_to_cpu_when_device_probe_is_unavailable(monkeypatch): + from obliteratus import device + + monkeypatch.setattr(device, "is_cuda", Mock(side_effect=ImportError("torch"))) + + assert interactive._detect_compute_tier() == "tiny" + + +def test_custom_model_selection_preserves_safe_tier_defaults(monkeypatch): + monkeypatch.setattr(interactive, "get_presets_by_tier", lambda _tier: [_preset()]) + monkeypatch.setattr(interactive.IntPrompt, "ask", lambda *_args, **_kwargs: 0) + monkeypatch.setattr( + interactive.Prompt, + "ask", + lambda *_args, **_kwargs: "org/custom-model", + ) + + selected = interactive._pick_model("tiny") + + assert selected.hf_id == "org/custom-model" + assert selected.recommended_dtype == "float32" + assert selected.recommended_quantization is None + + +def test_invalid_model_selection_falls_back_to_first_recommendation(monkeypatch): + first = _preset() + monkeypatch.setattr(interactive, "get_presets_by_tier", lambda _tier: [first]) + monkeypatch.setattr(interactive.IntPrompt, "ask", lambda *_args, **_kwargs: 99) + + assert interactive._pick_model("tiny") is first + + +def test_custom_strategy_and_sample_mappings_are_exact(monkeypatch): + answers = iter(["5", "3"]) + monkeypatch.setattr( + interactive.Prompt, + "ask", + lambda *_args, **_kwargs: next(answers), + ) + + strategies = interactive._pick_strategies() + + assert [item["name"] for item in strategies] == [ + "layer_removal", + "head_pruning", + "ffn_ablation", + "embedding_ablation", + ] + assert strategies[-1]["params"] == {"chunk_size": 48} + assert interactive._pick_sample_size() == 500 + + +def test_guided_run_builds_safe_config_and_returns_study_result(monkeypatch, tmp_path): + from obliteratus import device, runner + + preset = _preset() + study = SimpleNamespace( + name="Fast", + strategies=[{"name": "layer_removal", "params": {"limit": 1}}], + max_samples=7, + batch_size=2, + max_length=64, + ) + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(interactive, "_pick_compute_tier", lambda: "tiny") + monkeypatch.setattr(interactive, "_pick_model", lambda _tier: preset) + monkeypatch.setattr(interactive, "_pick_study_preset", lambda: study) + monkeypatch.setattr(device, "get_device", lambda: "cpu") + monkeypatch.setattr(interactive.Confirm, "ask", lambda *_args, **_kwargs: True) + run_study = Mock(return_value="study-result") + monkeypatch.setattr(runner, "run_study", run_study) + + assert interactive.run_interactive() == "study-result" + config = run_study.call_args.args[0] + assert config.model.name == "org/safe-model" + assert config.model.device == "cpu" + assert config.model.trust_remote_code is False + assert config.dataset.max_samples == 7 + assert config.output_dir == "results/org_safe-model" + + +def test_guided_run_cancellation_does_not_start_a_study(monkeypatch): + from obliteratus import device, runner + + monkeypatch.setattr(interactive, "_pick_compute_tier", lambda: "tiny") + monkeypatch.setattr(interactive, "_pick_model", lambda _tier: _preset()) + monkeypatch.setattr( + interactive, + "_pick_study_preset", + lambda: SimpleNamespace( + name="Fast", + strategies=[{"name": "layer_removal", "params": {}}], + max_samples=1, + batch_size=1, + max_length=8, + ), + ) + monkeypatch.setattr(device, "get_device", lambda: "cpu") + monkeypatch.setattr(interactive.Confirm, "ask", lambda *_args, **_kwargs: False) + run_study = Mock() + monkeypatch.setattr(runner, "run_study", run_study) + + assert interactive.run_interactive() is None + run_study.assert_not_called() + + +def test_quantized_guided_run_returns_quantized_result(monkeypatch): + from obliteratus import device + + monkeypatch.setattr(interactive, "_pick_compute_tier", lambda: "small") + monkeypatch.setattr( + interactive, + "_pick_model", + lambda _tier: _preset(quantization="4bit"), + ) + monkeypatch.setattr( + interactive, + "_pick_study_preset", + lambda: SimpleNamespace( + name="Fast", + strategies=[{"name": "layer_removal", "params": {}}], + max_samples=1, + batch_size=1, + max_length=8, + ), + ) + monkeypatch.setattr(device, "get_device", lambda: "cuda") + monkeypatch.setattr(interactive.Confirm, "ask", lambda *_args, **_kwargs: True) + run_quantized = Mock(return_value="quantized-result") + monkeypatch.setattr(interactive, "_run_quantized", run_quantized) + + assert interactive.run_interactive() == "quantized-result" + config, quantization = run_quantized.call_args.args + assert quantization == "4bit" + assert config.model.device == "auto" + + +def test_quantized_runner_sets_loader_contract_before_execution(monkeypatch): + from obliteratus import runner + + config = SimpleNamespace(model=SimpleNamespace(device="cuda", quantization=None)) + run_study = Mock(return_value="done") + monkeypatch.setattr(runner, "run_study", run_study) + + assert interactive._run_quantized(config, "8bit") == "done" + assert config.model.device == "auto" + assert config.model.quantization == "8bit" + run_study.assert_called_once_with(config) diff --git a/tests/test_local_ui_contracts.py b/tests/test_local_ui_contracts.py new file mode 100644 index 0000000..c03e9f6 --- /dev/null +++ b/tests/test_local_ui_contracts.py @@ -0,0 +1,132 @@ +"""Local launcher behavior, auth, dependency, and hardware contracts.""" + +from __future__ import annotations + +import sys +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from obliteratus import local_ui + + +@pytest.mark.parametrize( + ("vram", "expected"), + [([], "cpu"), ([0], "mps"), ([4], "small"), ([8], "medium"), ([24], "large"), ([80], "frontier")], +) +def test_compute_tier_boundaries(vram, expected): + gpus = [{"vram_gb": amount} for amount in vram] + assert local_ui._compute_tier(gpus) == expected + + +def test_gpu_detection_enumerates_cuda_devices(monkeypatch): + cuda = SimpleNamespace( + is_available=lambda: True, + device_count=lambda: 2, + get_device_properties=lambda index: SimpleNamespace( + name=f"GPU {index}", + total_memory=(index + 8) * 1024**3, + major=9, + minor=index, + ), + ) + torch = SimpleNamespace( + cuda=cuda, + backends=SimpleNamespace(mps=SimpleNamespace(is_available=lambda: False)), + ) + monkeypatch.setitem(sys.modules, "torch", torch) + + assert local_ui._detect_gpu() == [ + {"index": 0, "name": "GPU 0", "vram_gb": 8.0, "compute": "9.0"}, + {"index": 1, "name": "GPU 1", "vram_gb": 9.0, "compute": "9.1"}, + ] + + +def test_gpu_detection_degrades_to_cpu_when_runtime_probe_fails(monkeypatch): + torch = SimpleNamespace( + cuda=SimpleNamespace(is_available=Mock(side_effect=RuntimeError("driver unavailable"))), + ) + monkeypatch.setitem(sys.modules, "torch", torch) + + assert local_ui._detect_gpu() == [] + + +def test_launch_info_masks_password_and_reports_network_listener(monkeypatch): + console = Mock() + monkeypatch.setattr(local_ui, "console", console) + + local_ui._print_launch_info("0.0.0.0", 9000, True, ("operator", "secret")) + + rendered = console.print.call_args.args[0].renderable + assert "http://localhost:9000" in rendered + assert "http://:9000" in rendered + assert "operator:******" in rendered + assert "secret" not in rendered + + +def test_missing_gradio_exits_before_app_import(monkeypatch): + console = Mock() + monkeypatch.setattr(local_ui, "console", console) + monkeypatch.setitem(sys.modules, "gradio", None) + monkeypatch.delitem(sys.modules, "app", raising=False) + + with pytest.raises(SystemExit) as exc_info: + local_ui.launch_local_ui(quiet=True) + + assert exc_info.value.code == 1 + assert "Gradio is not installed" in console.print.call_args.args[0] + assert "app" not in sys.modules + + +def test_quiet_launch_skips_hardware_probes_and_forwards_server_contract(monkeypatch): + launch = Mock() + monkeypatch.setitem(sys.modules, "gradio", SimpleNamespace()) + monkeypatch.setitem(sys.modules, "app", SimpleNamespace(launch=launch)) + monkeypatch.setattr(local_ui, "_detect_gpu", Mock(side_effect=AssertionError("must not run"))) + monkeypatch.setattr(local_ui.sys, "path", ["/sentinel"]) + + kwargs = { + "host": "127.0.0.1", + "port": 9999, + "share": True, + "open_browser": False, + "auth": ("user", "password"), + "quiet": True, + } + local_ui.launch_local_ui(**kwargs) + local_ui.launch_local_ui(**kwargs) + + expected_root = str(local_ui.pathlib.Path(local_ui.__file__).resolve().parent.parent) + assert local_ui.sys.path == [expected_root, "/sentinel"] + assert launch.call_count == 2 + launch.assert_called_with( + server_name="127.0.0.1", + server_port=9999, + share=True, + inbrowser=False, + auth=("user", "password"), + quiet=True, + ) + + +def test_nonquiet_launch_reports_hardware_before_starting_app(monkeypatch): + launch = Mock() + gpus = [{"index": 0, "name": "GPU", "vram_gb": 24, "compute": "9.0"}] + monkeypatch.setitem(sys.modules, "gradio", SimpleNamespace()) + monkeypatch.setitem(sys.modules, "app", SimpleNamespace(launch=launch)) + monkeypatch.setattr(local_ui, "console", Mock()) + monkeypatch.setattr(local_ui, "_detect_gpu", Mock(return_value=gpus)) + system_info = Mock() + recommendations = Mock() + launch_info = Mock() + monkeypatch.setattr(local_ui, "_print_system_info", system_info) + monkeypatch.setattr(local_ui, "_print_recommendations", recommendations) + monkeypatch.setattr(local_ui, "_print_launch_info", launch_info) + + local_ui.launch_local_ui(host="localhost", port=7861, quiet=False) + + system_info.assert_called_once_with(gpus) + recommendations.assert_called_once_with("large") + launch_info.assert_called_once_with("localhost", 7861, False, None) + launch.assert_called_once() diff --git a/tests/test_tourney_contracts.py b/tests/test_tourney_contracts.py new file mode 100644 index 0000000..3876e81 --- /dev/null +++ b/tests/test_tourney_contracts.py @@ -0,0 +1,411 @@ +"""Deterministic tournament lifecycle, checkpoint, and rendering contracts.""" + +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from obliteratus import tourney + + +def _contender( + method: str, + score: float, + *, + output_dir: str = "", + error: str | None = None, + direction: str = "mean_diff", + cert: str = "GREEN", +) -> tourney.Contender: + return tourney.Contender( + method=method, + score=score, + metrics={ + "refusal_rate": max(0.0, 1.0 - score), + "coherence": max(0.0, score), + "kl_divergence": 0.1, + "perplexity": 12.0, + "direction_method": direction, + "spectral_certification": cert, + }, + output_dir=output_dir, + time_s=2.5, + error=error, + direction_method=direction, + spectral_cert=cert, + ) + + +@pytest.mark.parametrize( + ("certification", "expected"), + [("GREEN", 1.0), ("YELLOW", 0.975), ("RED", 0.95), (None, 0.975)], +) +def test_composite_score_preserves_documented_weighting(certification, expected): + metrics = { + "refusal_rate": 0.0, + "coherence": 1.0, + "kl_divergence": 0.0, + "perplexity": 0.0, + "spectral_certification": certification, + "degenerate_count": 0, + } + + assert tourney.composite_score(metrics) == pytest.approx(expected) + + +def test_composite_score_missing_metrics_and_degenerate_outputs_fail_safe(): + assert tourney.composite_score({}) == pytest.approx(0.225) + assert tourney.composite_score({"degenerate_count": 3}) == pytest.approx(0.1875) + + +def test_result_dictionary_sorts_contenders_without_mutating_round_order(): + low = _contender("low", 0.2) + high = _contender("high", 0.9) + rnd = tourney.TourneyRound( + round_num=1, + name="Qualifier", + contenders=[low, high], + prompt_volume=64, + advanced_to=["high"], + eliminated=["low"], + ) + result = tourney.TourneyResult( + model="org/model", + winner=high, + rounds=[rnd], + total_time_s=12.5, + timestamp="2026-08-16T00:00:00+00:00", + ) + + payload = result.to_dict() + + assert payload["winner"]["method"] == "high" + assert [item["method"] for item in payload["rounds"][0]["contenders"]] == [ + "high", + "low", + ] + assert rnd.contenders == [low, high] + + +def test_checkpoint_round_trip_preserves_completed_and_partial_metadata(tmp_path): + completed = _contender("complete", 0.8, direction="svd", cert="YELLOW") + completed.round_eliminated = 1 + partial = _contender("partial", 0.7, direction="pca", cert="RED") + result = tourney.TourneyResult( + model="org/model", + rounds=[ + tourney.TourneyRound( + round_num=1, + name="Qualifier", + contenders=[completed], + prompt_volume=64, + advanced_to=[], + eliminated=["complete"], + ) + ], + ) + + path = tourney._save_checkpoint( + output_dir=tmp_path, + result=result, + current_round_num=2, + current_round_name="Semifinals", + current_round_volume=128, + current_round_advance=1, + current_round_verify=30, + completed_methods=[partial], + remaining_methods=["remaining"], + alive=["partial", "remaining"], + model_name="org/model", + dataset_key="builtin", + quantization="4bit", + methods=["complete", "partial", "remaining"], + ) + + checkpoint = tourney._load_checkpoint(tmp_path) + restored, partials, remaining, interrupted = tourney._restore_rounds(checkpoint) + + assert path == tmp_path / tourney.CHECKPOINT_FILENAME + assert tourney._checkpoint_matches(checkpoint, "org/model", "builtin", "4bit") + assert not tourney._checkpoint_matches(checkpoint, "other/model", "builtin", "4bit") + assert restored.rounds[0].contenders[0].direction_method == "svd" + assert restored.rounds[0].contenders[0].spectral_cert == "YELLOW" + assert partials[0].direction_method == "pca" + assert partials[0].spectral_cert == "RED" + assert partials[0].round_eliminated == 0 + assert remaining == ["remaining"] + assert interrupted["verify_sample_size"] == 30 + + +@pytest.mark.parametrize("payload", ["[]", "null", "{}", '{"version": 2}', "not-json"]) +def test_checkpoint_loader_rejects_malformed_or_unsupported_roots(tmp_path, payload): + (tmp_path / tourney.CHECKPOINT_FILENAME).write_text(payload, encoding="utf-8") + + assert tourney._load_checkpoint(tmp_path) is None + + +def test_checkpoint_loader_returns_none_when_absent(tmp_path): + assert tourney._load_checkpoint(tmp_path) is None + + +@pytest.mark.parametrize("error", [OSError("unreadable"), UnicodeError("invalid encoding")]) +def test_checkpoint_loader_fails_closed_when_document_cannot_be_read( + monkeypatch, + tmp_path, + error, +): + path = tmp_path / tourney.CHECKPOINT_FILENAME + path.write_text('{"version": 1}', encoding="utf-8") + monkeypatch.setattr(Path, "read_text", Mock(side_effect=error)) + + assert tourney._load_checkpoint(tmp_path) is None + + +def test_markdown_and_html_render_all_outcomes_and_escape_html(): + winner = _contender("", 0.85, direction="") + middle = _contender("middle", 0.55, cert="YELLOW") + low = _contender("low", 0.2, cert="INCONCLUSIVE") + failed = _contender("failed", -1.0, error="boom") + rnd = tourney.TourneyRound( + round_num=1, + name="", + contenders=[failed, low, middle, winner], + prompt_volume=64, + advanced_to=[winner.method, middle.method], + eliminated=[low.method, failed.method], + ) + result = tourney.TourneyResult( + model="org/", + winner=winner, + rounds=[rnd], + total_time_s=120, + hub_repo="org/result", + ) + + markdown = tourney.render_bracket(result) + html = tourney.render_bracket_html(result) + + assert "Pushed to: [org/result]" in markdown + assert "| 4 | failed | — | ERROR" in markdown + assert "*out*" not in markdown # the only round is final + assert "<model>" in html + assert "<winner>" in html + assert "<script>alert(1)</script>" in html + assert "" not in html + assert "card-score good" in html + assert "card-score mid" in html + assert "card-score bad" in html + assert "badge-err" in html + + +def test_renderers_handle_no_winner_and_model_card_requires_winner(): + result = tourney.TourneyResult(model="model", total_time_s=0) + + assert "**No winner**" in tourney.render_bracket(result) + assert "No winner determined" in tourney.render_bracket_html(result) + assert tourney.generate_model_card(result) == "" + + +def test_model_card_includes_winner_metrics_and_bracket(): + winner = _contender("advanced", 0.9) + result = tourney.TourneyResult( + model="org/base-model", + winner=winner, + rounds=[ + tourney.TourneyRound( + round_num=1, + name="Final", + contenders=[winner], + prompt_volume=64, + advanced_to=["advanced"], + ) + ], + timestamp="2026-08-16T00:00:00+00:00", + ) + + card = tourney.generate_model_card(result) + + assert "base_model: org/base-model" in card + assert "Winning Method: `advanced`" in card + assert "# OBLITERATUS TOURNEY" in card + + +def test_runner_initialization_cleans_fresh_output_and_preserves_resume(tmp_path): + output = tmp_path / "tourney" + output.mkdir() + (output / "stale.txt").write_text("stale", encoding="utf-8") + + fresh = tourney.TourneyRunner("model", output_dir=str(output), methods=["a"]) + assert fresh.output_dir == output + assert not (output / "stale.txt").exists() + + (output / "checkpoint.txt").write_text("keep", encoding="utf-8") + resumed = tourney.TourneyRunner( + "model", + output_dir=str(output), + methods=["a"], + resume=True, + ) + assert (resumed.output_dir / "checkpoint.txt").read_text(encoding="utf-8") == "keep" + + +def test_runner_prompt_loading_is_bounded_by_shortest_source(monkeypatch, tmp_path): + runner = tourney.TourneyRunner("model", output_dir=str(tmp_path), methods=["a"]) + loader = Mock(return_value=(["h1", "h2", "h3"], ["s1", "s2"])) + monkeypatch.setattr("obliteratus.prompts.load_dataset_source", loader) + + harmful, harmless = runner._load_prompts(10) + + assert harmful == ["h1", "h2"] + assert harmless == ["s1", "s2"] + loader.assert_called_once_with("builtin") + + +@pytest.mark.parametrize( + ("message", "expected"), + [ + ("GPU quota exceeded for this session", True), + ("ZeroGPU token expired", True), + ("ordinary model failure", False), + ], +) +def test_quota_error_classification_is_narrow(message, expected): + assert tourney.TourneyRunner._is_quota_error(RuntimeError(message)) is expected + + +def test_run_one_method_uses_optional_gpu_wrapper(monkeypatch, tmp_path): + runner = tourney.TourneyRunner("model", output_dir=str(tmp_path), methods=["a"]) + direct = Mock(return_value=_contender("a", 0.8)) + monkeypatch.setattr(runner, "_run_method", direct) + + assert runner._run_one_method("a", ["h"], ["s"], "out", 20, None).method == "a" + wrapper = Mock(side_effect=lambda fn, *args: fn(*args)) + assert runner._run_one_method("a", ["h"], ["s"], "out", 20, wrapper).method == "a" + wrapper.assert_called_once() + + +def test_full_runner_ranks_rounds_cleans_losers_and_writes_results(monkeypatch, tmp_path): + logs: list[str] = [] + rounds: list[tourney.TourneyRound] = [] + runner = tourney.TourneyRunner( + "org/model", + methods=["alpha", "beta", "gamma", "delta"], + output_dir=str(tmp_path / "run"), + on_log=logs.append, + on_round=rounds.append, + ) + scores = {"alpha": 0.9, "beta": 0.7, "gamma": 0.4, "delta": 0.2} + monkeypatch.setattr(runner, "_load_prompts", lambda volume: (["h"] * volume, ["s"] * volume)) + monkeypatch.setattr( + tourney.shutil, + "disk_usage", + lambda _path: SimpleNamespace(free=int(4.5e9)), + ) + + def run_method(method, _harmful, _harmless, save_dir, _verify): + Path(save_dir).mkdir(parents=True, exist_ok=True) + return _contender(method, scores[method], output_dir=save_dir) + + monkeypatch.setattr(runner, "_run_method", run_method) + + result = runner.run() + + assert [rnd.name for rnd in result.rounds] == ["Qualifiers", "Semifinals", "Championship"] + assert result.winner.method == "alpha" + assert rounds == result.rounds + assert (runner.output_dir / "tourney_results.json").exists() + assert (runner.output_dir / "tourney_bracket.md").exists() + assert (runner.output_dir / "r3_alpha").exists() + assert not (runner.output_dir / "r3_beta").exists() + assert any("Low disk space" in line for line in logs) + + +def test_full_runner_does_not_crown_an_errored_only_contender(monkeypatch, tmp_path): + runner = tourney.TourneyRunner("model", methods=["broken"], output_dir=str(tmp_path)) + monkeypatch.setattr(runner, "_load_prompts", lambda _volume: (["h"], ["s"])) + monkeypatch.setattr( + runner, + "_run_method", + lambda method, *_args: _contender(method, -1.0, error="broken"), + ) + + result = runner.run() + + assert result.winner is None + assert json.loads((tmp_path / "tourney_results.json").read_text())["winner"] is None + + +def test_run_iter_saves_exact_resume_point_on_quota_exhaustion(monkeypatch, tmp_path): + runner = tourney.TourneyRunner( + "model", + methods=["alpha", "beta"], + output_dir=str(tmp_path), + ) + monkeypatch.setattr(runner, "_load_prompts", lambda _volume: (["h"], ["s"])) + + def run_one(method, *_args): + if method == "beta": + raise RuntimeError("GPU quota exceeded") + return _contender(method, 0.8) + + monkeypatch.setattr(runner, "_run_one_method", run_one) + iterator = runner.run_iter() + + assert "running `alpha`" in next(iterator)[0] + assert "running `beta`" in next(iterator)[0] + with pytest.raises(RuntimeError, match="GPU quota exceeded"): + next(iterator) + + checkpoint = tourney._load_checkpoint(tmp_path) + interrupted = checkpoint["interrupted_round"] + assert [item["method"] for item in interrupted["completed_methods"]] == ["alpha"] + assert interrupted["remaining_methods"] == ["beta"] + + +def test_run_iter_resumes_partial_round_without_repeating_completed_method(monkeypatch, tmp_path): + partial = _contender("alpha", 0.9, direction="pca", cert="YELLOW") + tourney._save_checkpoint( + output_dir=tmp_path, + result=tourney.TourneyResult(model="model"), + current_round_num=1, + current_round_name="Qualifiers", + current_round_volume=64, + current_round_advance=1, + current_round_verify=20, + completed_methods=[partial], + remaining_methods=["beta"], + alive=["alpha", "beta"], + model_name="model", + dataset_key="builtin", + quantization=None, + methods=["alpha", "beta"], + ) + runner = tourney.TourneyRunner( + "model", + methods=["alpha", "beta"], + output_dir=str(tmp_path), + resume=True, + ) + monkeypatch.setattr(runner, "_load_prompts", lambda _volume: (["h"], ["s"])) + called: list[str] = [] + + def run_one(method, *_args): + called.append(method) + return _contender(method, 0.7) + + monkeypatch.setattr(runner, "_run_one_method", run_one) + + events = list(runner.run_iter()) + + assert events[0][0].startswith("**Resuming tournament**") + assert events[-1][0] == "Tournament complete" + result = events[-1][1] + assert called == ["beta"] + assert result.winner.method == "alpha" + assert result.winner.direction_method == "pca" + assert result.winner.spectral_cert == "YELLOW" + assert not (tmp_path / tourney.CHECKPOINT_FILENAME).exists() diff --git a/tests/test_ui_watchtower_contracts.py b/tests/test_ui_watchtower_contracts.py new file mode 100644 index 0000000..40523f4 --- /dev/null +++ b/tests/test_ui_watchtower_contracts.py @@ -0,0 +1,249 @@ +"""Pure handler contracts for the Watchtower and one-click Gradio tabs.""" + +from __future__ import annotations + +import importlib +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +auto_obliterate = importlib.import_module("obliteratus.auto_obliterate") +ui_watchtower = importlib.import_module("obliteratus.ui_watchtower") + + +def test_component_update_delegates_to_gradio_when_available(monkeypatch): + update = Mock(return_value={"delegated": True}) + monkeypatch.setattr(ui_watchtower, "gr", SimpleNamespace(update=update)) + + assert ui_watchtower._component_update(interactive=True) == {"delegated": True} + update.assert_called_once_with(interactive=True) + + +def test_trending_choices_use_watchtower_and_fall_back_on_empty_or_error(monkeypatch): + watchtower = SimpleNamespace(get_model_choices=lambda: ["org/one", "org/two"]) + monkeypatch.setattr(ui_watchtower, "_get_watchtower", lambda: watchtower) + assert ui_watchtower._get_trending_choices() == ["org/one", "org/two"] + + watchtower.get_model_choices = lambda: [] + assert ui_watchtower._get_trending_choices()[0] == "meta-llama/Llama-3.1-8B-Instruct" + + monkeypatch.setattr( + ui_watchtower, + "_get_watchtower", + Mock(side_effect=RuntimeError("offline")), + ) + assert "Qwen/Qwen3-4B" in ui_watchtower._get_trending_choices() + + +@pytest.mark.parametrize("model_id", ["", " ", None]) +def test_one_click_rejects_missing_model_without_starting_work(model_id): + outputs = list(ui_watchtower._run_one_click(model_id, 3, 5)) + + assert len(outputs) == 1 + assert "Please enter or select" in outputs[0][0] + assert outputs[0][3]["interactive"] is False + + +def test_one_click_streams_progress_and_preserves_zero_refusal_rate(monkeypatch, tmp_path): + output = tmp_path / "model" + output.mkdir() + created: list[object] = [] + final = SimpleNamespace( + success=True, + final_output_dir=str(output), + final_refusal_rate=0.0, + total_time_seconds=12.5, + ) + + class FakeObliterator: + def __init__(self, **kwargs): + created.append(kwargs) + self._result = SimpleNamespace(final_output_dir=str(output)) + + def run(self): + yield "running", "step", "metrics" + return final + + def _format_metrics(self): + return "final metrics" + + monkeypatch.setattr(auto_obliterate, "AutoObliterator", FakeObliterator) + + outputs = list(ui_watchtower._run_one_click(" org/model ", 3.9, 5)) + + assert created == [ + { + "model_id": "org/model", + "max_iterations": 3, + "target_refusal_rate": 0.05, + } + ] + assert outputs[0][0] == "running" + assert outputs[0][3]["interactive"] is True + assert outputs[-1][0] == "✅ Complete!" + assert "Final refusal rate: 0.0" in outputs[-1][1] + assert "unknown" not in outputs[-1][1] + + +def test_one_click_converts_generator_failure_to_noninteractive_error(monkeypatch): + class FailingObliterator: + def __init__(self, **_kwargs): + self._result = SimpleNamespace(final_output_dir=None) + + def run(self): + yield "starting", "", "" + raise RuntimeError("device failed") + + monkeypatch.setattr(auto_obliterate, "AutoObliterator", FailingObliterator) + + outputs = list(ui_watchtower._run_one_click("org/model", 1, 5)) + + assert outputs[0][0] == "starting" + assert "device failed" in outputs[-1][0] + assert outputs[-1][3]["interactive"] is False + + +def test_download_result_selects_latest_numeric_iteration(monkeypatch, tmp_path): + monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path)) + base = tmp_path / ".obliteratus" / "auto_obliterate" / "org_model" + for name in ("iter_invalid", "iter_2", "iter_10"): + directory = base / name + directory.mkdir(parents=True) + (directory / "config.json").write_text("{}", encoding="utf-8") + + assert ui_watchtower._download_result("org/model") == str(base / "iter_10") + + +def test_download_result_handles_empty_id_and_unreadable_iteration(monkeypatch, tmp_path): + monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path)) + base = tmp_path / ".obliteratus" / "auto_obliterate" / "org_model" + broken = base / "iter_20" + broken.mkdir(parents=True) + (broken / "config.json").write_text("{}", encoding="utf-8") + valid = base / "iter_10" + valid.mkdir() + (valid / "config.json").write_text("{}", encoding="utf-8") + resolve = Path.resolve + + def controlled_resolve(path, *args, **kwargs): + if path == broken: + raise OSError("unreadable") + return resolve(path, *args, **kwargs) + + monkeypatch.setattr(Path, "resolve", controlled_resolve) + + assert ui_watchtower._download_result("") is None + assert ui_watchtower._download_result("org/model") == str(valid) + + +def test_download_result_rejects_symlink_that_escapes_managed_root(monkeypatch, tmp_path): + monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path)) + base = tmp_path / ".obliteratus" / "auto_obliterate" / "org_model" + valid = base / "iter_10" + valid.mkdir(parents=True) + (valid / "config.json").write_text("{}", encoding="utf-8") + outside = tmp_path / "outside" + outside.mkdir() + (outside / "config.json").write_text("{}", encoding="utf-8") + (base / "iter_99").symlink_to(outside, target_is_directory=True) + + assert ui_watchtower._download_result("../../org/model") is None + assert ui_watchtower._download_result("org/model") == str(valid) + + +def test_scan_handler_returns_data_and_escaped_failure(monkeypatch): + watchtower = SimpleNamespace( + scan=lambda on_log: (on_log("scanned") or [SimpleNamespace()]), + get_stats=lambda: {"total_tracked": 1}, + format_table=lambda: [["org/model"]], + ) + monkeypatch.setattr(ui_watchtower, "_get_watchtower", lambda: watchtower) + + status, table, log = ui_watchtower._scan_now() + + assert "1" in status + assert table == [["org/model"]] + assert log == "scanned" + + monkeypatch.setattr( + ui_watchtower, + "_get_watchtower", + Mock(side_effect=RuntimeError("")), + ) + status, table, log = ui_watchtower._scan_now() + assert "