fix(stt): validate selected voice provider availability

This commit is contained in:
Zehua Wang 2026-07-14 20:31:42 -04:00 committed by Teknium
parent eaa2dd6d09
commit 7a56ab2aa2
2 changed files with 136 additions and 19 deletions

View file

@ -505,10 +505,19 @@ class TestCheckVoiceRequirements:
monkeypatch.setattr("tools.voice_mode._audio_available", lambda: True)
monkeypatch.setattr("tools.voice_mode.detect_audio_environment",
lambda: {"available": True, "warnings": []})
monkeypatch.setattr("tools.transcription_tools._get_provider", lambda cfg: "my-custom-stt")
monkeypatch.setattr("tools.transcription_tools._resolve_command_stt_provider_config",
lambda p, c: {"command": "whisper_cpp"} if p == "my-custom-stt" else None)
monkeypatch.setattr(
"tools.transcription_tools._load_stt_config",
lambda: {
"enabled": True,
"provider": "my-custom-stt",
"providers": {
"my-custom-stt": {
"type": "command",
"command": "whisper_cpp {input}",
},
},
},
)
from tools.voice_mode import check_voice_requirements
result = check_voice_requirements()
@ -521,9 +530,26 @@ class TestCheckVoiceRequirements:
monkeypatch.setattr("tools.voice_mode._audio_available", lambda: True)
monkeypatch.setattr("tools.voice_mode.detect_audio_environment",
lambda: {"available": True, "warnings": []})
monkeypatch.setattr("tools.transcription_tools._get_provider", lambda cfg: "none")
monkeypatch.setattr("tools.transcription_tools._resolve_command_stt_provider_config",
lambda p, c: {"command": "whisper_cpp"} if p == "my-custom-stt" else None)
monkeypatch.setattr(
"tools.transcription_tools._load_stt_config",
lambda: {
"enabled": True,
"provider": "unknown-selected",
"providers": {
"unrelated-command": {
"type": "command",
"command": "whisper_cpp {input}",
},
},
},
)
monkeypatch.setattr(
"agent.transcription_registry.get_provider", lambda p: None,
)
monkeypatch.setattr(
"hermes_cli.plugins._ensure_plugins_discovered",
lambda force=False: None,
)
from tools.voice_mode import check_voice_requirements
@ -537,11 +563,20 @@ class TestCheckVoiceRequirements:
monkeypatch.setattr("tools.voice_mode._audio_available", lambda: True)
monkeypatch.setattr("tools.voice_mode.detect_audio_environment",
lambda: {"available": True, "warnings": []})
monkeypatch.setattr("tools.transcription_tools._get_provider", lambda cfg: "my-plugin-stt")
monkeypatch.setattr("tools.transcription_tools._resolve_command_stt_provider_config",
lambda p, c: None)
monkeypatch.setattr("tools.voice_mode._check_plugin_stt_provider",
lambda p: p == "my-plugin-stt")
monkeypatch.setattr(
"tools.transcription_tools._load_stt_config",
lambda: {"enabled": True, "provider": "my-plugin-stt"},
)
plugin_provider = MagicMock()
plugin_provider.is_available.return_value = True
monkeypatch.setattr(
"agent.transcription_registry.get_provider",
lambda p: plugin_provider if p == "my-plugin-stt" else None,
)
monkeypatch.setattr(
"hermes_cli.plugins._ensure_plugins_discovered",
lambda force=False: None,
)
from tools.voice_mode import check_voice_requirements
@ -550,6 +585,33 @@ class TestCheckVoiceRequirements:
assert result["stt_available"] is True
assert "STT provider: OK (plugin: my-plugin-stt)" in result["details"]
def test_unavailable_plugin_stt_provider(self, monkeypatch):
"""A registered but unavailable plugin does not satisfy requirements."""
monkeypatch.setattr("tools.voice_mode._audio_available", lambda: True)
monkeypatch.setattr("tools.voice_mode.detect_audio_environment",
lambda: {"available": True, "warnings": []})
monkeypatch.setattr(
"tools.transcription_tools._load_stt_config",
lambda: {"enabled": True, "provider": "my-plugin-stt"},
)
plugin_provider = MagicMock()
plugin_provider.is_available.return_value = False
monkeypatch.setattr(
"agent.transcription_registry.get_provider",
lambda p: plugin_provider if p == "my-plugin-stt" else None,
)
monkeypatch.setattr(
"hermes_cli.plugins._ensure_plugins_discovered",
lambda force=False: None,
)
from tools.voice_mode import check_voice_requirements
result = check_voice_requirements()
assert result["available"] is False
assert result["stt_available"] is False
assert "STT provider: MISSING" in result["details"]
# ============================================================================
# AudioRecorder

View file

@ -1270,13 +1270,42 @@ def listen_for_speech(
# Requirements check
# ============================================================================
def _check_plugin_stt_provider(provider: str) -> bool:
"""Return True when *provider* is backed by a registered TranscriptionProvider plugin."""
"""Return True when *provider* resolves to an available STT plugin."""
if not provider:
return False
key = provider.lower().strip()
if key == "none":
return False
try:
from agent.transcription_registry import get_provider
return get_provider(provider.lower().strip()) is not None
except ImportError:
from hermes_cli.plugins import _ensure_plugins_discovered
_ensure_plugins_discovered()
plugin_provider = get_provider(key)
if plugin_provider is None:
# Match the transcription dispatcher: long-lived processes may
# need one refresh after plugins or configuration change.
_ensure_plugins_discovered(force=True)
plugin_provider = get_provider(key)
except Exception as exc: # noqa: BLE001 - discovery failure is non-fatal
logger.debug(
"STT plugin requirements check skipped for '%s': %s", key, exc,
)
return False
if plugin_provider is None:
return False
try:
return bool(plugin_provider.is_available())
except Exception as exc: # noqa: BLE001 - plugins must not break status
logger.warning(
"STT plugin provider '%s' is_available() raised during requirements "
"check: %s - treating as unavailable",
key,
exc,
exc_info=True,
)
return False
@ -1288,11 +1317,37 @@ def check_voice_requirements() -> Dict[str, Any]:
``missing_packages``, and ``details``.
"""
# Determine STT provider availability
from tools.transcription_tools import _get_provider, _load_stt_config, _resolve_command_stt_provider_config, is_stt_enabled
from tools.transcription_tools import (
_get_provider,
_load_stt_config,
_resolve_command_stt_provider_config,
is_stt_enabled,
)
stt_config = _load_stt_config()
stt_enabled = is_stt_enabled(stt_config)
stt_provider = _get_provider(stt_config)
stt_available = stt_enabled and stt_provider != "none"
native_stt_available = stt_provider in {
"local",
"local_command",
"groq",
"openai",
"mistral",
"xai",
"elevenlabs",
}
command_stt_config = None
plugin_stt_available = False
if stt_enabled and not native_stt_available:
command_stt_config = _resolve_command_stt_provider_config(
stt_provider, stt_config,
)
if command_stt_config is None:
plugin_stt_available = _check_plugin_stt_provider(stt_provider)
stt_available = stt_enabled and (
native_stt_available
or command_stt_config is not None
or plugin_stt_available
)
missing: List[str] = []
termux_capture = _termux_voice_capture_available()
@ -1330,9 +1385,9 @@ def check_voice_requirements() -> Dict[str, Any]:
details_parts.append("STT provider: OK (xAI Grok STT)")
elif stt_provider == "elevenlabs":
details_parts.append("STT provider: OK (ElevenLabs Scribe)")
elif _resolve_command_stt_provider_config(stt_provider, stt_config):
elif command_stt_config is not None:
details_parts.append(f"STT provider: OK (command: {stt_provider})")
elif _check_plugin_stt_provider(stt_provider):
elif plugin_stt_available:
details_parts.append(f"STT provider: OK (plugin: {stt_provider})")
else:
details_parts.append(