"""CPU-only behavioral tests for telemetry-driven adaptive defaults.""" from __future__ import annotations import pytest from obliteratus import adaptive_defaults as adaptive def _record( method="advanced", *, architecture="LlamaForCausalLM", params_b=7, refusal=0.1, coherence=0.9, config=None, session="session", ): return { "session_id": session, "timestamp": session, "model": { "architecture": architecture, "num_layers": 32, "hidden_size": 4096, "total_params": int(params_b * 1e9), }, "method": method, "method_config": ( config if config is not None else {"strength": 1.0, "cot_aware": False} ), "quality_metrics": { "refusal_rate": refusal, "coherence": coherence, "kl_divergence": 0.1, "perplexity": 10.0, }, } @pytest.mark.parametrize( ("params", "expected"), [(-1, "tiny"), (0.5, "tiny"), (4, "small"), (16, "medium"), (80, "large"), (81, "frontier")], ) def test_parameter_buckets_have_closed_boundaries(params, expected): assert adaptive._param_bucket(params) == expected def test_architecture_key_handles_schema_estimates_moe_and_reasoning(): assert adaptive._extract_arch_key({"model": "legacy"}) is None assert adaptive._extract_arch_key({"model": {"architecture": "unknown"}}) is None assert adaptive._extract_arch_key(_record()) == ( "dense", "standard", "medium", ) record = _record(architecture="Qwen3_MoE_Reasoning", params_b=120) assert adaptive._extract_arch_key(record) == ("large_moe", "reasoning", "frontier") record = _record(params_b=8) record["model"]["total_params"] = 0 record["method_config"] = {"per_expert_directions": True, "cot_aware": True} assert adaptive._extract_arch_key(record) == ("small_moe", "reasoning", "medium") def test_composite_score_defaults_and_complete_metrics(): assert adaptive._composite_score({}) == pytest.approx(0.15) assert adaptive._composite_score({ "refusal_rate": 0.0, "coherence": 1.0, "kl_divergence": 0.0, "perplexity": 0.0, }) == pytest.approx(1.0) def test_method_statistics_ranges_and_bucket_ranking(): empty = adaptive.MethodStats("empty") assert (empty.mean_score, empty.best_score, empty.median_score) == (0, 0, 0) assert empty.best_config_ranges() == {} stats = adaptive.MethodStats( "strong", n_runs=4, scores=[0.1, 0.9, 0.8, 0.7], configs=[ {"strength": 1, "enabled": False, "label": "skip"}, {"strength": 4, "enabled": True}, {"strength": 3, "enabled": True}, {"strength": 2, "enabled": False}, ], ) assert stats.best_config_ranges() == {"strength": 4, "enabled": True} bucket = adaptive.BucketKnowledge( ("dense", "standard", "small"), methods={"empty": empty, "strong": stats}, total_runs=4, ) assert bucket.best_method == "strong" assert bucket.ranked_methods[0][0] == "strong" assert adaptive.BucketKnowledge(("dense", "standard", "tiny")).best_method is None def test_build_knowledge_base_filters_invalid_runs_and_aggregates_metrics(): records = [ _record(session="one"), _record(method="basic", refusal=0.4, coherence=0.5, config={}, session="two"), {**_record(session="error"), "error": "failed"}, {**_record(session="no-method"), "method": ""}, {**_record(session="no-metrics"), "quality_metrics": {}}, {**_record(session="legacy"), "model": "legacy"}, ] knowledge = adaptive.build_knowledge_base(records) bucket = knowledge[("dense", "standard", "medium")] assert bucket.total_runs == 2 assert bucket.methods["advanced"].n_runs == 1 assert bucket.methods["advanced"].refusal_rates == [0.1] assert bucket.methods["basic"].configs == [] def test_fetch_records_caches_deduplicates_and_tolerates_sources(monkeypatch): adaptive._cache.clear() monkeypatch.setattr(adaptive, "_cache_ts", 0.0) import obliteratus.telemetry as telemetry monkeypatch.setattr(telemetry, "read_telemetry", lambda: [_record(session="same")]) monkeypatch.setattr( telemetry, "fetch_hub_records", lambda: [_record(session="same"), _record(session="other")], ) first = adaptive._fetch_all_records() assert len(first) == 2 monkeypatch.setattr(telemetry, "read_telemetry", lambda: (_ for _ in ()).throw(RuntimeError())) assert adaptive._fetch_all_records() is first adaptive._cache.clear() monkeypatch.setattr(adaptive, "_cache_ts", 0.0) monkeypatch.setattr(telemetry, "fetch_hub_records", lambda: (_ for _ in ()).throw(RuntimeError())) assert adaptive._fetch_all_records() == [] def test_recommendation_exact_fallback_confidence_and_formatting(): records = [ _record("advanced", refusal=0.05, coherence=0.95, session=f"a-{index}") for index in range(5) ] + [ _record("basic", refusal=0.5, coherence=0.4, session=f"b-{index}") for index in range(5) ] knowledge = adaptive.build_knowledge_base(records) rec = adaptive.get_adaptive_recommendation("dense", "standard", 7, knowledge=knowledge) assert rec.recommended_method == "advanced" assert rec.confidence == "medium" assert rec.best_refusal_rate == 0.05 assert "Runner-up" in rec.reason assert rec.to_dict()["arch_key"] == ["dense", "standard", "medium"] formatted = adaptive.format_recommendation(rec) assert "Adaptive Recommendation" in formatted assert "strength" in formatted sparse_knowledge = adaptive.build_knowledge_base([ _record("advanced", params_b=1, session="small"), _record("advanced", params_b=30, session="large"), ]) fallback = adaptive.get_adaptive_recommendation( "dense", "reasoning", 7, knowledge=sparse_knowledge, ) assert fallback.confidence == "low" assert fallback.arch_key == ("dense", "*", "*") none = adaptive.get_adaptive_recommendation("small_moe", "standard", 7, knowledge={}) assert none.confidence == "none" assert "No telemetry data" in adaptive.format_recommendation(none) def test_global_insights_reports_rankings_buckets_and_hyperparameters(): knowledge = adaptive.build_knowledge_base([ _record("advanced", config={"strength": 1.5, "enabled": True}, session="one"), _record("basic", config={"strength": 0.5, "enabled": False}, session="two"), ]) insights = adaptive.get_global_insights(knowledge) assert insights["total_records"] == 2 assert insights["overall_best_methods"][0]["method"] == "advanced" assert insights["hyperparameter_trends"]["strength"]["type"] == "numeric" assert insights["hyperparameter_trends"]["enabled"]["type"] == "bool"