Merge remote-tracking branch 'origin/main' into fix/dictation-listener-registration-1707

This commit is contained in:
Palash Debnath
2026-09-01 20:00:42 +05:30
12 changed files with 653 additions and 3 deletions
+2
View File
@@ -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).
+7
View File
@@ -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
+64
View File
@@ -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(
+59 -1
View File
@@ -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
+119
View File
@@ -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
+22 -1
View File
@@ -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
+9
View File
@@ -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)
<a id="pkg_resources-missing"></a>
@@ -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:
+34 -1
View File
@@ -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) ─────────────
+48
View File
@@ -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):
+272
View File
@@ -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
+15
View File
@@ -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():