diff --git a/CHANGELOG.md b/CHANGELOG.md index c10e6fdb..89f91706 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,8 @@ the frozen-backend fallback mirror it for their toolchains. ### Added +- Engine status and diagnostic bundles now record loaded execution provider, device, precision, fallback stage, accelerator identity, runtime versions, and parent-process memory visibility (#1717) + ### Docs - The CosyVoice guide now states that packaged builds have no one-click runtime installer and records the exact readiness checks exposed by [Discussion 1631](https://github.com/debpalash/VoiceStudio/discussions/1631). diff --git a/backend/api/public_engine_metadata.py b/backend/api/public_engine_metadata.py index d626c7ab..73da4ec7 100644 --- a/backend/api/public_engine_metadata.py +++ b/backend/api/public_engine_metadata.py @@ -49,6 +49,13 @@ def public_backends(entries: list[dict]) -> list[dict]: item["routing_reason"] = _public_routing_reason( item.get("routing_status"), item["routing_reason"] ) + evidence = item.get("execution_evidence") + if isinstance(evidence, dict) and evidence.get("cpu_fallback_reason") is not None: + evidence = dict(evidence) + evidence["cpu_fallback_reason"] = _public_routing_reason( + "cpu_fallback", evidence["cpu_fallback_reason"] + ) + item["execution_evidence"] = evidence safe.append(item) return safe diff --git a/backend/core/diagnose.py b/backend/core/diagnose.py index 27a25e85..1f3ed317 100644 --- a/backend/core/diagnose.py +++ b/backend/core/diagnose.py @@ -21,6 +21,7 @@ Check shape: """ from __future__ import annotations +import importlib import os import platform import shutil @@ -367,10 +368,44 @@ def run_diagnostics(include_network: bool = True, deep: bool = False) -> dict: counts = {OK: 0, WARN: 0, FAIL: 0} for c in checks: counts[c["status"]] += 1 + engine_execution = [] + for family in ("tts", "asr"): + active = "unknown" + try: + module = importlib.import_module(f"services.{family}_backend") + active = module.active_backend_id() + row = next((item for item in module.list_backends() if item.get("id") == active), None) + if row is not None: + engine_execution.append({ + "family": family, + "engine_id": active, + **row["execution_evidence"], + }) + except Exception: # noqa: BLE001 - evidence must not break diagnostics + # Preserve the other family's successful evidence and make this + # collection failure explicit without exposing exception text. + engine_execution.append({ + "family": family, + "engine_id": active, + "implementation_variant": None, + "declared_device_families": [], + "evidence_state": "collection_failed", + "actual_execution_provider": None, + "actual_execution_device": None, + "gpu_name": None, + "gpu_architecture": None, + "precision_or_quantization": None, + "cpu_fallback_reason": None, + "cpu_fallback_stage": None, + "parent_memory_observable": None, + "runtime_versions": {}, + }) + return { "app_version": APP_VERSION, "platform": scrub_text(platform.platform()), "checks": checks, + "engine_execution": engine_execution, "summary": { "ok": counts[FAIL] == 0, "passed": counts[OK], @@ -395,6 +430,35 @@ def format_text(report: dict) -> str: lines.append(f"{tag[c['status']]} {c['label']}: {c['detail']}") if c.get("hint"): lines.append(f" hint: {c['hint']}") + if report.get("engine_execution"): + lines.append("") + lines.append("Engine execution evidence:") + for item in report["engine_execution"]: + if item.get("actual_execution_provider"): + provider = item["actual_execution_provider"] + elif item.get("evidence_state") == "subprocess_loaded_provider_unreported": + provider = "loaded child; provider not reported" + else: + provider = "not loaded" + precision = item.get("precision_or_quantization") or "unknown" + device = item.get("actual_execution_device") or "unknown" + gpu = item.get("gpu_name") or "none" + architecture = item.get("gpu_architecture") or "unknown" + fallback_stage = item.get("cpu_fallback_stage") or "none" + fallback_reason = item.get("cpu_fallback_reason") or "none" + versions = ",".join( + f"{name}={version}" + for name, version in sorted(item.get("runtime_versions", {}).items()) + ) or "none" + visible = "yes" if item.get("parent_memory_observable") else "no" + lines.append( + f" {item['family']}:{item['engine_id']} provider={provider}; " + f"device={device}; gpu={gpu}; architecture={architecture}; " + f"precision={precision}; fallback-stage={fallback_stage}; " + f"fallback-reason={fallback_reason}; runtimes={versions}; " + f"evidence-state={item.get('evidence_state', 'unknown')}; " + f"parent-memory-visible={visible}" + ) s = report["summary"] lines.append("") lines.append( diff --git a/backend/services/asr_backend.py b/backend/services/asr_backend.py index 8e8c90df..116f2b79 100644 --- a/backend/services/asr_backend.py +++ b/backend/services/asr_backend.py @@ -30,6 +30,7 @@ import re import contextlib import threading import time +import weakref from utils.containment import contain_system_exit from abc import ABC, abstractmethod @@ -304,6 +305,16 @@ class ASRBackend(ABC): # broken GPU path, strictly worse than the honest `cpu_fallback`.) gpu_compat: tuple[str, ...] = ("cpu",) + def execution_evidence_loaded(self) -> bool: + """Whether this instance has live model state worth reporting.""" + if getattr(self, "runs_out_of_process", False): + proc = getattr(self, "_proc", None) + return proc is not None and proc.poll() is None + return any( + getattr(self, attr, None) is not None + for attr in ("_model", "_asr", "_pipeline", "_pipe", "_transcriber", "_rec") + ) + @classmethod @abstractmethod def is_available(cls) -> tuple[bool, str]: @@ -950,6 +961,8 @@ class FasterWhisperBackend(ASRBackend): # (after the #551 compute_type / #255 OOM→CPU fallback chain). self._device: str | None = None self._compute_type: str | None = None + self._fallback_reason: str | None = None + self._fallback_stage: str | None = None @classmethod def is_available(cls) -> tuple[bool, str]: @@ -1027,6 +1040,8 @@ class FasterWhisperBackend(ASRBackend): except Exception: # noqa: BLE001 — cache clear is best-effort pass device = "cpu" + self._fallback_reason = "CUDA memory was exhausted while loading the engine" + self._fallback_stage = "model_load" candidates = _compute_type_candidates(device) compute_type = candidates[0] continue @@ -2354,6 +2369,8 @@ _LAST_ERRORS: dict[str, str] = {} # failing ASR wholesale. Per-process by design: repairing the env requires a # reinstall / ``uv sync --reinstall`` and an app restart anyway. _DEEP_IMPORT_BROKEN: dict[str, str] = {} +_RUNTIME_EVIDENCE: dict[str, dict] = {} +_RUNTIME_INSTANCES: weakref.WeakValueDictionary[str, "ASRBackend"] = weakref.WeakValueDictionary() def _deep_import_reason(cls: type["ASRBackend"], exc: ImportError) -> str: @@ -2384,6 +2401,7 @@ def list_backends() -> list[dict]: """ from core.device_caps import detect_host_caps from core.scrub import scrub_text + from services.engine_evidence import snapshot as execution_snapshot from services.engine_routing import routing_fields caps = detect_host_caps() @@ -2408,6 +2426,24 @@ def list_backends() -> list[dict]: _LAST_ERRORS[bid] = scrub_text(msg) isolation = "subprocess" if getattr(cls, "_is_subprocess_isolated", False) else "in-process" gpu_compat = getattr(cls, "gpu_compat", ("cpu",)) + routing = routing_fields(gpu_compat, caps) + # Cached load-time facts are valid only while their exact backend still + # owns live model state. Recompute from that instance so unload/reaping + # cannot leave ghost GPU/provider evidence in diagnostics. + instance = ( + _ISOLATED_INSTANCES.get(bid) + if isolation == "subprocess" + else _RUNTIME_INSTANCES.get(bid) + ) + execution_evidence = execution_snapshot( + engine_id=bid, + engine_cls=cls, + instance=instance, + routing=routing, + caps=caps, + ) + if execution_evidence["evidence_state"] == "not_loaded": + _RUNTIME_EVIDENCE.pop(bid, None) out.append({ "id": bid, "display_name": cls.display_name, @@ -2419,7 +2455,14 @@ def list_backends() -> list[dict]: "last_error": _LAST_ERRORS.get(bid), "isolation_mode": isolation, "gpu_compat": list(gpu_compat), - **routing_fields(gpu_compat, caps), + **routing, + "execution_evidence": execution_evidence or execution_snapshot( + engine_id=bid, + engine_cls=cls, + instance=None, + routing=routing, + caps=caps, + ), }) return out @@ -2660,6 +2703,21 @@ def load_active_asr_backend(*, asr_pipe=None) -> ASRBackend: raise ASRModelMissingError(missing) try: backend.ensure_loaded() + from core.device_caps import detect_host_caps + from services.engine_evidence import snapshot as execution_snapshot + from services.engine_routing import routing_fields + + cls = type(backend) + caps = detect_host_caps() + routing = routing_fields(getattr(cls, "gpu_compat", ("cpu",)), caps) + _RUNTIME_EVIDENCE[bid] = execution_snapshot( + engine_id=bid, + engine_cls=cls, + instance=backend, + routing=routing, + caps=caps, + ) + _RUNTIME_INSTANCES[bid] = backend return backend except ImportError as e: # ModuleNotFoundError and its ImportError parent ("cannot import diff --git a/backend/services/engine_evidence.py b/backend/services/engine_evidence.py new file mode 100644 index 00000000..bc6af1ab --- /dev/null +++ b/backend/services/engine_evidence.py @@ -0,0 +1,119 @@ +"""Sanitized, reproducible execution evidence for TTS and ASR engines.""" +from __future__ import annotations + +import importlib.metadata +import platform +from typing import Any + + +def _version(distribution: str) -> str | None: + try: + return importlib.metadata.version(distribution) + except importlib.metadata.PackageNotFoundError: + return None + + +def _value(instance: object, *names: str) -> str | None: + for name in names: + try: + value = getattr(instance, name, None) + if value is not None and not callable(value): + text = str(value).strip() + if text and len(text) <= 80 and "/" not in text and "\\" not in text: + return text + except Exception: + continue + return None + + +def runtime_versions(engine_id: str) -> dict[str, str]: + """Relevant installed library versions, never paths or environment values.""" + names = {"python": platform.python_version()} + candidates = ["torch"] + low = engine_id.lower() + if "faster" in low or "whisperx" in low: + candidates.extend(["ctranslate2", "faster-whisper"]) + if "sherpa" in low or "moonshine" in low: + candidates.append("onnxruntime") + if "mlx" in low: + candidates.append("mlx") + for name in candidates: + if (version := _version(name)) is not None: + names[name] = version + return names + + +def snapshot( + *, + engine_id: str, + engine_cls: type, + instance: object | None, + routing: dict[str, Any], + caps: object, +) -> dict[str, Any]: + """Return fixed-shape evidence; actual fields stay null until an instance loads.""" + isolated = bool( + getattr(engine_cls, "_is_subprocess_isolated", False) + or getattr(engine_cls, "runs_out_of_process", False) + ) + loaded = False + probe_failed = False + if instance is not None: + try: + contract = getattr(instance, "execution_evidence_loaded", False) + loaded = bool(contract() if callable(contract) else contract) + except Exception: # noqa: BLE001 - third-party lifecycle descriptors may raise + probe_failed = True + + actual_device = None + provider = None + precision = None + if loaded: + actual_device = _value(instance, "_device", "device", "execution_device") + provider = _value(instance, "_provider", "provider", "execution_provider") + precision = _value( + instance, "_compute_type", "compute_type", "_dtype", "dtype", "quantization" + ) + if provider is None and actual_device is not None: + provider = actual_device + + runtime_fallback_reason = _value(instance, "_fallback_reason", "fallback_reason") if loaded else None + runtime_fallback_stage = _value(instance, "_fallback_stage", "fallback_stage") if loaded else None + status = routing.get("routing_status") + fallback = status == "cpu_fallback" or runtime_fallback_reason is not None + evidence_state = "not_loaded" + if probe_failed: + evidence_state = "probe_error" + elif loaded: + evidence_state = "loaded" + if isolated and provider is None and actual_device is None: + evidence_state = "subprocess_loaded_provider_unreported" + return { + "implementation_variant": f"{engine_cls.__module__}.{engine_cls.__name__}", + "declared_device_families": list(getattr(engine_cls, "gpu_compat", ("cpu",))), + "evidence_state": evidence_state, + "actual_execution_provider": provider, + "actual_execution_device": actual_device, + "gpu_name": getattr(caps, "device_name", "") or None, + "gpu_architecture": _gpu_architecture(getattr(caps, "family", "cpu")), + "precision_or_quantization": precision, + "cpu_fallback_reason": runtime_fallback_reason or (routing.get("routing_reason") if fallback else None), + "cpu_fallback_stage": runtime_fallback_stage or ("routing_preflight" if fallback else None), + "parent_memory_observable": not isolated, + "runtime_versions": runtime_versions(engine_id), + } + + +def _gpu_architecture(family: str) -> str | None: + if family not in {"cuda", "rocm"}: + return "apple-silicon" if family == "mps" else None + try: + import torch + + if family == "rocm": + props = torch.cuda.get_device_properties(0) + return str(getattr(props, "gcnArchName", "") or "") or None + major, minor = torch.cuda.get_device_capability(0) + return f"sm_{major}{minor}" + except Exception: + return None diff --git a/backend/services/tts_backend.py b/backend/services/tts_backend.py index bf02c7d9..1fb1c8b7 100644 --- a/backend/services/tts_backend.py +++ b/backend/services/tts_backend.py @@ -407,6 +407,13 @@ class TTSBackend(ABC): # entirely (it drives the shared model_manager singleton). _MODEL_ATTRS: tuple[str, ...] = ("_model", "_tts") + def execution_evidence_loaded(self) -> bool: + """Whether this instance has live model state worth reporting.""" + if self.runs_out_of_process: + proc = getattr(self, "_proc", None) + return proc is not None and proc.poll() is None + return any(getattr(self, attr, None) is not None for attr in self._MODEL_ATTRS) + def unload(self) -> None: """Release the heavy model this backend holds, and free device caches. @@ -2392,6 +2399,7 @@ def list_backends() -> list[dict]: # Routing is host-aware but the host caps are constant per process, so probe # ONCE here and resolve each engine's effective device against the same caps. from core.device_caps import detect_host_caps + from services.engine_evidence import snapshot as execution_snapshot from services.engine_routing import routing_fields caps = detect_host_caps() installable = _sidecar_installable_ids() @@ -2425,6 +2433,12 @@ def list_backends() -> list[dict]: # descriptor, not a bool, so report None (= model-dependent) there # instead of an always-truthy false positive. _clone = getattr(cls, "supports_cloning", True) + routing = routing_fields(gpu_compat, caps, getattr(cls, "min_vram_gb", 0.0)) + loaded_instance = None + if _active_instance_id == bid: + loaded_instance = _active_instance + if loaded_instance is None: + loaded_instance = _ENGINE_INSTANCES.get(cls) out.append({ "id": bid, "display_name": cls.display_name, @@ -2451,7 +2465,14 @@ def list_backends() -> list[dict]: "min_vram_gb": getattr(cls, "min_vram_gb", 0.0) or None, # effective_device / routing_status / routing_reason (scrubbed); # the reason now also carries the under-provisioned-GPU caveat. - **routing_fields(gpu_compat, caps, getattr(cls, "min_vram_gb", 0.0)), + **routing, + "execution_evidence": execution_snapshot( + engine_id=bid, + engine_cls=cls, + instance=loaded_instance, + routing=routing, + caps=caps, + ), }) # #981: mlx-audio multiplexes 7+ curated models behind one backend id # — surface the roster + the currently-active pick so Settings can diff --git a/docs/install/troubleshooting.md b/docs/install/troubleshooting.md index 5f238afa..ccb6cdd4 100644 --- a/docs/install/troubleshooting.md +++ b/docs/install/troubleshooting.md @@ -30,6 +30,15 @@ Before digging through the entries below, let the app diagnose itself: tails) you can drag straight onto the GitHub issue. Home paths and anything token-shaped are redacted before they leave your machine. + The report's `engine_execution` rows distinguish declared compatibility + from observed runtime state. `evidence_state: not_loaded` means no actual + provider can yet be claimed. Run the deep self-check for TTS, or issue a + representative ASR request for ASR evidence. Loaded rows include the execution provider/device, precision + or quantization when the engine exposes it, GPU identity, CPU-fallback stage, + relevant library versions, and whether parent-process memory counters cover + the engine. Subprocess engines report memory visibility as false because + their accelerator allocations belong to the child process. + ## 1. `pkg_resources` missing (ModuleNotFoundError) diff --git a/tests/backend/services/test_tts_backend_registry.py b/tests/backend/services/test_tts_backend_registry.py index a7ef1843..798e35f7 100644 --- a/tests/backend/services/test_tts_backend_registry.py +++ b/tests/backend/services/test_tts_backend_registry.py @@ -173,6 +173,8 @@ def test_list_backends_shape(registry_sandbox): # below the floor gets a caveat in `routing_reason` BEFORE it spends # the full compute budget finding out its card is too small. "min_vram_gb", + # Sanitized actual-vs-declared provider/device evidence (#1717). + "execution_evidence", } mlx_audio_extra = {"curated_models", "active_model_id"} for entry in out: diff --git a/tests/test_diagnose.py b/tests/test_diagnose.py index a3ab7836..10d26ff2 100644 --- a/tests/test_diagnose.py +++ b/tests/test_diagnose.py @@ -13,7 +13,7 @@ def report(): def test_report_shape(report): - assert set(report) == {"app_version", "platform", "checks", "summary"} + assert set(report) == {"app_version", "platform", "checks", "engine_execution", "summary"} ids = [c["id"] for c in report["checks"]] assert len(ids) == len(set(ids)), "check ids must be unique" for c in report["checks"]: @@ -72,6 +72,39 @@ def test_format_text_ascii_and_exit_signal(report): assert ("looks healthy" in text) == report["summary"]["ok"] +def test_asr_evidence_failure_does_not_drop_tts_evidence(monkeypatch): + from services import asr_backend, tts_backend + + evidence = { + "evidence_state": "loaded", + "actual_execution_provider": "cpu", + "actual_execution_device": "cpu", + "precision_or_quantization": "float32", + "cpu_fallback_reason": None, + "cpu_fallback_stage": None, + "runtime_versions": {"python": "3.11"}, + "parent_memory_observable": True, + } + monkeypatch.setattr(tts_backend, "active_backend_id", lambda: "tts-ok") + monkeypatch.setattr( + tts_backend, + "list_backends", + lambda: [{"id": "tts-ok", "available": True, "execution_evidence": evidence}], + ) + monkeypatch.setattr(asr_backend, "active_backend_id", lambda: "asr-broken") + def fail_asr_registry(): + raise RuntimeError("registry failed") + + monkeypatch.setattr(asr_backend, "list_backends", fail_asr_registry) + + result = run_diagnostics(include_network=False) + rows = {row["family"]: row for row in result["engine_execution"]} + assert rows["tts"]["engine_id"] == "tts-ok" + assert rows["tts"]["evidence_state"] == "loaded" + assert rows["asr"]["engine_id"] == "asr-broken" + assert rows["asr"]["evidence_state"] == "collection_failed" + + # ── Deep synthesis check (mocked — no real model load in CI) ───────────── diff --git a/tests/test_diagnostic_bundle.py b/tests/test_diagnostic_bundle.py index 8eb97a67..2cd79201 100644 --- a/tests/test_diagnostic_bundle.py +++ b/tests/test_diagnostic_bundle.py @@ -2,6 +2,7 @@ import json import os import zipfile +from types import SimpleNamespace import pytest @@ -47,6 +48,53 @@ def test_bundle_members_and_meta(bundle_env): assert meta["app_version"] report = json.loads(zf.read("self_check.json")) assert report["summary"]["passed"] >= 1 + assert "engine_execution" in report + text_report = zf.read("self_check.txt").decode() + if report["engine_execution"]: + assert "Engine execution evidence:" in text_report + row = report["engine_execution"][0] + assert f"{row['family']}:{row['engine_id']}" in text_report + assert f"evidence-state={row['evidence_state']}" in text_report + assert "device=" in text_report + assert "precision=" in text_report + assert "fallback-stage=" in text_report + + +def test_asr_import_failure_preserves_tts_execution_evidence(monkeypatch): + from core import diagnose + + evidence = { + "implementation_variant": "fake", + "declared_device_families": ["cpu"], + "evidence_state": "loaded", + "actual_execution_provider": "cpu", + "actual_execution_device": "cpu", + "gpu_name": None, + "gpu_architecture": None, + "precision_or_quantization": "fp32", + "cpu_fallback_reason": None, + "cpu_fallback_stage": None, + "parent_memory_observable": True, + "runtime_versions": {}, + } + tts = SimpleNamespace( + active_backend_id=lambda: "fake-tts", + list_backends=lambda: [{"id": "fake-tts", "execution_evidence": evidence}], + ) + real_import = diagnose.importlib.import_module + + def import_family(name): + if name == "services.tts_backend": + return tts + if name == "services.asr_backend": + raise ImportError("unavailable") + return real_import(name) + + monkeypatch.setattr(diagnose.importlib, "import_module", import_family) + rows = diagnose.run_diagnostics(include_network=False)["engine_execution"] + + assert next(row for row in rows if row["family"] == "tts")["engine_id"] == "fake-tts" + assert next(row for row in rows if row["family"] == "asr")["evidence_state"] == "collection_failed" def test_bundle_log_tails_are_scrubbed(bundle_env): diff --git a/tests/test_engine_execution_evidence.py b/tests/test_engine_execution_evidence.py new file mode 100644 index 00000000..29599840 --- /dev/null +++ b/tests/test_engine_execution_evidence.py @@ -0,0 +1,272 @@ +from dataclasses import dataclass + +@dataclass +class _Caps: + family: str = "rocm" + device_name: str = "AMD Radeon RX 6700 XT" + + +class _TorchEngine: + execution_evidence_loaded = True + gpu_compat = ("rocm", "cpu") + _device = "cuda:0" + _dtype = "float16" + + +class _FasterWhisper: + execution_evidence_loaded = True + gpu_compat = ("cuda", "cpu") + _device = "cpu" + _compute_type = "int8" + + +class _OnnxEngine: + execution_evidence_loaded = True + gpu_compat = ("cpu",) + _provider = "CPUExecutionProvider" + _dtype = "int8" + + +class _SidecarEngine: + execution_evidence_loaded = True + gpu_compat = ("rocm", "cpu") + runs_out_of_process = True + _device = "cuda:0" + + +class _LoadFallbackEngine: + execution_evidence_loaded = True + gpu_compat = ("cuda", "cpu") + _device = "cpu" + _fallback_reason = "CUDA memory was exhausted while loading the engine" + _fallback_stage = "model_load" + + +def _snap(cls, routing): + from services.engine_evidence import snapshot + + return snapshot( + engine_id=cls.__name__, engine_cls=cls, instance=cls(), routing=routing, caps=_Caps() + ) + + +def test_loaded_rocm_torch_engine_reports_actual_device(): + evidence = _snap(_TorchEngine, {"routing_status": "accelerated", "routing_reason": None}) + assert evidence["actual_execution_device"] == "cuda:0" + assert evidence["precision_or_quantization"] == "float16" + assert evidence["gpu_name"] == "AMD Radeon RX 6700 XT" + + +def test_faster_whisper_cpu_fallback_names_reason_and_stage(): + evidence = _snap( + _FasterWhisper, + {"routing_status": "cpu_fallback", "routing_reason": "ROCm is unsupported"}, + ) + assert evidence["actual_execution_device"] == "cpu" + assert evidence["cpu_fallback_reason"] == "ROCm is unsupported" + assert evidence["cpu_fallback_stage"] == "routing_preflight" + + +def test_cpu_onnx_and_subprocess_observability_are_explicit(): + cpu = _snap(_OnnxEngine, {"routing_status": "cpu_only", "routing_reason": None}) + sidecar = _snap(_SidecarEngine, {"routing_status": "accelerated", "routing_reason": None}) + assert cpu["actual_execution_provider"] == "CPUExecutionProvider" + assert cpu["parent_memory_observable"] is True + assert sidecar["parent_memory_observable"] is False + + +def test_subprocess_state_follows_live_child_not_wrapper_presence(): + from services.engine_evidence import snapshot + + class _Process: + def __init__(self, returncode): + self.returncode = returncode + + def poll(self): + return self.returncode + + class _OpaqueSidecar: + gpu_compat = ("cpu",) + runs_out_of_process = True + + def __init__(self, process): + self._proc = process + + def execution_evidence_loaded(self): + return self._proc is not None and self._proc.poll() is None + + routing = {"routing_status": "cpu_only", "routing_reason": None} + for process, expected in ( + (None, "not_loaded"), + (_Process(1), "not_loaded"), + (_Process(None), "subprocess_loaded_provider_unreported"), + ): + evidence = snapshot( + engine_id="opaque", + engine_cls=_OpaqueSidecar, + instance=_OpaqueSidecar(process), + routing=routing, + caps=_Caps(), + ) + assert evidence["evidence_state"] == expected + + +def test_public_inventory_replaces_nested_private_fallback_detail(): + from api.public_engine_metadata import public_backends + + entry = { + "routing_status": "cpu_fallback", + "routing_reason": "/home/alice/private driver error", + "execution_evidence": {"cpu_fallback_reason": "/home/alice/private driver error"}, + } + public = public_backends([entry])[0] + expected = "GPU acceleration is unavailable; this engine will use CPU." + assert public["routing_reason"] == expected + assert public["execution_evidence"]["cpu_fallback_reason"] == expected + + +def test_constructed_in_process_backend_is_not_loaded_until_contract_says_so(): + from services.engine_evidence import snapshot + + class _Lazy: + gpu_compat = ("cpu",) + execution_evidence_loaded = False + _device = "cpu" + + routing = {"routing_status": "cpu_only", "routing_reason": None} + evidence = snapshot( + engine_id="lazy", engine_cls=_Lazy, instance=_Lazy(), routing=routing, caps=_Caps() + ) + assert evidence["evidence_state"] == "not_loaded" + assert evidence["actual_execution_device"] is None + + +def test_post_load_fallback_overrides_preflight_prediction(): + evidence = _snap( + _LoadFallbackEngine, + {"routing_status": "accelerated", "routing_reason": None}, + ) + assert evidence["actual_execution_device"] == "cpu" + assert evidence["cpu_fallback_stage"] == "model_load" + assert "memory" in evidence["cpu_fallback_reason"] + + +def test_lifecycle_probe_lookup_and_call_failures_are_explicit(): + from services.engine_evidence import snapshot + + class _RaisingDescriptor: + gpu_compat = ("cpu",) + + @property + def execution_evidence_loaded(self): + raise RuntimeError("descriptor failed") + + class _RaisingCallable: + gpu_compat = ("cpu",) + + def execution_evidence_loaded(self): + raise RuntimeError("probe failed") + + routing = {"routing_status": "cpu_only", "routing_reason": None} + for cls in (_RaisingDescriptor, _RaisingCallable): + evidence = snapshot( + engine_id="broken-probe", + engine_cls=cls, + instance=cls(), + routing=routing, + caps=_Caps(), + ) + assert evidence["evidence_state"] == "probe_error" + assert evidence["actual_execution_device"] is None + + +def test_public_runtime_fallback_overrides_accelerated_preflight_category(): + from api.public_engine_metadata import public_backends + + public = public_backends( + [{ + "routing_status": "accelerated", + "routing_reason": None, + "execution_evidence": { + "cpu_fallback_reason": "/private/model load failed with hf_secret", + "cpu_fallback_stage": "model_load", + }, + }] + )[0] + assert public["execution_evidence"]["cpu_fallback_reason"] == ( + "GPU acceleration is unavailable; this engine will use CPU." + ) + + +def test_stopped_asr_sidecar_invalidates_cached_loaded_evidence(monkeypatch): + from services import asr_backend + + class _StoppedProcess: + def poll(self): + return 0 + + class _StoppedSidecar: + id = "stopped" + display_name = "Stopped sidecar" + gpu_compat = ("cpu",) + _is_subprocess_isolated = True + runs_out_of_process = True + + def __init__(self): + self._proc = _StoppedProcess() + + @classmethod + def is_available(cls): + return True, "ready" + + def execution_evidence_loaded(self): + return self._proc.poll() is None + + monkeypatch.setattr(asr_backend, "_REGISTRY", {"stopped": _StoppedSidecar}) + monkeypatch.setattr(asr_backend, "_ISOLATED_INSTANCES", {"stopped": _StoppedSidecar()}) + monkeypatch.setattr( + asr_backend, + "_RUNTIME_EVIDENCE", + {"stopped": {"evidence_state": "loaded", "actual_execution_device": "cuda:0"}}, + ) + row = asr_backend.list_backends()[0] + assert row["execution_evidence"]["evidence_state"] == "not_loaded" + assert "stopped" not in asr_backend._RUNTIME_EVIDENCE + + +def test_unloaded_in_process_asr_invalidates_cached_loaded_evidence(monkeypatch): + from services import asr_backend + + class _Backend: + id = "released" + display_name = "Released backend" + gpu_compat = ("cuda", "cpu") + + def __init__(self): + self._model = object() + + @classmethod + def is_available(cls): + return True, "ready" + + def execution_evidence_loaded(self): + return self._model is not None + + def unload(self): + self._model = None + + instance = _Backend() + monkeypatch.setattr(asr_backend, "_REGISTRY", {"released": _Backend}) + monkeypatch.setattr(asr_backend, "_RUNTIME_INSTANCES", {"released": instance}) + monkeypatch.setattr( + asr_backend, + "_RUNTIME_EVIDENCE", + {"released": {"evidence_state": "loaded", "actual_execution_device": "cuda:0"}}, + ) + + instance.unload() + row = asr_backend.list_backends()[0] + + assert row["execution_evidence"]["evidence_state"] == "not_loaded" + assert row["execution_evidence"]["actual_execution_device"] is None + assert "released" not in asr_backend._RUNTIME_EVIDENCE diff --git a/tests/test_engines.py b/tests/test_engines.py index fa515138..817003e9 100644 --- a/tests/test_engines.py +++ b/tests/test_engines.py @@ -20,6 +20,13 @@ def test_tts_registry_lists_all_backends(): assert {"omnivoice", "voxcpm2", "moss-tts-nano"}.issubset(ids) for r in rows: assert set(r) >= {"id", "display_name", "available", "reason"} + evidence = r["execution_evidence"] + assert evidence["implementation_variant"] + assert evidence["declared_device_families"] == r["gpu_compat"] + assert evidence["evidence_state"] in { + "loaded", "not_loaded", "subprocess_loaded_provider_unreported" + } + assert evidence["runtime_versions"]["python"] def test_tts_voxcpm2_unavailable_message_is_actionable(): @@ -195,6 +202,14 @@ def test_asr_registry_lists_backends(): rows = asr_backend.list_backends() ids = {r["id"] for r in rows} assert {"mlx-whisper", "pytorch-whisper"}.issubset(ids) + required = { + "actual_execution_provider", + "actual_execution_device", + "cpu_fallback_reason", + "cpu_fallback_stage", + "runtime_versions", + } + assert all(required.issubset(row["execution_evidence"]) for row in rows) def test_asr_auto_detects():