From ffdab00eaea0776c2777c3aafa545e262d642215 Mon Sep 17 00:00:00 2001 From: Dave Fobare Date: Thu, 14 May 2026 10:37:44 -0400 Subject: [PATCH] Make hardware profile an explicit runtime policy --- SourceCode/configs/hardware_profiles.json | 52 +++++ SourceCode/core/kernel_commands/service.py | 17 +- SourceCode/orchestrator/main.py | 16 +- SourceCode/shared_tools/hardware_profiles.py | 212 ++++++++++++++++++ SourceCode/shared_tools/inference_router.py | 73 ++++++- tests/test_hardware_profiles.py | 215 +++++++++++++++++++ 6 files changed, 572 insertions(+), 13 deletions(-) create mode 100644 SourceCode/configs/hardware_profiles.json create mode 100644 SourceCode/shared_tools/hardware_profiles.py create mode 100644 tests/test_hardware_profiles.py diff --git a/SourceCode/configs/hardware_profiles.json b/SourceCode/configs/hardware_profiles.json new file mode 100644 index 0000000..5406867 --- /dev/null +++ b/SourceCode/configs/hardware_profiles.json @@ -0,0 +1,52 @@ +{ + "default_profile": "8gb_vram_16gb_ram", + "profiles": { + "8gb_vram_16gb_ram": { + "name": "8gb_vram_16gb_ram", + "display_name": "Default: 8GB VRAM / 16GB RAM", + "description": "Conservative default local profile.", + "hardware": { + "system_ram_gb": 16, + "gpu_backend": "generic", + "gpu_vram_gb": 8, + "unified_memory": false + }, + "scheduler": { + "max_context_tokens": 4096, + "warning_context_tokens": 4096, + "max_stage_context_tokens": 1800, + "max_parallel_models": 1, + "max_active_model_calls": 1, + "on_deck_depth": 1, + "warm_depth": 1, + "allow_neural_prefetch": true + }, + "inference": { + "preferred_backends": ["ollama", "llama.cpp"], + "default_keep_alive": "10m", + "heavy_keep_alive": "0", + "release_heavy_after_call": true, + "max_loaded_models": 1 + }, + "model_policy": { + "normal_max_b": 9, + "heavy_max_b": 14, + "premium_min_b": 24, + "allow_premium": false, + "premium_requires_manual": true, + "allow_14b_with_warning": true, + "reject_heavier_fallbacks": false + }, + "lane_caps": {}, + "validation": { + "startup_mode": "warn", + "strict_mode_available": true, + "warn_on_missing_models": true, + "warn_on_unreachable_backends": true, + "warn_on_context_over_cap": true, + "warn_on_parallelism_over_cap": true, + "warn_on_premium_auto_escalation": true + } + } + } +} diff --git a/SourceCode/core/kernel_commands/service.py b/SourceCode/core/kernel_commands/service.py index 665bca9..74d7446 100644 --- a/SourceCode/core/kernel_commands/service.py +++ b/SourceCode/core/kernel_commands/service.py @@ -11,6 +11,11 @@ from cag.memory_store import CAGMemoryStore from orchestrator.main import OathweaverOrchestrator from orchestrator.pipelines import replay_turn +from shared_tools.hardware_profiles import ( + hardware_profile_summary, + hardware_profile_to_scheduler, + resolve_active_hardware_profile, +) class KernelCommandService: @@ -226,11 +231,19 @@ def benchmark_workflow_eval( return evaluator.evaluate_run(run_id=target_run, hardware_profile_name=hardware_profile) def apply_hardware_profile(self, *, profile_name: str = "8gb_vram_16gb_ram") -> dict[str, Any]: + active = resolve_active_hardware_profile(self.repo_root, profile_name) + summary = hardware_profile_summary(active) + if hasattr(self.orchestrator, "resource_budget_manager"): + self.orchestrator.resource_budget_manager.profile = hardware_profile_to_scheduler(active) + if hasattr(self.orchestrator, "hardware_profile"): + self.orchestrator.hardware_profile = active + if not summary.get("resolution_warnings"): + return {"ok": True, "profile": summary} + profile = profile_by_name(profile_name) - # Phase-12 guarantee: scheduler and context budget honor benchmark profile. if hasattr(self.orchestrator, "resource_budget_manager"): self.orchestrator.resource_budget_manager.profile = profile.to_scheduler_profile() - return {"ok": True, "profile": profile.as_dict()} + return {"ok": True, "profile": profile.as_dict(), "warnings": summary.get("resolution_warnings", [])} def stage_resume( self, diff --git a/SourceCode/orchestrator/main.py b/SourceCode/orchestrator/main.py index 441c7b9..6e6b18b 100755 --- a/SourceCode/orchestrator/main.py +++ b/SourceCode/orchestrator/main.py @@ -25,6 +25,11 @@ from shared_tools.domain_reputation import DomainReputation from shared_tools.feedback_learning import ORIGIN_REFLECTION from shared_tools.handoff_queue import HandoffQueue +from shared_tools.hardware_profiles import ( + hardware_profile_summary, + hardware_profile_to_scheduler, + resolve_active_hardware_profile, +) from shared_tools.model_routing import load_model_routing, lane_model_config from shared_tools.inference_router import InferenceRouter from shared_tools.web_research import build_web_progress_payload @@ -266,7 +271,10 @@ def __init__(self, repo_root: Path) -> None: self.model_runtime = build_model_runtime(repo_root) self.context_pack_store = ContextPackStore(repo_root) self.context_compiler = ContextCompiler(context_pack_store=self.context_pack_store) - self.resource_budget_manager = ResourceBudgetManager() + self.hardware_profile = resolve_active_hardware_profile(repo_root) + self.resource_budget_manager = ResourceBudgetManager( + profile=hardware_profile_to_scheduler(self.hardware_profile) + ) self.specialist_registry = SpecialistRegistry() self.bench_manager = BenchManager(repo_root) self.on_deck_runtime = OnDeckRuntime( @@ -685,7 +693,7 @@ def _execute_pipeline_turn( "mode": mode, "history": list(history or []), "hardware_token_budget": adaptive_stage_budget, - "hardware_profile": self.resource_budget_manager.profile.as_dict(), + "hardware_profile": hardware_profile_summary(self.hardware_profile), } scratch: dict[str, Any] = { "web_note": "", @@ -1167,7 +1175,7 @@ def _on_deck_planner( "stage_outputs": dict(stage_outputs), "stage_audits": dict(stage_audits), "stage_timings_ms": dict(stage_timings_ms), - "hardware_profile": self.resource_budget_manager.profile.as_dict(), + "hardware_profile": hardware_profile_summary(self.hardware_profile), "promoted_memory_ids": list(promoted_memory_ids), "started_at": started_at, "finished_at": finished_at, @@ -1267,7 +1275,7 @@ def _on_deck_planner( stage_outputs=stage_outputs, stage_audits=stage_audits, stage_timings_ms=stage_timings_ms, - hardware_profile=self.resource_budget_manager.profile.as_dict(), + hardware_profile=hardware_profile_summary(self.hardware_profile), promoted_memory_ids=promoted_memory_ids, started_at=started_at, finished_at=finished_at, diff --git a/SourceCode/shared_tools/hardware_profiles.py b/SourceCode/shared_tools/hardware_profiles.py new file mode 100644 index 0000000..9f7b992 --- /dev/null +++ b/SourceCode/shared_tools/hardware_profiles.py @@ -0,0 +1,212 @@ +from __future__ import annotations + +import json +import os +from copy import deepcopy +from pathlib import Path +from typing import Any + +from scheduler.resource_budget import DEFAULT_PROFILE, HardwareBudgetProfile + +ENV_HARDWARE_PROFILE = "OATHWEAVER_HARDWARE_PROFILE" +CONFIG_RELATIVE_PATH = Path("SourceCode") / "configs" / "hardware_profiles.json" + + +def _coerce_int(value: Any, default: int) -> int: + try: + return int(value) + except Exception: + return int(default) + + +def _coerce_float(value: Any, default: float) -> float: + try: + return float(value) + except Exception: + return float(default) + + +def _builtin_config() -> dict[str, Any]: + default = DEFAULT_PROFILE.as_dict() + return { + "default_profile": DEFAULT_PROFILE.name, + "profiles": { + DEFAULT_PROFILE.name: { + "name": DEFAULT_PROFILE.name, + "display_name": "Default: 8GB VRAM / 16GB RAM", + "description": "Built-in conservative default local profile.", + "hardware": { + "system_ram_gb": default["ram_gb"], + "gpu_backend": "generic", + "gpu_vram_gb": default["vram_gb"], + "unified_memory": False, + }, + "scheduler": { + "max_context_tokens": default["max_context_tokens"], + "warning_context_tokens": default["max_context_tokens"], + "max_stage_context_tokens": default["max_stage_context_tokens"], + "max_parallel_models": default["max_parallel_models"], + "max_active_model_calls": default["max_parallel_models"], + "on_deck_depth": default["on_deck_depth"], + "warm_depth": default["warm_depth"], + "allow_neural_prefetch": default["allow_neural_prefetch"], + }, + "inference": { + "preferred_backends": ["ollama", "llama.cpp"], + "default_keep_alive": "10m", + "heavy_keep_alive": "0", + "release_heavy_after_call": True, + "max_loaded_models": default["max_parallel_models"], + }, + "model_policy": { + "normal_max_b": 9, + "heavy_max_b": 14, + "premium_min_b": 24, + "allow_premium": False, + "premium_requires_manual": True, + "allow_14b_with_warning": True, + "reject_heavier_fallbacks": False, + }, + "lane_caps": {}, + "validation": { + "startup_mode": "warn", + "strict_mode_available": True, + "warn_on_missing_models": True, + "warn_on_unreachable_backends": True, + "warn_on_context_over_cap": True, + "warn_on_parallelism_over_cap": True, + "warn_on_premium_auto_escalation": True, + }, + } + }, + } + + +def load_hardware_profiles(repo_root: Path) -> dict[str, Any]: + """Load structured hardware profiles, falling back to the legacy default.""" + config_path = Path(repo_root) / CONFIG_RELATIVE_PATH + if not config_path.exists(): + return _builtin_config() + try: + payload = json.loads(config_path.read_text(encoding="utf-8")) + except Exception: + return _builtin_config() + if not isinstance(payload, dict): + return _builtin_config() + profiles = payload.get("profiles") + if not isinstance(profiles, dict) or not profiles: + return _builtin_config() + return payload + + +def resolve_active_hardware_profile(repo_root: Path, name: str | None = None) -> dict[str, Any]: + """Resolve the active hardware profile by explicit name, env var, then config default.""" + payload = load_hardware_profiles(repo_root) + profiles = payload.get("profiles") if isinstance(payload.get("profiles"), dict) else {} + default_name = str(payload.get("default_profile") or DEFAULT_PROFILE.name).strip() or DEFAULT_PROFILE.name + requested = str(name or os.getenv(ENV_HARDWARE_PROFILE, "") or default_name).strip() + if not requested: + requested = default_name + key = requested.lower() + + selected: dict[str, Any] | None = None + for profile_name, profile in profiles.items(): + if str(profile_name).strip().lower() == key and isinstance(profile, dict): + selected = dict(profile) + break + + warnings: list[str] = [] + if selected is None: + fallback = profiles.get(default_name) + selected = dict(fallback) if isinstance(fallback, dict) else _builtin_config()["profiles"][DEFAULT_PROFILE.name] + warnings.append(f"Unknown hardware profile {requested!r}; using {selected.get('name', DEFAULT_PROFILE.name)!r}.") + + selected.setdefault("name", requested if not warnings else selected.get("name", DEFAULT_PROFILE.name)) + if warnings: + selected["_resolution_warnings"] = warnings + return deepcopy(selected) + + +def hardware_profile_to_scheduler(profile: dict[str, Any]) -> HardwareBudgetProfile: + scheduler = profile.get("scheduler") if isinstance(profile.get("scheduler"), dict) else {} + hardware = profile.get("hardware") if isinstance(profile.get("hardware"), dict) else {} + return HardwareBudgetProfile( + name=str(profile.get("name") or DEFAULT_PROFILE.name), + vram_gb=_coerce_float(hardware.get("gpu_vram_gb"), DEFAULT_PROFILE.vram_gb), + ram_gb=_coerce_float(hardware.get("system_ram_gb"), DEFAULT_PROFILE.ram_gb), + max_context_tokens=_coerce_int(scheduler.get("max_context_tokens"), DEFAULT_PROFILE.max_context_tokens), + max_parallel_models=_coerce_int(scheduler.get("max_parallel_models"), DEFAULT_PROFILE.max_parallel_models), + on_deck_depth=_coerce_int(scheduler.get("on_deck_depth"), DEFAULT_PROFILE.on_deck_depth), + warm_depth=_coerce_int(scheduler.get("warm_depth"), DEFAULT_PROFILE.warm_depth), + max_stage_context_tokens=_coerce_int( + scheduler.get("max_stage_context_tokens"), + DEFAULT_PROFILE.max_stage_context_tokens, + ), + allow_neural_prefetch=bool(scheduler.get("allow_neural_prefetch", DEFAULT_PROFILE.allow_neural_prefetch)), + ) + + +def hardware_profile_to_router_policy(profile: dict[str, Any]) -> dict[str, Any]: + scheduler = profile.get("scheduler") if isinstance(profile.get("scheduler"), dict) else {} + hardware = profile.get("hardware") if isinstance(profile.get("hardware"), dict) else {} + model_policy = profile.get("model_policy") if isinstance(profile.get("model_policy"), dict) else {} + lane_caps = profile.get("lane_caps") if isinstance(profile.get("lane_caps"), dict) else {} + return { + "name": str(profile.get("name") or DEFAULT_PROFILE.name), + "max_context": _coerce_int(scheduler.get("max_context_tokens"), DEFAULT_PROFILE.max_context_tokens), + "warning_context": _coerce_int( + scheduler.get("warning_context_tokens"), + _coerce_int(scheduler.get("max_context_tokens"), DEFAULT_PROFILE.max_context_tokens), + ), + "max_concurrency": _coerce_int( + scheduler.get("max_active_model_calls"), + _coerce_int(scheduler.get("max_parallel_models"), DEFAULT_PROFILE.max_parallel_models), + ), + "allow_premium": bool(model_policy.get("allow_premium", False)), + "normal_max_b": _coerce_float(model_policy.get("normal_max_b"), 9.0), + "heavy_max_b": _coerce_float(model_policy.get("heavy_max_b"), 14.0), + "premium_min_b": _coerce_float(model_policy.get("premium_min_b"), 24.0), + "premium_requires_manual": bool(model_policy.get("premium_requires_manual", True)), + "allow_14b_with_warning": bool(model_policy.get("allow_14b_with_warning", True)), + "reject_heavier_fallbacks": bool(model_policy.get("reject_heavier_fallbacks", False)), + "gpu_backend": str(hardware.get("gpu_backend") or "generic"), + "gpu_vram_gb": _coerce_float(hardware.get("gpu_vram_gb"), DEFAULT_PROFILE.vram_gb), + "lane_caps": deepcopy(lane_caps), + } + + +def hardware_profile_summary(profile: dict[str, Any]) -> dict[str, Any]: + scheduler_profile = hardware_profile_to_scheduler(profile) + scheduler = profile.get("scheduler") if isinstance(profile.get("scheduler"), dict) else {} + summary = scheduler_profile.as_dict() + summary.update( + { + "display_name": str(profile.get("display_name") or profile.get("name") or scheduler_profile.name), + "description": str(profile.get("description") or ""), + "warning_context_tokens": _coerce_int( + scheduler.get("warning_context_tokens"), + scheduler_profile.max_context_tokens, + ), + "max_active_model_calls": _coerce_int( + scheduler.get("max_active_model_calls"), + scheduler_profile.max_parallel_models, + ), + "hardware": deepcopy(profile.get("hardware") if isinstance(profile.get("hardware"), dict) else {}), + "scheduler": deepcopy(profile.get("scheduler") if isinstance(profile.get("scheduler"), dict) else {}), + "inference": deepcopy(profile.get("inference") if isinstance(profile.get("inference"), dict) else {}), + "model_policy": deepcopy(profile.get("model_policy") if isinstance(profile.get("model_policy"), dict) else {}), + "lane_caps": deepcopy(profile.get("lane_caps") if isinstance(profile.get("lane_caps"), dict) else {}), + "validation": deepcopy(profile.get("validation") if isinstance(profile.get("validation"), dict) else {}), + } + ) + if isinstance(profile.get("_resolution_warnings"), list): + summary["resolution_warnings"] = list(profile.get("_resolution_warnings") or []) + return summary + + +def active_router_policy_from_env(repo_root: Path) -> dict[str, Any] | None: + """Return a router policy only when the operator explicitly selected a profile.""" + selected = str(os.getenv(ENV_HARDWARE_PROFILE, "")).strip() + if not selected: + return None + return hardware_profile_to_router_policy(resolve_active_hardware_profile(repo_root, selected)) diff --git a/SourceCode/shared_tools/inference_router.py b/SourceCode/shared_tools/inference_router.py index a088229..8b317fe 100644 --- a/SourceCode/shared_tools/inference_router.py +++ b/SourceCode/shared_tools/inference_router.py @@ -18,6 +18,7 @@ from pathlib import Path from typing import Any +from shared_tools.hardware_profiles import active_router_policy_from_env, hardware_profile_to_router_policy from shared_tools.llamacpp_client import LlamaCppClient from shared_tools.model_routing import load_model_routing from shared_tools.ollama_client import OllamaClient @@ -591,6 +592,8 @@ def estimate_fit( profile: dict[str, Any] | str | None = None, ) -> dict[str, Any]: """Conservative policy estimate, not a hardware memory calculator.""" + if profile is None: + profile = active_router_policy_from_env(self.repo_root) name = str(model or "").strip() caps = self.capabilities(name) warnings: list[str] = [] @@ -695,10 +698,31 @@ def health_report(self) -> dict[str, Any]: "warnings": warnings, } - def validate_config(self, *, strict: bool = False, check_remote: bool = True) -> dict[str, Any]: + def validate_config( + self, + *, + strict: bool = False, + check_remote: bool = True, + profile: dict[str, Any] | None = None, + ) -> dict[str, Any]: errors: list[str] = [] warnings: list[str] = [] referenced = self._configured_model_names() + profile_policy: dict[str, Any] | None = None + if isinstance(profile, dict): + if "scheduler" in profile or "model_policy" in profile: + profile_policy = hardware_profile_to_router_policy(profile) + else: + profile_policy = dict(profile) + else: + profile_policy = active_router_policy_from_env(self.repo_root) + profile_name = str((profile_policy or {}).get("name") or "").strip() + max_context = int((profile_policy or {}).get("max_context") or 32768) + warning_context = int((profile_policy or {}).get("warning_context") or 16384) + max_concurrency = int((profile_policy or {}).get("max_concurrency") or 2) + allow_premium = bool((profile_policy or {}).get("allow_premium", False)) + lane_caps = (profile_policy or {}).get("lane_caps") + lane_caps = lane_caps if isinstance(lane_caps, dict) else {} servers = self._routing.get("llama_cpp_servers") if servers is not None and not isinstance(servers, dict): @@ -747,16 +771,50 @@ def validate_config(self, *, strict: bool = False, check_remote: bool = True) -> ctx = int(cfg.get("num_ctx") or 0) except Exception: ctx = 0 - if ctx > 32768: - errors.append(f"{path}.num_ctx={ctx} exceeds the default hard cap 32768.") - elif ctx > 16384: - warnings.append(f"{path}.num_ctx={ctx} exceeds the conservative local warning threshold 16384.") + if ctx > max_context: + if profile_name: + warnings.append(f"{path}.num_ctx={ctx} exceeds hardware profile {profile_name} cap {max_context}.") + else: + errors.append(f"{path}.num_ctx={ctx} exceeds the default hard cap 32768.") + elif ctx > warning_context: + if profile_name: + warnings.append( + f"{path}.num_ctx={ctx} exceeds hardware profile {profile_name} warning threshold {warning_context}." + ) + else: + warnings.append(f"{path}.num_ctx={ctx} exceeds the conservative local warning threshold 16384.") try: parallel = int(cfg.get("parallel_agents") or 0) except Exception: parallel = 0 - if parallel > 2: - warnings.append(f"{path}.parallel_agents={parallel} exceeds the conservative local default 2.") + if parallel > max_concurrency: + if profile_name: + warnings.append( + f"{path}.parallel_agents={parallel} exceeds hardware profile {profile_name} concurrency {max_concurrency}." + ) + else: + warnings.append(f"{path}.parallel_agents={parallel} exceeds the conservative local default 2.") + lane_key = str(path).split(".", 1)[0] + cap = lane_caps.get(lane_key) + cap = cap if isinstance(cap, dict) else {} + try: + lane_max_ctx = int(cap.get("max_context_tokens") or 0) + except Exception: + lane_max_ctx = 0 + if lane_max_ctx and ctx > lane_max_ctx: + warnings.append(f"{path}.num_ctx={ctx} exceeds {profile_name or 'profile'} lane cap {lane_max_ctx}.") + try: + lane_max_parallel = int(cap.get("max_parallel_agents") or 0) + except Exception: + lane_max_parallel = 0 + if lane_max_parallel and parallel > lane_max_parallel: + warnings.append( + f"{path}.parallel_agents={parallel} exceeds {profile_name or 'profile'} lane cap {lane_max_parallel}." + ) + if cap.get("allow_premium") is False and (primary in premium_models or self._weight_class(primary) == "premium"): + warnings.append(f"{path}.model uses premium model {primary}, but {profile_name or 'profile'} lane policy disallows premium.") + if profile_name and not allow_premium and (primary in premium_models or self._weight_class(primary) == "premium"): + warnings.append(f"{path}.model uses premium model {primary}, but hardware profile {profile_name} disallows premium.") if check_remote: backends = self.list_backends() @@ -785,6 +843,7 @@ def validate_config(self, *, strict: bool = False, check_remote: bool = True) -> "errors": errors, "warnings": warnings, "model_count": len(referenced), + "profile": profile_name, } def explain_route( diff --git a/tests/test_hardware_profiles.py b/tests/test_hardware_profiles.py new file mode 100644 index 0000000..9e208bd --- /dev/null +++ b/tests/test_hardware_profiles.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock + +from tests.common import ROOT # noqa: F401 + +from shared_tools.hardware_profiles import ( + CONFIG_RELATIVE_PATH, + ENV_HARDWARE_PROFILE, + hardware_profile_to_router_policy, + hardware_profile_to_scheduler, + resolve_active_hardware_profile, +) +from shared_tools.inference_router import InferenceRouter + + +CUSTOM_PROFILE_NAME = "test_cuda_profile" + + +def _write_custom_profiles(repo_root: Path) -> None: + config_path = repo_root / CONFIG_RELATIVE_PATH + config_path.parent.mkdir(parents=True, exist_ok=True) + config_path.write_text( + json.dumps( + { + "default_profile": "8gb_vram_16gb_ram", + "profiles": { + "8gb_vram_16gb_ram": { + "name": "8gb_vram_16gb_ram", + "hardware": { + "system_ram_gb": 16, + "gpu_backend": "generic", + "gpu_vram_gb": 8, + "unified_memory": False, + }, + "scheduler": { + "max_context_tokens": 4096, + "warning_context_tokens": 4096, + "max_stage_context_tokens": 1800, + "max_parallel_models": 1, + "max_active_model_calls": 1, + "on_deck_depth": 1, + "warm_depth": 1, + "allow_neural_prefetch": True, + }, + "model_policy": { + "normal_max_b": 9, + "heavy_max_b": 14, + "premium_min_b": 24, + "allow_premium": False, + "premium_requires_manual": True, + "allow_14b_with_warning": True, + "reject_heavier_fallbacks": False, + }, + "lane_caps": {}, + "validation": {"startup_mode": "warn"}, + }, + CUSTOM_PROFILE_NAME: { + "name": CUSTOM_PROFILE_NAME, + "display_name": "Synthetic CUDA profile", + "description": "Synthetic test profile for hardware policy conversion.", + "hardware": { + "system_ram_gb": 24, + "gpu_backend": "cuda", + "gpu_vram_gb": 6, + "unified_memory": False, + }, + "scheduler": { + "max_context_tokens": 6144, + "warning_context_tokens": 4096, + "max_stage_context_tokens": 1600, + "max_parallel_models": 1, + "max_active_model_calls": 1, + "on_deck_depth": 1, + "warm_depth": 0, + "allow_neural_prefetch": False, + }, + "inference": { + "preferred_backends": ["llama.cpp", "ollama"], + "default_keep_alive": "2m", + "heavy_keep_alive": "0", + "release_heavy_after_call": True, + "max_loaded_models": 1, + }, + "model_policy": { + "normal_max_b": 9, + "heavy_max_b": 14, + "premium_min_b": 24, + "allow_premium": False, + "premium_requires_manual": True, + "allow_14b_with_warning": True, + "reject_heavier_fallbacks": False, + }, + "lane_caps": { + "chat_layer": { + "max_context_tokens": 4096, + "max_parallel_agents": 1, + } + }, + "validation": {"startup_mode": "warn"}, + }, + }, + }, + indent=2, + ), + encoding="utf-8", + ) + + +class HardwareProfileTests(unittest.TestCase): + def setUp(self) -> None: + self._previous_profile = os.environ.pop(ENV_HARDWARE_PROFILE, None) + self.tmp = tempfile.TemporaryDirectory(prefix="hardware_profiles_") + self.repo_root = Path(self.tmp.name) + _write_custom_profiles(self.repo_root) + + def tearDown(self) -> None: + os.environ.pop(ENV_HARDWARE_PROFILE, None) + if self._previous_profile is not None: + os.environ[ENV_HARDWARE_PROFILE] = self._previous_profile + self.tmp.cleanup() + + def test_active_profile_resolver_returns_default_when_env_is_unset(self) -> None: + profile = resolve_active_hardware_profile(self.repo_root) + + self.assertEqual(profile["name"], "8gb_vram_16gb_ram") + + def test_active_profile_resolver_returns_named_profile_from_env(self) -> None: + os.environ[ENV_HARDWARE_PROFILE] = CUSTOM_PROFILE_NAME + + profile = resolve_active_hardware_profile(self.repo_root) + + self.assertEqual(profile["name"], CUSTOM_PROFILE_NAME) + self.assertEqual(profile["hardware"]["system_ram_gb"], 24) + self.assertEqual(profile["scheduler"]["warm_depth"], 0) + + def test_unknown_profile_falls_back_to_default_with_warning(self) -> None: + profile = resolve_active_hardware_profile(self.repo_root, "not_a_real_profile") + + self.assertEqual(profile["name"], "8gb_vram_16gb_ram") + self.assertTrue(profile.get("_resolution_warnings")) + + def test_named_profile_converts_to_scheduler_profile(self) -> None: + profile = resolve_active_hardware_profile(self.repo_root, CUSTOM_PROFILE_NAME) + + scheduler = hardware_profile_to_scheduler(profile) + + self.assertEqual(scheduler.ram_gb, 24.0) + self.assertEqual(scheduler.vram_gb, 6.0) + self.assertEqual(scheduler.max_context_tokens, 6144) + self.assertEqual(scheduler.max_parallel_models, 1) + self.assertEqual(scheduler.warm_depth, 0) + + def test_named_profile_converts_to_router_policy(self) -> None: + profile = resolve_active_hardware_profile(self.repo_root, CUSTOM_PROFILE_NAME) + + policy = hardware_profile_to_router_policy(profile) + + self.assertEqual(policy["max_context"], 6144) + self.assertEqual(policy["warning_context"], 4096) + self.assertEqual(policy["max_concurrency"], 1) + self.assertFalse(policy["allow_premium"]) + + def test_router_fit_uses_named_policy_when_explicit(self) -> None: + router = InferenceRouter(self.repo_root) + policy = hardware_profile_to_router_policy( + resolve_active_hardware_profile(self.repo_root, CUSTOM_PROFILE_NAME) + ) + + normal = router.estimate_fit("qwen3:8b", 4096, profile=policy) + premium = router.estimate_fit("qwen3:30b-a3b-q4_K_M", 4096, profile=policy) + + self.assertTrue(normal["fits"]) + self.assertEqual(normal["profile"], CUSTOM_PROFILE_NAME) + self.assertFalse(premium["fits"]) + self.assertEqual(premium["profile"], CUSTOM_PROFILE_NAME) + + def test_router_fit_uses_env_profile_without_explicit_profile(self) -> None: + os.environ[ENV_HARDWARE_PROFILE] = CUSTOM_PROFILE_NAME + router = InferenceRouter(self.repo_root) + + fit = router.estimate_fit("qwen3:8b", 6144, concurrency=1) + + self.assertTrue(fit["fits"]) + self.assertEqual(fit["profile"], CUSTOM_PROFILE_NAME) + + def test_validate_config_warns_when_route_exceeds_profile_lane_caps(self) -> None: + router = InferenceRouter(self.repo_root) + router._routing = { + "chat_layer": { + "model": "qwen3:8b", + "num_ctx": 6144, + "parallel_agents": 2, + }, + "premium_models": [], + } + router.list_backends = MagicMock(return_value=[]) + policy = hardware_profile_to_router_policy( + resolve_active_hardware_profile(self.repo_root, CUSTOM_PROFILE_NAME) + ) + + report = router.validate_config(check_remote=False, profile=policy) + + self.assertTrue(report["ok"]) + self.assertEqual(report["profile"], CUSTOM_PROFILE_NAME) + self.assertTrue(any("lane cap" in item for item in report["warnings"])) + + +if __name__ == "__main__": + unittest.main()