diff --git a/AGENTS.md b/AGENTS.md index 0d0379f5f..65156b2c0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -200,6 +200,7 @@ Each domain has exactly **one** write-owning module (or one tightly-scoped famil | Schedules (`config/schedules.json`) | `solstone/think/schedule_config.py` | | Push devices (`config/push_devices.json`) | `solstone/think/push/devices.py` | | Local inference operational telemetry (`health/local-inference/YYYYMMDD.jsonl`) | `solstone/think/providers/local_admission.py` | +| Parakeet server placement record (`health/parakeet-cpp.placement`) | `solstone/think/providers/parakeet_server.py` | | Hosted backup binding (`backup/hosted/binding.json`) | `solstone/think/backup/hosted.py` | | Convey config (`config/convey.json`) | `solstone/convey/config.py` + `solstone/think/facets.py` | | Chat config (`config/chat.json`) | `solstone/apps/chat/config.py` | diff --git a/solstone/observe/transcribe/_parakeet_cpp.py b/solstone/observe/transcribe/_parakeet_cpp.py index d0f937980..696c1c207 100644 --- a/solstone/observe/transcribe/_parakeet_cpp.py +++ b/solstone/observe/transcribe/_parakeet_cpp.py @@ -219,9 +219,10 @@ def get_model_info(config: dict) -> dict: """Return parakeet.cpp model metadata for transcript JSONL headers.""" _require_linux() device = _validate_config(config) + placement = parakeet_server.read_parakeet_placement() return { "model": parakeet_readiness.PARAKEET_CPP_MODEL_FILENAME, - "device": device, + "device": placement or device, "compute_type": _COMPUTE_TYPE, "per_word_confidence": True, } diff --git a/solstone/observe/transcribe/failure-and-telemetry.md b/solstone/observe/transcribe/failure-and-telemetry.md index 6210747cd..4bd7ff274 100644 --- a/solstone/observe/transcribe/failure-and-telemetry.md +++ b/solstone/observe/transcribe/failure-and-telemetry.md @@ -101,7 +101,7 @@ One event name, five outcomes. Every attempt emits exactly one event. | `reason` | machine reason (table above) | deferred, failed | | `error` | exception **type name** — never the message (see below) | failed | | `backend` | STT backend name (`parakeet-cpp`, `gemini`, …) | whenever resolved | -| `device` | resolved device (`auto` / `cpu`) | whenever known (see below) | +| `device` | resolved placement (`cpu` / `gpu`) when a placement record exists; configured device otherwise | whenever known (see below) | | `model` | model filename | success, and failures after the backend reported it | | `audio_seconds` | original decoded length, 1 dp | whenever decoded | | `reduced_seconds` | length after silence-trimming, 1 dp | when reduction ran | @@ -171,8 +171,9 @@ right now. - **`model` on deferred events.** `get_model_info()` is cheap for the parakeet-cpp and cloud backends, but on Apple Silicon it shells out to the CoreML helper (`--version`, 10 s timeout). Rather than hoist a subprocess probe onto a path whose whole point is - *not* to do expensive work, deferred events omit `model`. `device` is still reported - when the config names one. + *not* to do expensive work, deferred events omit `model`. `device` reports the + supervisor placement for parakeet-cpp when that record exists, and otherwise falls + back to the configured value when the config names one. ## Rollback diff --git a/solstone/think/backup/engine.py b/solstone/think/backup/engine.py index 84b1af4d6..c4d2af5b7 100644 --- a/solstone/think/backup/engine.py +++ b/solstone/think/backup/engine.py @@ -59,6 +59,7 @@ BACKUP_EXCLUDES = ( ".tmp*", "supervisor.ready", "supervisor.start_time", + "parakeet-cpp.placement", "scheduler.json", "talents.json", "agents.json", diff --git a/solstone/think/check.py b/solstone/think/check.py index fabad63fe..6aadd1b45 100644 --- a/solstone/think/check.py +++ b/solstone/think/check.py @@ -178,11 +178,19 @@ def _render_nodes_present_but_inaccessible() -> bool: def _linux_gpu_check(probe: object) -> FitCheck: try: from solstone.think.providers import local_cuda, local_vulkan, memory + from solstone.think.providers.parakeet_placement import cpu_placement_suffix - if not bool(getattr(probe, "detected")): + try: devices = local_vulkan.detect_gpus() probe_ok = local_vulkan.gpu_probe_ok() - selected = local_vulkan.select_device(devices) if probe_ok else None + except Exception: + if not bool(getattr(probe, "detected")): + raise + # Fail toward current behavior: Vulkan failure on NVIDIA yields no placement line. + devices = [] + probe_ok = False + selected = local_vulkan.select_device(devices) if probe_ok else None + if not bool(getattr(probe, "detected")): inaccessible = ( not probe_ok or selected is None ) and _render_nodes_present_but_inaccessible() @@ -223,7 +231,18 @@ def _linux_gpu_check(probe: object) -> FitCheck: return FitCheck( "gpu", "ok", - f"Vulkan GPU {selected.name} with {memory.gb_label(vram_bytes)} GB", + ( + f"Vulkan GPU {selected.name} with {memory.gb_label(vram_bytes)} GB" + + cpu_placement_suffix( + devices=devices, + selected=selected, + local_vulkan=local_vulkan, + unified_memory=False, + # sol check runs before install and cannot rely on journal + # config; use the bundled-brain default for this advisory. + brain_lane_active=True, + ) + ), required_bytes=GPU_MIN_BYTES, available_bytes=vram_bytes, ) @@ -257,6 +276,18 @@ def _linux_gpu_check(probe: object) -> FitCheck: detail = f"NVIDIA GPU with {memory.gb_label(gpu_bytes)} GB" if getattr(probe, "memory_source") == local_cuda.MEMORY_SOURCE_SYSTEM_AVAILABLE: detail = f"{detail} (unified memory)" + detail += cpu_placement_suffix( + devices=devices, + selected=selected, + local_vulkan=local_vulkan, + unified_memory=( + getattr(probe, "memory_source") + == local_cuda.MEMORY_SOURCE_SYSTEM_AVAILABLE + ), + # sol check runs before install and cannot rely on journal config; use + # the bundled-brain default for this advisory. + brain_lane_active=True, + ) return FitCheck( "gpu", "ok", diff --git a/solstone/think/providers/fit_report.py b/solstone/think/providers/fit_report.py index 380ea9979..3f6662cda 100644 --- a/solstone/think/providers/fit_report.py +++ b/solstone/think/providers/fit_report.py @@ -74,6 +74,8 @@ def build_local_fit_report(model_id: str) -> FitReport: probe = None choice = None + devices: list[Any] = [] + brain_lane_active = True if sys.platform.startswith("linux"): probe = local_cuda.probe_nvidia_gpu() choice = local_cuda.select_local_backend( @@ -86,6 +88,16 @@ def build_local_fit_report(model_id: str) -> FitReport: if choice.backend == "cuda" else "llama-server tarball" ) + devices = local_vulkan.detect_gpus() + try: + from solstone.think.models import is_local_provider_needed + from solstone.think.providers.local_endpoint import resolve_local_endpoint + + brain_lane_active = ( + is_local_provider_needed() and resolve_local_endpoint().is_bundled + ) + except Exception: + brain_lane_active = True else: unknown_server = "llama-server tarball" @@ -103,7 +115,13 @@ def build_local_fit_report(model_id: str) -> FitReport: if sys.platform.startswith("linux") and probe is not None and choice is not None: checks.append( - _local_gpu_check(probe, choice, local_vulkan.detect_gpus(), local_vulkan) + _local_gpu_check( + probe, + choice, + devices, + local_vulkan, + brain_lane_active=brain_lane_active, + ) ) return FitReport(artifact="local provider artifacts", checks=tuple(checks)) @@ -388,8 +406,11 @@ def _local_gpu_check( choice: Any, devices: list[Any], local_vulkan: Any, + *, + brain_lane_active: bool, ) -> FitCheck: from solstone.think.providers import local_cuda + from solstone.think.providers.parakeet_placement import cpu_placement_suffix backend = getattr(choice, "backend") reason = getattr(choice, "reason") @@ -401,6 +422,14 @@ def _local_gpu_check( ) memory_source = getattr(probe, "memory_source") + selected = local_vulkan.select_device(devices) + placement_suffix = cpu_placement_suffix( + devices=devices, + selected=selected, + local_vulkan=local_vulkan, + unified_memory=memory_source == local_cuda.MEMORY_SOURCE_SYSTEM_AVAILABLE, + brain_lane_active=brain_lane_active, + ) if memory_source == local_cuda.MEMORY_SOURCE_UNAVAILABLE: return FitCheck( "gpu", @@ -412,6 +441,7 @@ def _local_gpu_check( detail = f"CUDA backend selected: {reason}" if memory_source == local_cuda.MEMORY_SOURCE_SYSTEM_AVAILABLE: detail = f"{detail}; GPU tiering memory uses system MemAvailable" + detail = f"{detail}{placement_suffix}" return FitCheck("gpu", "ok", detail) probe_ok = local_vulkan.gpu_probe_ok() @@ -421,7 +451,6 @@ def _local_gpu_check( "unknown", f"Vulkan GPU probe did not complete; resolved backend is {backend}: {reason}", ) - selected = local_vulkan.select_device(devices) if selected is None: return FitCheck( "gpu", @@ -431,7 +460,10 @@ def _local_gpu_check( return FitCheck( "gpu", "ok", - f"Vulkan GPU selected: {selected.name}; resolved backend is {backend}: {reason}", + ( + f"Vulkan GPU selected: {selected.name}; resolved backend is {backend}: " + f"{reason}{placement_suffix}" + ), ) diff --git a/solstone/think/providers/local_server.py b/solstone/think/providers/local_server.py index 653c7cda2..e5b66e525 100644 --- a/solstone/think/providers/local_server.py +++ b/solstone/think/providers/local_server.py @@ -38,19 +38,28 @@ class ServerTier: context_tokens: int parallel_slots: int prompt_cache_mib: int + resident_mib: int | None # Tunable estimates — keep all tier values in these two instances; do not # scatter literals elsewhere. The threshold is the only other tunable. _CAPABLE_TIER_MIN_VRAM_MIB = 16000 +# ``resident_mib`` is measured floor-tier brain residency under production +# launch args. ``None`` on capable is load-bearing: unmeasured residency means +# the co-location placement predicate can never fire at >=16 GiB. _CAPABLE_TIER = ServerTier( - name="capable", context_tokens=32768, parallel_slots=2, prompt_cache_mib=2048 + name="capable", + context_tokens=32768, + parallel_slots=2, + prompt_cache_mib=2048, + resident_mib=None, ) _FLOOR_TIER = ServerTier( name="floor", context_tokens=LOCAL_MIN_CONTEXT_TOKENS, parallel_slots=1, prompt_cache_mib=0, + resident_mib=4541, ) # COPY REVIEW: placeholder owner-facing copy; founder-gated before ship. diff --git a/solstone/think/providers/parakeet_placement.py b/solstone/think/providers/parakeet_placement.py new file mode 100644 index 000000000..a548dabf5 --- /dev/null +++ b/solstone/think/providers/parakeet_placement.py @@ -0,0 +1,199 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Pure placement decision for supervised parakeet.cpp co-location.""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any + +from solstone.think.providers.local_server import select_server_tier + +# Measured worst-case ordinary segment residency: 300 s is the observer-contract +# cap. The 1024 MiB margin covers display framebuffer, compositor allocations, +# driver overhead, and allocator fragmentation on the same monitor-driving GPU. +# It intentionally uses 1024, not 512, to put 10 GiB cards (10240 MiB) on the +# CPU side of the 10587 MiB threshold: without margin, only 677 MiB would remain +# after the two measured residents, which is not reliable display-attached +# operating slack. +PARAKEET_WORST_CASE_MIB = 5022 +CO_FIT_MARGIN_MIB = 1024 + +CPU_PLACEMENT_COPY = ( + "sol thinks on your GPU; transcription runs on your CPU on this machine" +) +_DISCRETE_CLASSIFICATION = "discrete" + + +@dataclass(frozen=True) +class ParakeetPlacementDecision: + force_cpu: bool + reason_code: str + tier_name: str | None + tier_resident_mib: int | None + parakeet_worst_case_mib: int + margin_mib: int + required_mib: int | None + vram_mib: int | None + + +def _decision( + *, + force_cpu: bool, + reason_code: str, + tier_name: str | None, + tier_resident_mib: int | None, + required_mib: int | None, + vram_mib: int | None, +) -> ParakeetPlacementDecision: + return ParakeetPlacementDecision( + force_cpu=force_cpu, + reason_code=reason_code, + tier_name=tier_name, + tier_resident_mib=tier_resident_mib, + parakeet_worst_case_mib=PARAKEET_WORST_CASE_MIB, + margin_mib=CO_FIT_MARGIN_MIB, + required_mib=required_mib, + vram_mib=vram_mib, + ) + + +def is_discrete(device: Any, local_vulkan: Any) -> bool: + """Return whether a pre-enumerated Vulkan device is classified discrete.""" + return local_vulkan.classify(device) == _DISCRETE_CLASSIFICATION + + +def discrete_hardware_gpu_count( + devices: Sequence[Any], + local_vulkan: Any, +) -> int: + """Count hardware GPUs classified discrete from an existing Vulkan enumeration.""" + return sum( + 1 + for device in devices + if local_vulkan.is_hardware_device(device) and is_discrete(device, local_vulkan) + ) + + +def cpu_placement_suffix( + *, + devices: Sequence[Any], + selected: Any | None, + local_vulkan: Any, + unified_memory: bool, + brain_lane_active: bool, +) -> str: + """Return the advisory suffix when auto-placement resolves STT to CPU.""" + if selected is None: + return "" + decision = decide_parakeet_auto_placement( + vram_mib=getattr(selected, "vram_mib", None), + selected_device_is_discrete=is_discrete(selected, local_vulkan), + discrete_hardware_gpu_count=discrete_hardware_gpu_count(devices, local_vulkan), + unified_memory=unified_memory, + brain_lane_active=brain_lane_active, + ) + return f"; {CPU_PLACEMENT_COPY}" if decision.force_cpu else "" + + +def decide_parakeet_auto_placement( + vram_mib: int | None, + selected_device_is_discrete: bool, + discrete_hardware_gpu_count: int, + unified_memory: bool, + brain_lane_active: bool, +) -> ParakeetPlacementDecision: + """Return whether parakeet.cpp auto-placement must use CPU. + + This is intentionally pure: callers provide probe/config facts, and this + function performs only tier selection and arithmetic. + """ + if not brain_lane_active: + return _decision( + force_cpu=False, + reason_code="brain_lane_inactive", + tier_name=None, + tier_resident_mib=None, + required_mib=None, + vram_mib=vram_mib, + ) + if not selected_device_is_discrete: + return _decision( + force_cpu=False, + reason_code="selected_device_not_discrete", + tier_name=None, + tier_resident_mib=None, + required_mib=None, + vram_mib=vram_mib, + ) + if discrete_hardware_gpu_count != 1: + return _decision( + force_cpu=False, + reason_code="discrete_gpu_count_not_one", + tier_name=None, + tier_resident_mib=None, + required_mib=None, + vram_mib=vram_mib, + ) + if unified_memory: + return _decision( + force_cpu=False, + reason_code="unified_memory", + tier_name=None, + tier_resident_mib=None, + required_mib=None, + vram_mib=vram_mib, + ) + if vram_mib is None: + return _decision( + force_cpu=False, + reason_code="vram_unknown", + tier_name=None, + tier_resident_mib=None, + required_mib=None, + vram_mib=None, + ) + + tier = select_server_tier(vram_mib) + if tier.resident_mib is None: + return _decision( + force_cpu=False, + reason_code="tier_residency_unmeasured", + tier_name=tier.name, + tier_resident_mib=None, + required_mib=None, + vram_mib=vram_mib, + ) + + required_mib = tier.resident_mib + PARAKEET_WORST_CASE_MIB + CO_FIT_MARGIN_MIB + if vram_mib < required_mib: + return _decision( + force_cpu=True, + reason_code="co_location_requires_cpu", + tier_name=tier.name, + tier_resident_mib=tier.resident_mib, + required_mib=required_mib, + vram_mib=vram_mib, + ) + return _decision( + force_cpu=False, + reason_code="co_location_fits_gpu", + tier_name=tier.name, + tier_resident_mib=tier.resident_mib, + required_mib=required_mib, + vram_mib=vram_mib, + ) + + +__all__ = [ + "CO_FIT_MARGIN_MIB", + "CPU_PLACEMENT_COPY", + "PARAKEET_WORST_CASE_MIB", + "ParakeetPlacementDecision", + "cpu_placement_suffix", + "decide_parakeet_auto_placement", + "discrete_hardware_gpu_count", + "is_discrete", +] diff --git a/solstone/think/providers/parakeet_server.py b/solstone/think/providers/parakeet_server.py index 4c35d8735..4555d6091 100644 --- a/solstone/think/providers/parakeet_server.py +++ b/solstone/think/providers/parakeet_server.py @@ -6,16 +6,19 @@ from __future__ import annotations from dataclasses import dataclass +from pathlib import Path from solstone.think import parakeet_readiness from solstone.think.providers.parakeet_install import ParakeetProviderError -from solstone.think.utils import read_service_port +from solstone.think.utils import get_journal, read_service_port STATE_READY = "ready" STATE_FAILED = "failed" _HOST = "127.0.0.1" _SERVICE_NAME = "parakeet-cpp" +_PLACEMENT_FILE = "parakeet-cpp.placement" +_VALID_PLACEMENTS = {"cpu", "gpu"} class ParakeetServerNotReady(ParakeetProviderError): @@ -43,6 +46,33 @@ def _base_url(port: int) -> str: return f"http://{_HOST}:{port}" +def _placement_path() -> Path: + return Path(get_journal()) / "health" / _PLACEMENT_FILE + + +def write_parakeet_placement(device: str) -> None: + """Persist the resolved parakeet.cpp serving placement for telemetry.""" + if device not in _VALID_PLACEMENTS: + raise ValueError(f"invalid parakeet placement: {device!r}") + path = _placement_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(device) + + +def read_parakeet_placement() -> str | None: + """Read the resolved parakeet.cpp serving placement, if valid.""" + try: + device = _placement_path().read_text().strip() + except FileNotFoundError: + return None + return device if device in _VALID_PLACEMENTS else None + + +def clear_parakeet_placement() -> None: + """Remove any stale parakeet.cpp serving placement record.""" + _placement_path().unlink(missing_ok=True) + + def _probe_health(port: int, timeout_s: float = 1.0) -> tuple[str, str | None]: import httpx @@ -88,6 +118,9 @@ __all__ = [ "STATE_READY", "ParakeetServerInfo", "ParakeetServerNotReady", + "clear_parakeet_placement", "connect", "probe_state", + "read_parakeet_placement", + "write_parakeet_placement", ] diff --git a/solstone/think/supervisor.py b/solstone/think/supervisor.py index d94c18e61..25b977773 100644 --- a/solstone/think/supervisor.py +++ b/solstone/think/supervisor.py @@ -64,6 +64,7 @@ from solstone.think.processing import ( evaluate_drain_gate, load_processing_settings, ) +from solstone.think.providers import parakeet_server from solstone.think.providers.memory import read_available_bytes from solstone.think.providers.mlx_server import MLX_SERVER_PROCESS_NAME from solstone.think.readiness import START_TIME_TOLERANCE_S, clear_ready, signal_ready @@ -2420,12 +2421,19 @@ def start_local_server() -> RunnerManagedProcess | None: def start_parakeet_server() -> RunnerManagedProcess | None: """Launch the supervisor-owned parakeet-server when STT opts into it.""" + parakeet_server.clear_parakeet_placement() if not linux_stt_uses_parakeet_cpp(): return None from solstone.think.providers import local_vulkan, parakeet_install + from solstone.think.providers.parakeet_placement import ( + decide_parakeet_auto_placement, + discrete_hardware_gpu_count, + is_discrete, + ) config_device = _configured_parakeet_device() + effective_device = config_device selected = None if config_device == "auto": devices = local_vulkan.detect_gpus() @@ -2440,8 +2448,42 @@ def start_parakeet_server() -> RunnerManagedProcess | None: else "none" ), ) + selected_is_discrete = selected is not None and is_discrete( + selected, local_vulkan + ) + if selected is not None and selected_is_discrete: + from solstone.think.providers import local_cuda + from solstone.think.providers.local_endpoint import resolve_local_endpoint + + discrete_count = discrete_hardware_gpu_count(devices, local_vulkan) + probe = local_cuda.probe_nvidia_gpu() + decision = decide_parakeet_auto_placement( + vram_mib=selected.vram_mib, + selected_device_is_discrete=selected_is_discrete, + discrete_hardware_gpu_count=discrete_count, + unified_memory=( + probe.memory_source == local_cuda.MEMORY_SOURCE_SYSTEM_AVAILABLE + ), + brain_lane_active=( + is_local_provider_needed() and resolve_local_endpoint().is_bundled + ), + ) + if decision.force_cpu: + logging.info( + "parakeet-server auto placement resolved to CPU: " + "tier=%s tier_resident_mib=%s " + "parakeet_worst_case_mib=%d margin_mib=%d " + "required_mib=%s gpu_vram_mib=%s placement=cpu", + decision.tier_name, + decision.tier_resident_mib, + decision.parakeet_worst_case_mib, + decision.margin_mib, + decision.required_mib, + decision.vram_mib, + ) + effective_device = "cpu" - plan = resolve_parakeet_server_launch_plan(config_device, selected) + plan = resolve_parakeet_server_launch_plan(effective_device, selected) try: binary_path, gguf_path = parakeet_install.ensure_artifacts_installed( plan.binary_backend @@ -2465,6 +2507,9 @@ def start_parakeet_server() -> RunnerManagedProcess | None: env, ) if status == "ready": + parakeet_server.write_parakeet_placement( + "gpu" if plan.binary_backend == "vulkan" else "cpu" + ) return managed if plan.binary_backend == "vulkan" and status in {"crashed", "timeout"}: @@ -2499,6 +2544,7 @@ def start_parakeet_server() -> RunnerManagedProcess | None: "continuing startup", PARAKEET_SERVER_READY_TIMEOUT_S, ) + parakeet_server.write_parakeet_placement("cpu") return cpu_managed if plan.binary_backend == "cpu": @@ -2514,8 +2560,12 @@ def start_parakeet_server() -> RunnerManagedProcess | None: "continuing startup", PARAKEET_SERVER_READY_TIMEOUT_S, ) + parakeet_server.write_parakeet_placement("cpu") return managed + parakeet_server.write_parakeet_placement( + "gpu" if plan.binary_backend == "vulkan" else "cpu" + ) return managed diff --git a/tests/test_backup_engine.py b/tests/test_backup_engine.py index 7473147dd..6afab0338 100644 --- a/tests/test_backup_engine.py +++ b/tests/test_backup_engine.py @@ -245,6 +245,8 @@ def test_run_backup_unlocks_then_calls_restic_with_expected_argv( "--exclude", "supervisor.start_time", "--exclude", + "parakeet-cpp.placement", + "--exclude", "scheduler.json", "--exclude", "talents.json", diff --git a/tests/test_check.py b/tests/test_check.py index 3a85ab4c5..0600b287b 100644 --- a/tests/test_check.py +++ b/tests/test_check.py @@ -12,6 +12,16 @@ from solstone.think import utils as think_utils from solstone.think.providers import local_cuda, local_vulkan, memory GB = 1024**3 +PLACEMENT_LINE = ( + "sol thinks on your GPU; transcription runs on your CPU on this machine" +) + + +@pytest.fixture(autouse=True) +def _reset_vulkan_detect_cache(): + local_vulkan.reset_detect_cache() + yield + local_vulkan.reset_detect_cache() def _patch_platform( @@ -50,6 +60,10 @@ def _patch_linux_ok(monkeypatch: pytest.MonkeyPatch) -> None: _patch_platform(monkeypatch) _patch_memory(monkeypatch) _patch_disk(monkeypatch) + monkeypatch.setattr( + local_vulkan, "detect_gpus", lambda: [_vulkan_device(vram_mib=24576)] + ) + monkeypatch.setattr(local_vulkan, "gpu_probe_ok", lambda: True) def _nvidia_probe( @@ -90,12 +104,13 @@ def _vulkan_device( *, index: int = 0, name: str = "Vulkan GPU", + device_type: int = local_vulkan.VK_TYPE_DISCRETE, vram_mib: int = 8192, ) -> local_vulkan.VulkanDevice: return local_vulkan.VulkanDevice( index=index, name=name, - device_type=local_vulkan.VK_TYPE_DISCRETE, + device_type=device_type, vram_mib=vram_mib, ) @@ -170,6 +185,90 @@ def test_linux_nvidia_vulkan_backend_recommendation( assert "solstone-journal-cuda" not in output +def test_linux_nvidia_small_single_discrete_mentions_cpu_transcription( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _patch_linux_ok(monkeypatch) + monkeypatch.setattr( + local_cuda, + "probe_nvidia_gpu", + lambda: _nvidia_probe(vram_mib=6144), + ) + monkeypatch.setattr( + local_vulkan, "detect_gpus", lambda: [_vulkan_device(vram_mib=6144)] + ) + + result = check.build_check_report() + + gpu = _checks(result)["gpu"] + assert gpu.severity == "ok" + assert gpu.detail == f"NVIDIA GPU with 6 GB; {PLACEMENT_LINE}" + + +@pytest.mark.parametrize( + ("probe", "devices", "probe_ok"), + [ + (_nvidia_probe(vram_mib=12288), [_vulkan_device(vram_mib=12288)], True), + ( + _nvidia_probe(vram_mib=6144), + [ + _vulkan_device(index=0, vram_mib=6144), + _vulkan_device(index=1, vram_mib=6144), + ], + True, + ), + ( + _nvidia_probe( + vram_mib=6144, + memory_source=local_cuda.MEMORY_SOURCE_SYSTEM_AVAILABLE, + ), + [_vulkan_device(vram_mib=6144)], + True, + ), + (_nvidia_probe(vram_mib=16384), [_vulkan_device(vram_mib=16384)], True), + (_nvidia_probe(vram_mib=6144), [], False), + ], +) +def test_linux_nvidia_cpu_transcription_line_absent_outside_predicate( + monkeypatch: pytest.MonkeyPatch, + probe: local_cuda.NvidiaProbe, + devices: list[local_vulkan.VulkanDevice], + probe_ok: bool, +) -> None: + _patch_linux_ok(monkeypatch) + monkeypatch.setattr(local_cuda, "probe_nvidia_gpu", lambda: probe) + monkeypatch.setattr(local_vulkan, "detect_gpus", lambda: devices) + monkeypatch.setattr(local_vulkan, "gpu_probe_ok", lambda: probe_ok) + + result = check.build_check_report() + + gpu = _checks(result)["gpu"] + assert gpu.severity == "ok" + assert PLACEMENT_LINE not in gpu.detail + + +def test_linux_nvidia_vulkan_detection_exception_keeps_current_detail( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _patch_linux_ok(monkeypatch) + monkeypatch.setattr( + local_cuda, + "probe_nvidia_gpu", + lambda: _nvidia_probe(vram_mib=6144), + ) + monkeypatch.setattr( + local_vulkan, + "detect_gpus", + lambda: (_ for _ in ()).throw(RuntimeError("vulkan failed")), + ) + + result = check.build_check_report() + + gpu = _checks(result)["gpu"] + assert gpu.severity == "ok" + assert gpu.detail == "NVIDIA GPU with 6 GB" + + def test_linux_vulkan_ok( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture, diff --git a/tests/test_fit_report.py b/tests/test_fit_report.py index 6ccca7123..dc41d7751 100644 --- a/tests/test_fit_report.py +++ b/tests/test_fit_report.py @@ -7,10 +7,54 @@ from pathlib import Path import pytest -from solstone.think.providers import fit_report, local_install +from solstone.think.providers import fit_report, local_cuda, local_install, local_vulkan from solstone.think.providers.local import LocalProviderError from solstone.think.providers.memory import MemoryVerdict +PLACEMENT_LINE = ( + "sol thinks on your GPU; transcription runs on your CPU on this machine" +) + + +@pytest.fixture(autouse=True) +def _reset_vulkan_detect_cache(): + local_vulkan.reset_detect_cache() + yield + local_vulkan.reset_detect_cache() + + +def _nvidia_probe( + *, + vram_mib: int, + memory_source: str = local_cuda.MEMORY_SOURCE_NVIDIA_VRAM, +) -> local_cuda.NvidiaProbe: + return local_cuda.NvidiaProbe( + index=0, + compute_cap="sm_89", + driver_cuda_version=13, + vram_mib=vram_mib, + tiering_memory_mib=vram_mib, + memory_source=memory_source, + detected=True, + ) + + +def _vulkan_device( + *, + index: int = 0, + vram_mib: int, +) -> local_vulkan.VulkanDevice: + return local_vulkan.VulkanDevice( + index=index, + name=f"Test GPU {index}", + device_type=local_vulkan.VK_TYPE_DISCRETE, + vram_mib=vram_mib, + ) + + +def _choice(backend: str = "cuda") -> local_cuda.BackendChoice: + return local_cuda.BackendChoice(backend=backend, reason="test choice") + def test_overall_collapses_unknown_to_warning() -> None: report = fit_report.FitReport( @@ -37,6 +81,59 @@ def test_overall_blocked_wins() -> None: assert report.overall == "blocked" +def test_local_gpu_check_mentions_cpu_transcription_on_small_bundled_brain() -> None: + check = fit_report._local_gpu_check( + _nvidia_probe(vram_mib=6144), + _choice(), + [_vulkan_device(vram_mib=6144)], + local_vulkan, + brain_lane_active=True, + ) + + assert check.severity == "ok" + assert check.detail == f"CUDA backend selected: test choice; {PLACEMENT_LINE}" + + +@pytest.mark.parametrize( + ("probe", "devices", "brain_lane_active"), + [ + (_nvidia_probe(vram_mib=6144), [_vulkan_device(vram_mib=6144)], False), + ( + _nvidia_probe(vram_mib=6144), + [ + _vulkan_device(index=0, vram_mib=6144), + _vulkan_device(index=1, vram_mib=6144), + ], + True, + ), + ( + _nvidia_probe( + vram_mib=6144, + memory_source=local_cuda.MEMORY_SOURCE_SYSTEM_AVAILABLE, + ), + [_vulkan_device(vram_mib=6144)], + True, + ), + (_nvidia_probe(vram_mib=16384), [_vulkan_device(vram_mib=16384)], True), + ], +) +def test_local_gpu_check_omits_cpu_transcription_line_outside_predicate( + probe: local_cuda.NvidiaProbe, + devices: list[local_vulkan.VulkanDevice], + brain_lane_active: bool, +) -> None: + check = fit_report._local_gpu_check( + probe, + _choice(), + devices, + local_vulkan, + brain_lane_active=brain_lane_active, + ) + + assert check.severity == "ok" + assert PLACEMENT_LINE not in check.detail + + def test_disk_unknown_size_warns_when_known_size_fits( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_local.py b/tests/test_local.py index dd1bfbf45..5452e6904 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -2116,6 +2116,7 @@ def test_select_server_tier_vram_thresholds(): context_tokens=16384, parallel_slots=1, prompt_cache_mib=0, + resident_mib=4541, ), ), ( @@ -2125,6 +2126,7 @@ def test_select_server_tier_vram_thresholds(): context_tokens=16384, parallel_slots=1, prompt_cache_mib=0, + resident_mib=4541, ), ), ( @@ -2134,6 +2136,7 @@ def test_select_server_tier_vram_thresholds(): context_tokens=32768, parallel_slots=2, prompt_cache_mib=2048, + resident_mib=None, ), ), ( @@ -2143,6 +2146,7 @@ def test_select_server_tier_vram_thresholds(): context_tokens=32768, parallel_slots=2, prompt_cache_mib=2048, + resident_mib=None, ), ), ] @@ -2152,6 +2156,8 @@ def test_select_server_tier_vram_thresholds(): assert tier == expected assert tier.context_tokens >= 16384 assert tier.context_tokens > 0 + assert local_server._FLOOR_TIER.resident_mib == 4541 + assert local_server._CAPABLE_TIER.resident_mib is None @pytest.mark.parametrize( diff --git a/tests/test_parakeet_placement.py b/tests/test_parakeet_placement.py new file mode 100644 index 000000000..e5958833a --- /dev/null +++ b/tests/test_parakeet_placement.py @@ -0,0 +1,169 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import pytest + +from solstone.think.providers import local_vulkan +from solstone.think.providers.parakeet_placement import ( + CO_FIT_MARGIN_MIB, + PARAKEET_WORST_CASE_MIB, + cpu_placement_suffix, + decide_parakeet_auto_placement, + discrete_hardware_gpu_count, + is_discrete, +) + +PLACEMENT_LINE = ( + "sol thinks on your GPU; transcription runs on your CPU on this machine" +) + + +def _device( + *, + index: int = 0, + name: str = "Test GPU", + device_type: int = local_vulkan.VK_TYPE_DISCRETE, + vram_mib: int = 6144, +) -> local_vulkan.VulkanDevice: + return local_vulkan.VulkanDevice( + index=index, + name=name, + device_type=device_type, + vram_mib=vram_mib, + ) + + +def _decision( + vram_mib: int | None, + *, + selected_device_is_discrete: bool = True, + discrete_hardware_gpu_count: int = 1, + unified_memory: bool = False, + brain_lane_active: bool = True, +): + return decide_parakeet_auto_placement( + vram_mib=vram_mib, + selected_device_is_discrete=selected_device_is_discrete, + discrete_hardware_gpu_count=discrete_hardware_gpu_count, + unified_memory=unified_memory, + brain_lane_active=brain_lane_active, + ) + + +def test_floor_tier_small_cards_force_cpu() -> None: + decision = _decision(6144) + + assert decision.force_cpu is True + assert decision.reason_code == "co_location_requires_cpu" + assert decision.tier_name == "floor" + assert decision.tier_resident_mib == 4541 + assert decision.parakeet_worst_case_mib == PARAKEET_WORST_CASE_MIB + assert decision.margin_mib == CO_FIT_MARGIN_MIB + assert decision.required_mib == 10587 + assert decision.vram_mib == 6144 + + +@pytest.mark.parametrize("vram_mib", [10240, 11264, 12288]) +def test_margin_places_real_floor_tier_cards(vram_mib: int) -> None: + decision = _decision(vram_mib) + + assert decision.force_cpu is (vram_mib == 10240) + + +def test_capable_tier_unmeasured_residency_keeps_gpu() -> None: + decision = _decision(16000) + + assert decision.force_cpu is False + assert decision.reason_code == "tier_residency_unmeasured" + assert decision.tier_name == "capable" + assert decision.tier_resident_mib is None + assert decision.required_mib is None + + +@pytest.mark.parametrize( + ("kwargs", "reason_code"), + [ + ({"vram_mib": None}, "vram_unknown"), + ({"vram_mib": 6144, "brain_lane_active": False}, "brain_lane_inactive"), + ( + {"vram_mib": 6144, "discrete_hardware_gpu_count": 2}, + "discrete_gpu_count_not_one", + ), + ( + {"vram_mib": 6144, "selected_device_is_discrete": False}, + "selected_device_not_discrete", + ), + ({"vram_mib": 6144, "unified_memory": True}, "unified_memory"), + ], +) +def test_non_matching_inputs_keep_today(kwargs: dict, reason_code: str) -> None: + decision = _decision(**kwargs) + + assert decision.force_cpu is False + assert decision.reason_code == reason_code + + +def test_is_discrete_centralizes_vulkan_classification() -> None: + assert is_discrete(_device(), local_vulkan) is True + assert ( + is_discrete( + _device(device_type=local_vulkan.VK_TYPE_INTEGRATED), + local_vulkan, + ) + is False + ) + + +def test_discrete_hardware_gpu_count_ignores_integrated_and_software() -> None: + devices = [ + _device(index=0), + _device(index=1, device_type=local_vulkan.VK_TYPE_INTEGRATED), + _device(index=2, name="llvmpipe", device_type=local_vulkan.VK_TYPE_CPU), + ] + + assert discrete_hardware_gpu_count(devices, local_vulkan) == 1 + + +def test_cpu_placement_suffix_owns_joiner_and_copy() -> None: + selected = _device(vram_mib=6144) + + assert ( + cpu_placement_suffix( + devices=[selected], + selected=selected, + local_vulkan=local_vulkan, + unified_memory=False, + brain_lane_active=True, + ) + == f"; {PLACEMENT_LINE}" + ) + + +@pytest.mark.parametrize( + ("selected", "unified_memory", "brain_lane_active"), + [ + (None, False, True), + (_device(vram_mib=6144), True, True), + (_device(vram_mib=6144), False, False), + (_device(vram_mib=12288), False, True), + ], +) +def test_cpu_placement_suffix_absent_outside_predicate( + selected: local_vulkan.VulkanDevice | None, + unified_memory: bool, + brain_lane_active: bool, +) -> None: + devices = [selected] if selected is not None else [] + + assert ( + cpu_placement_suffix( + devices=devices, + selected=selected, + local_vulkan=local_vulkan, + unified_memory=unified_memory, + brain_lane_active=brain_lane_active, + ) + == "" + ) diff --git a/tests/test_parakeet_server.py b/tests/test_parakeet_server.py index 0896ef49c..a8217be0e 100644 --- a/tests/test_parakeet_server.py +++ b/tests/test_parakeet_server.py @@ -5,6 +5,7 @@ from __future__ import annotations import sys import types +from pathlib import Path import pytest @@ -24,6 +25,36 @@ def _install_fake_httpx(monkeypatch: pytest.MonkeyPatch, get): return fake_httpx +def test_placement_record_round_trips_and_clears( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + + assert parakeet_server.read_parakeet_placement() is None + parakeet_server.clear_parakeet_placement() + parakeet_server.write_parakeet_placement("gpu") + assert parakeet_server.read_parakeet_placement() == "gpu" + parakeet_server.write_parakeet_placement("cpu") + assert parakeet_server.read_parakeet_placement() == "cpu" + parakeet_server.clear_parakeet_placement() + assert parakeet_server.read_parakeet_placement() is None + + +def test_placement_record_validation( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + with pytest.raises(ValueError, match="invalid parakeet placement"): + parakeet_server.write_parakeet_placement("vulkan") + + path = tmp_path / "journal" / "health" / "parakeet-cpp.placement" + path.parent.mkdir(parents=True) + path.write_text("vulkan") + assert parakeet_server.read_parakeet_placement() is None + + def test_no_port_probe_failed_and_connect_not_ready(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(parakeet_server, "read_service_port", lambda _service: None) diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index 9f98543f8..9e55a0990 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -3363,10 +3363,18 @@ def test_log_context_assertion(caplog): from solstone.think.providers import local_server floor = local_server.ServerTier( - name="floor", context_tokens=16384, parallel_slots=1, prompt_cache_mib=0 + name="floor", + context_tokens=16384, + parallel_slots=1, + prompt_cache_mib=0, + resident_mib=4541, ) capable = local_server.ServerTier( - name="capable", context_tokens=32768, parallel_slots=2, prompt_cache_mib=2048 + name="capable", + context_tokens=32768, + parallel_slots=2, + prompt_cache_mib=2048, + resident_mib=None, ) with caplog.at_level(logging.INFO): diff --git a/tests/test_supervisor_parakeet.py b/tests/test_supervisor_parakeet.py index 790c96e9c..b6c6007bd 100644 --- a/tests/test_supervisor_parakeet.py +++ b/tests/test_supervisor_parakeet.py @@ -3,13 +3,26 @@ from __future__ import annotations +import logging from pathlib import Path from types import SimpleNamespace import pytest from solstone.think import supervisor -from solstone.think.providers import local_vulkan, parakeet_install, parakeet_server +from solstone.think.providers import ( + local_cuda, + local_vulkan, + parakeet_install, + parakeet_server, +) + + +@pytest.fixture(autouse=True) +def _reset_vulkan_detect_cache(): + local_vulkan.reset_detect_cache() + yield + local_vulkan.reset_detect_cache() class _FakeProcess: @@ -32,6 +45,70 @@ class _FakeManaged: self.cleanup_called = True +def _nvidia_probe( + *, + vram_mib: int, + memory_source: str = local_cuda.MEMORY_SOURCE_NVIDIA_VRAM, +) -> local_cuda.NvidiaProbe: + return local_cuda.NvidiaProbe( + index=0, + compute_cap="sm_75", + driver_cuda_version=13, + vram_mib=vram_mib, + tiering_memory_mib=vram_mib, + memory_source=memory_source, + detected=True, + ) + + +def _patch_ready_parakeet_launch( + monkeypatch, + launches: list[dict[str, object]], + *, + poll_sequence: list[tuple[int | None, int | None]] | None = None, +) -> list[tuple[str, int]]: + def fake_ensure(backend: str): + return Path(f"/tmp/{backend}/parakeet-server"), Path("/tmp/model.gguf") + + monkeypatch.setattr(parakeet_install, "ensure_artifacts_installed", fake_ensure) + monkeypatch.setattr(supervisor, "find_available_port", lambda: 45123) + ports: list[tuple[str, int]] = [] + monkeypatch.setattr( + supervisor, + "write_service_port", + lambda service, port: ports.append((service, port)), + ) + monkeypatch.setattr(supervisor, "parakeet_physical_thread_count", lambda: 6) + monkeypatch.setattr( + supervisor, "_parakeet_runtime_library_dirs", lambda: [Path("/parakeet/lib")] + ) + monkeypatch.setattr( + parakeet_server, "probe_state", lambda: (parakeet_server.STATE_READY, None) + ) + + sequence = poll_sequence or [(None, None)] + + def fake_launch_process( + name, cmd, *, restart=False, shutdown_timeout=15, ref=None, env=None + ): + index = min(len(launches), len(sequence) - 1) + poll_value, returncode = sequence[index] + managed = _FakeManaged(poll_value, returncode) + launches.append( + { + "name": name, + "cmd": cmd, + "restart": restart, + "env": env, + "managed": managed, + } + ) + return managed + + monkeypatch.setattr(supervisor, "_launch_process", fake_launch_process) + return ports + + def test_parakeet_server_is_sweepable_orphan_name() -> None: assert ( supervisor.PARAKEET_SERVER_PROCESS_NAME in supervisor._LOCAL_SERVER_PROCTITLES @@ -141,21 +218,31 @@ def test_with_library_path_prepends_dirs() -> None: def test_start_parakeet_server_vulkan_crash_falls_back_to_cpu( monkeypatch, + tmp_path, ) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) monkeypatch.delenv("GGML_VK_VISIBLE_DEVICES", raising=False) monkeypatch.delenv("LD_LIBRARY_PATH", raising=False) monkeypatch.setattr(supervisor.sys, "platform", "linux") monkeypatch.setattr(supervisor, "linux_stt_uses_parakeet_cpp", lambda: True) monkeypatch.setattr(supervisor, "_configured_parakeet_device", lambda: "auto") + monkeypatch.setattr(supervisor, "is_local_provider_needed", lambda: True) + monkeypatch.setattr( + "solstone.think.providers.local_endpoint.resolve_local_endpoint", + lambda: SimpleNamespace(is_bundled=True), + ) gpu = local_vulkan.VulkanDevice( 2, "NVIDIA Test GPU", local_vulkan.VK_TYPE_DISCRETE, - 8192, + 12288, ) monkeypatch.setattr(local_vulkan, "detect_gpus", lambda: [gpu]) monkeypatch.setattr(local_vulkan, "select_device", lambda devices: devices[0]) monkeypatch.setattr(local_vulkan, "classify", lambda _device: "discrete") + monkeypatch.setattr( + local_cuda, "probe_nvidia_gpu", lambda: _nvidia_probe(vram_mib=12288) + ) def fake_ensure(backend: str): return Path(f"/tmp/{backend}/parakeet-server"), Path("/tmp/model.gguf") @@ -218,6 +305,114 @@ def test_start_parakeet_server_vulkan_crash_falls_back_to_cpu( assert launches[0]["managed"].cleanup_called is True assert terminated[0][0] is launches[0]["managed"] assert ports == [("parakeet-cpp", 45123)] + assert parakeet_server.read_parakeet_placement() == "cpu" + + +def test_start_parakeet_server_forces_cpu_on_small_single_discrete_bundled_brain( + monkeypatch, + tmp_path, + caplog, +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + monkeypatch.delenv("GGML_VK_VISIBLE_DEVICES", raising=False) + monkeypatch.setattr(supervisor.sys, "platform", "linux") + monkeypatch.setattr(supervisor, "linux_stt_uses_parakeet_cpp", lambda: True) + monkeypatch.setattr(supervisor, "_configured_parakeet_device", lambda: "auto") + monkeypatch.setattr(supervisor, "is_local_provider_needed", lambda: True) + monkeypatch.setattr( + "solstone.think.providers.local_endpoint.resolve_local_endpoint", + lambda: SimpleNamespace(is_bundled=True), + ) + gpu = local_vulkan.VulkanDevice( + 2, + "NVIDIA Test GPU", + local_vulkan.VK_TYPE_DISCRETE, + 6144, + ) + monkeypatch.setattr(local_vulkan, "detect_gpus", lambda: [gpu]) + monkeypatch.setattr(local_vulkan, "select_device", lambda devices: devices[0]) + monkeypatch.setattr(local_vulkan, "classify", lambda _device: "discrete") + monkeypatch.setattr( + local_cuda, "probe_nvidia_gpu", lambda: _nvidia_probe(vram_mib=6144) + ) + launches: list[dict[str, object]] = [] + ports = _patch_ready_parakeet_launch(monkeypatch, launches) + + caplog.set_level(logging.INFO) + result = supervisor.start_parakeet_server() + + assert result is launches[0]["managed"] + assert len(launches) == 1 + assert launches[0]["cmd"][0] == "/tmp/cpu/parakeet-server" + assert "GGML_VK_VISIBLE_DEVICES" not in launches[0]["env"] + assert ports == [("parakeet-cpp", 45123)] + assert parakeet_server.read_parakeet_placement() == "cpu" + assert ( + "parakeet-server auto placement resolved to CPU: tier=floor " + "tier_resident_mib=4541 parakeet_worst_case_mib=5022 margin_mib=1024 " + "required_mib=10587 gpu_vram_mib=6144 placement=cpu" + ) in caplog.text + + +def test_start_parakeet_server_brain_lane_inactive_keeps_auto_vulkan( + monkeypatch, + tmp_path, +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + monkeypatch.setattr(supervisor.sys, "platform", "linux") + monkeypatch.setattr(supervisor, "linux_stt_uses_parakeet_cpp", lambda: True) + monkeypatch.setattr(supervisor, "_configured_parakeet_device", lambda: "auto") + monkeypatch.setattr(supervisor, "is_local_provider_needed", lambda: False) + gpu = local_vulkan.VulkanDevice( + 2, + "NVIDIA Test GPU", + local_vulkan.VK_TYPE_DISCRETE, + 6144, + ) + monkeypatch.setattr(local_vulkan, "detect_gpus", lambda: [gpu]) + monkeypatch.setattr(local_vulkan, "select_device", lambda devices: devices[0]) + monkeypatch.setattr(local_vulkan, "classify", lambda _device: "discrete") + monkeypatch.setattr( + local_cuda, "probe_nvidia_gpu", lambda: _nvidia_probe(vram_mib=6144) + ) + launches: list[dict[str, object]] = [] + _patch_ready_parakeet_launch(monkeypatch, launches) + + result = supervisor.start_parakeet_server() + + assert result is launches[0]["managed"] + assert len(launches) == 1 + assert launches[0]["cmd"][0] == "/tmp/vulkan/parakeet-server" + assert launches[0]["env"]["GGML_VK_VISIBLE_DEVICES"] == "2" + assert parakeet_server.read_parakeet_placement() == "gpu" + + +def test_start_parakeet_server_explicit_cpu_skips_auto_placement( + monkeypatch, + tmp_path, +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + monkeypatch.setattr(supervisor.sys, "platform", "linux") + monkeypatch.setattr(supervisor, "linux_stt_uses_parakeet_cpp", lambda: True) + monkeypatch.setattr(supervisor, "_configured_parakeet_device", lambda: "cpu") + monkeypatch.setattr( + local_vulkan, + "detect_gpus", + lambda: pytest.fail("explicit CPU should not enumerate Vulkan devices"), + ) + monkeypatch.setattr( + local_cuda, + "probe_nvidia_gpu", + lambda: pytest.fail("explicit CPU should not probe NVIDIA"), + ) + launches: list[dict[str, object]] = [] + _patch_ready_parakeet_launch(monkeypatch, launches) + + result = supervisor.start_parakeet_server() + + assert result is launches[0]["managed"] + assert launches[0]["cmd"][0] == "/tmp/cpu/parakeet-server" + assert parakeet_server.read_parakeet_placement() == "cpu" @pytest.mark.parametrize( @@ -287,7 +482,11 @@ def test_linux_stt_uses_parakeet_cpp_truth_table( assert supervisor.linux_stt_uses_parakeet_cpp() is expected -def test_start_parakeet_server_early_returns_for_non_linux(monkeypatch) -> None: +def test_start_parakeet_server_early_returns_for_non_linux( + monkeypatch, tmp_path +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + parakeet_server.write_parakeet_placement("gpu") monkeypatch.setattr(supervisor.sys, "platform", "darwin") monkeypatch.setattr(supervisor.platform, "machine", lambda: "arm64") monkeypatch.setattr( @@ -297,9 +496,14 @@ def test_start_parakeet_server_early_returns_for_non_linux(monkeypatch) -> None: ) assert supervisor.start_parakeet_server() is None + assert parakeet_server.read_parakeet_placement() is None -def test_start_parakeet_server_early_returns_for_other_backend(monkeypatch) -> None: +def test_start_parakeet_server_early_returns_for_other_backend( + monkeypatch, tmp_path +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + parakeet_server.write_parakeet_placement("gpu") monkeypatch.setattr(supervisor.sys, "platform", "linux") monkeypatch.setattr(supervisor.platform, "machine", lambda: "x86_64") monkeypatch.setattr( @@ -309,11 +513,15 @@ def test_start_parakeet_server_early_returns_for_other_backend(monkeypatch) -> N ) assert supervisor.start_parakeet_server() is None + assert parakeet_server.read_parakeet_placement() is None def test_start_parakeet_server_starts_background_install_when_missing( monkeypatch, + tmp_path, ) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + parakeet_server.write_parakeet_placement("gpu") monkeypatch.setattr(supervisor, "linux_stt_uses_parakeet_cpp", lambda: True) monkeypatch.setattr(supervisor, "_configured_parakeet_device", lambda: "cpu") started: list[str] = [] @@ -353,6 +561,7 @@ def test_start_parakeet_server_starts_background_install_when_missing( assert supervisor.start_parakeet_server() is None assert started == ["parakeet-cpp-provider-bootstrap"] + assert parakeet_server.read_parakeet_placement() is None def test_parakeet_bootstrap_worker_requests_start_after_install(monkeypatch) -> None: diff --git a/tests/test_transcribe_parakeet_cpp.py b/tests/test_transcribe_parakeet_cpp.py index 0a29c4b02..68c1a50f9 100644 --- a/tests/test_transcribe_parakeet_cpp.py +++ b/tests/test_transcribe_parakeet_cpp.py @@ -329,3 +329,46 @@ def test_get_model_info_does_not_connect(monkeypatch: pytest.MonkeyPatch) -> Non "compute_type": "q8_0", "per_word_confidence": True, } + + +@pytest.mark.parametrize("placement", ["gpu", "cpu"]) +def test_get_model_info_reports_supervisor_placement_record( + monkeypatch: pytest.MonkeyPatch, + tmp_path, + placement: str, +) -> None: + monkeypatch.setattr(parakeet_cpp.sys, "platform", "linux") + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + parakeet_cpp.parakeet_server.write_parakeet_placement(placement) + + info = parakeet_cpp.get_model_info({"device": "auto"}) + + assert info["device"] == placement + assert info["device"] != "auto" + + +def test_get_model_info_uses_configured_device_without_placement_record( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +) -> None: + monkeypatch.setattr(parakeet_cpp.sys, "platform", "linux") + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + + info = parakeet_cpp.get_model_info({"device": "auto"}) + + assert info["device"] == "auto" + + +def test_get_model_info_ignores_invalid_placement_record( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +) -> None: + monkeypatch.setattr(parakeet_cpp.sys, "platform", "linux") + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + placement_path = tmp_path / "journal" / "health" / "parakeet-cpp.placement" + placement_path.parent.mkdir(parents=True) + placement_path.write_text("vulkan") + + info = parakeet_cpp.get_model_info({"device": "cpu"}) + + assert info["device"] == "cpu" diff --git a/tests/test_transcribe_telemetry.py b/tests/test_transcribe_telemetry.py index b5922b654..709054d91 100644 --- a/tests/test_transcribe_telemetry.py +++ b/tests/test_transcribe_telemetry.py @@ -51,13 +51,18 @@ def _backend_module() -> MagicMock: backend_module = MagicMock() backend_module.get_model_info.return_value = { "model": "parakeet-v3-q8_0.gguf", - "device": "auto", + "device": "gpu", "compute_type": "q8_0", } return backend_module -def _run_success(raw_path: Path, audio_buffer, vad_result) -> dict: +def _run_success( + raw_path: Path, + audio_buffer, + vad_result, + backend_module: MagicMock | None = None, +) -> dict: """Run a successful process_audio and return the emitted event kwargs.""" from solstone.observe.transcribe.main import process_audio @@ -79,7 +84,7 @@ def _run_success(raw_path: Path, audio_buffer, vad_result) -> dict: ), patch( "solstone.observe.transcribe.main.get_backend", - return_value=_backend_module(), + return_value=backend_module or _backend_module(), ), patch("solstone.observe.transcribe.main._embed_statements", return_value=None), patch( @@ -110,8 +115,10 @@ def test_success_event_carries_stage_timings_and_envelope( assert all(isinstance(v, int) and v >= 0 for v in timings.values()) assert kwargs["backend"] == "parakeet-cpp" - assert kwargs["device"] == "auto" + assert kwargs["device"] == "gpu" assert kwargs["model"] == "parakeet-v3-q8_0.gguf" + header = json.loads(raw_path.with_suffix(".jsonl").read_text().splitlines()[0]) + assert header["device"] == "gpu" assert kwargs["audio_seconds"] == 10.0 assert isinstance(kwargs["peak_rss_mib"], int) assert kwargs["peak_rss_mib"] > 0 @@ -131,6 +138,29 @@ def test_success_event_is_content_free( assert banned not in kwargs +@pytest.mark.parametrize("placement", ["gpu", "cpu"]) +def test_success_event_and_header_use_parakeet_placement_record( + monkeypatch: pytest.MonkeyPatch, + raw_path: Path, + audio_buffer: np.ndarray, + vad_result: VadResult, + placement: str, +) -> None: + from solstone.observe.transcribe import _parakeet_cpp as parakeet_cpp + + monkeypatch.setattr(parakeet_cpp.sys, "platform", "linux") + monkeypatch.setenv("SOLSTONE_JOURNAL", str(raw_path.parents[4])) + parakeet_cpp.parakeet_server.write_parakeet_placement(placement) + backend_module = MagicMock() + backend_module.get_model_info.side_effect = parakeet_cpp.get_model_info + + kwargs = _run_success(raw_path, audio_buffer, vad_result, backend_module) + + assert kwargs["device"] == placement + header = json.loads(raw_path.with_suffix(".jsonl").read_text().splitlines()[0]) + assert header["device"] == placement + + def test_failed_event_is_content_free_even_when_the_exception_message_is_not( raw_path: Path, audio_buffer: np.ndarray, vad_result: VadResult ) -> None: