mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-29 18:46:59 +00:00
fix(stt): scope upload size limits to remote providers
This commit is contained in:
parent
884900ffd6
commit
ffc3ce27d5
6 changed files with 280 additions and 58 deletions
|
|
@ -396,7 +396,7 @@ class TestTranscribeCommandSTT:
|
|||
audio = _make_silent_wav(tmp_path / "input.wav")
|
||||
# Write the model into the transcript so we can assert it propagated.
|
||||
interpreter = sys.executable
|
||||
payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])"
|
||||
payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])"
|
||||
cfg = {
|
||||
"command": f'"{interpreter}" -c "{payload}" {{model}} {{output_path}}',
|
||||
"model": "config-model",
|
||||
|
|
@ -410,7 +410,7 @@ class TestTranscribeCommandSTT:
|
|||
def test_config_model_used_when_no_override(self, tmp_path):
|
||||
audio = _make_silent_wav(tmp_path / "input.wav")
|
||||
interpreter = sys.executable
|
||||
payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])"
|
||||
payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])"
|
||||
cfg = {
|
||||
"command": f'"{interpreter}" -c "{payload}" {{model}} {{output_path}}',
|
||||
"model": "config-model",
|
||||
|
|
@ -421,7 +421,7 @@ class TestTranscribeCommandSTT:
|
|||
def test_language_from_provider_config_wins(self, tmp_path):
|
||||
audio = _make_silent_wav(tmp_path / "input.wav")
|
||||
interpreter = sys.executable
|
||||
payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])"
|
||||
payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])"
|
||||
cfg = {
|
||||
"command": f'"{interpreter}" -c "{payload}" {{language}} {{output_path}}',
|
||||
"language": "fr",
|
||||
|
|
@ -435,7 +435,7 @@ class TestTranscribeCommandSTT:
|
|||
def test_language_falls_back_to_stt_section(self, tmp_path):
|
||||
audio = _make_silent_wav(tmp_path / "input.wav")
|
||||
interpreter = sys.executable
|
||||
payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])"
|
||||
payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])"
|
||||
cfg = {
|
||||
"command": f'"{interpreter}" -c "{payload}" {{language}} {{output_path}}',
|
||||
}
|
||||
|
|
@ -447,7 +447,7 @@ class TestTranscribeCommandSTT:
|
|||
def test_language_defaults_to_en(self, tmp_path):
|
||||
audio = _make_silent_wav(tmp_path / "input.wav")
|
||||
interpreter = sys.executable
|
||||
payload = "import sys; open(sys.argv[2], 'w').write(sys.argv[1])"
|
||||
payload = "import sys; open(sys.argv[2], 'w', encoding='utf-8').write(sys.argv[1])"
|
||||
cfg = {
|
||||
"command": f'"{interpreter}" -c "{payload}" {{language}} {{output_path}}',
|
||||
}
|
||||
|
|
@ -486,6 +486,24 @@ class TestTranscribeAudioDispatchToCommandProvider:
|
|||
assert result["transcript"] == "dispatched via command"
|
||||
assert result["provider"] == "fake-cli"
|
||||
|
||||
def test_oversized_command_provider_file_is_rejected(self, tmp_path):
|
||||
from tools.transcription_tools import MAX_FILE_SIZE
|
||||
|
||||
audio = tmp_path / "oversized.wav"
|
||||
with audio.open("wb") as audio_file:
|
||||
audio_file.seek(MAX_FILE_SIZE)
|
||||
audio_file.write(b"\0")
|
||||
cfg = self._config_with_command_provider("fake-cli", "unused {input_path}")
|
||||
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value=cfg), \
|
||||
patch("tools.transcription_tools._transcribe_command_stt",
|
||||
return_value={"success": True, "transcript": "hi"}) as mock_command:
|
||||
result = transcribe_audio(str(audio))
|
||||
|
||||
assert result["success"] is False
|
||||
assert "File too large" in result["error"]
|
||||
mock_command.assert_not_called()
|
||||
|
||||
def test_builtin_name_shadow_does_not_route_to_command(self, tmp_path):
|
||||
# User mis-configures stt.providers.openai as a command — must NOT
|
||||
# hijack the real OpenAI built-in. The built-in elif chain owns
|
||||
|
|
|
|||
|
|
@ -66,6 +66,14 @@ def _reset_registry():
|
|||
transcription_registry._reset_for_tests()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_audio_file(tmp_path):
|
||||
"""Return a real, small input so E2E tests exercise validation too."""
|
||||
audio_path = tmp_path / "audio.mp3"
|
||||
audio_path.write_bytes(b"fake audio data")
|
||||
return str(audio_path)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Built-in always wins
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -209,42 +217,39 @@ class TestTranscribeAudioE2E:
|
|||
"""transcribe_audio() routes plugin dispatch correctly when the
|
||||
configured name is unknown to the built-in branches.
|
||||
|
||||
Note: we mock _validate_audio_file and _get_provider so the real
|
||||
file-validation and provider-resolution don't fire — we're testing
|
||||
the plugin-dispatch wiring, not those helpers.
|
||||
Provider resolution is mocked to isolate routing, while a real small
|
||||
input keeps the validation chain active.
|
||||
"""
|
||||
|
||||
def test_unknown_name_with_plugin_dispatches(self):
|
||||
def test_unknown_name_with_plugin_dispatches(self, sample_audio_file):
|
||||
from unittest.mock import patch
|
||||
provider = _FakeProvider(name="openrouter")
|
||||
transcription_registry.register_provider(provider)
|
||||
|
||||
with patch("tools.transcription_tools._validate_audio_file", return_value=None), \
|
||||
patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
|
||||
patch("tools.transcription_tools.is_stt_enabled", return_value=True), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openrouter"):
|
||||
result = transcription_tools.transcribe_audio("/tmp/audio.mp3")
|
||||
result = transcription_tools.transcribe_audio(sample_audio_file)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["transcript"] == "fake transcript"
|
||||
assert result["provider"] == "openrouter"
|
||||
|
||||
def test_unknown_name_without_plugin_falls_to_legacy_error(self):
|
||||
def test_unknown_name_without_plugin_falls_to_legacy_error(self, sample_audio_file):
|
||||
"""When no plugin is registered for the unknown name, the
|
||||
dispatcher returns None and transcribe_audio falls through to
|
||||
the legacy 'No STT provider available' error message."""
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch("tools.transcription_tools._validate_audio_file", return_value=None), \
|
||||
patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
|
||||
patch("tools.transcription_tools.is_stt_enabled", return_value=True), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openrouter"):
|
||||
result = transcription_tools.transcribe_audio("/tmp/audio.mp3")
|
||||
result = transcription_tools.transcribe_audio(sample_audio_file)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "No STT provider" in result["error"]
|
||||
|
||||
def test_builtin_name_does_not_consult_plugin_registry(self):
|
||||
def test_builtin_name_does_not_consult_plugin_registry(self, sample_audio_file):
|
||||
"""Even if a plugin's name collides with a built-in (which the
|
||||
registry blocks, but defense in depth matters), transcribe_audio
|
||||
with provider='groq' goes through the legacy elif chain, never
|
||||
|
|
@ -255,12 +260,11 @@ class TestTranscribeAudioE2E:
|
|||
provider = _FakeProvider(name="openrouter")
|
||||
transcription_registry.register_provider(provider)
|
||||
|
||||
with patch("tools.transcription_tools._validate_audio_file", return_value=None), \
|
||||
patch("tools.transcription_tools._load_stt_config", return_value={"provider": "groq"}), \
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "groq"}), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="groq"), \
|
||||
patch("tools.transcription_tools._transcribe_groq",
|
||||
return_value={"success": True, "transcript": "from groq", "provider": "groq"}) as mock_groq:
|
||||
result = transcription_tools.transcribe_audio("/tmp/audio.mp3")
|
||||
result = transcription_tools.transcribe_audio(sample_audio_file)
|
||||
|
||||
assert result["provider"] == "groq"
|
||||
assert result["transcript"] == "from groq"
|
||||
|
|
@ -268,6 +272,25 @@ class TestTranscribeAudioE2E:
|
|||
# Plugin was never called
|
||||
assert provider.last_call is None
|
||||
|
||||
def test_oversized_plugin_file_is_rejected_before_dispatch(self, tmp_path):
|
||||
from unittest.mock import patch
|
||||
|
||||
provider = _FakeProvider(name="openrouter")
|
||||
transcription_registry.register_provider(provider)
|
||||
audio_path = tmp_path / "oversized.mp3"
|
||||
with audio_path.open("wb") as audio_file:
|
||||
audio_file.seek(transcription_tools.MAX_FILE_SIZE)
|
||||
audio_file.write(b"\0")
|
||||
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
|
||||
patch("tools.transcription_tools.is_stt_enabled", return_value=True), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openrouter"):
|
||||
result = transcription_tools.transcribe_audio(str(audio_path))
|
||||
|
||||
assert result["success"] is False
|
||||
assert "File too large" in result["error"]
|
||||
assert provider.last_call is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Availability gating (codex review feedback on PR #30493)
|
||||
|
|
@ -330,7 +353,7 @@ class TestAvailabilityGate:
|
|||
assert "not available" in result["error"]
|
||||
assert provider.last_call is None
|
||||
|
||||
def test_unavailable_plugin_at_transcribe_audio_level(self):
|
||||
def test_unavailable_plugin_at_transcribe_audio_level(self, sample_audio_file):
|
||||
"""End-to-end: ``stt.provider: openrouter`` + plugin reports
|
||||
unavailable → ``transcribe_audio`` returns the unavailability
|
||||
envelope, NOT the generic "No STT provider available" message.
|
||||
|
|
@ -339,11 +362,10 @@ class TestAvailabilityGate:
|
|||
provider = _FakeProvider(name="openrouter", available=False)
|
||||
transcription_registry.register_provider(provider)
|
||||
|
||||
with patch("tools.transcription_tools._validate_audio_file", return_value=None), \
|
||||
patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
|
||||
patch("tools.transcription_tools.is_stt_enabled", return_value=True), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openrouter"):
|
||||
result = transcription_tools.transcribe_audio("/tmp/audio.mp3")
|
||||
result = transcription_tools.transcribe_audio(sample_audio_file)
|
||||
|
||||
assert result["success"] is False
|
||||
# Must surface the plugin's unavailability — NOT the generic
|
||||
|
|
@ -364,7 +386,7 @@ class TestLanguageForwardingFromConfig:
|
|||
``stt.local.language``).
|
||||
"""
|
||||
|
||||
def test_language_read_from_provider_namespaced_config(self):
|
||||
def test_language_read_from_provider_namespaced_config(self, sample_audio_file):
|
||||
"""``stt.openrouter.language: ja`` reaches the plugin's
|
||||
transcribe() call as language='ja'."""
|
||||
from unittest.mock import patch
|
||||
|
|
@ -375,16 +397,15 @@ class TestLanguageForwardingFromConfig:
|
|||
"provider": "openrouter",
|
||||
"openrouter": {"language": "ja"},
|
||||
}
|
||||
with patch("tools.transcription_tools._validate_audio_file", return_value=None), \
|
||||
patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \
|
||||
patch("tools.transcription_tools.is_stt_enabled", return_value=True), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openrouter"):
|
||||
transcription_tools.transcribe_audio("/tmp/audio.mp3")
|
||||
transcription_tools.transcribe_audio(sample_audio_file)
|
||||
|
||||
assert provider.last_call is not None
|
||||
assert provider.last_call["kwargs"]["language"] == "ja"
|
||||
|
||||
def test_model_from_provider_namespaced_config(self):
|
||||
def test_model_from_provider_namespaced_config(self, sample_audio_file):
|
||||
"""``stt.openrouter.model: whisper-large-v3`` reaches the
|
||||
plugin as model='whisper-large-v3' when caller doesn't
|
||||
override."""
|
||||
|
|
@ -396,15 +417,14 @@ class TestLanguageForwardingFromConfig:
|
|||
"provider": "openrouter",
|
||||
"openrouter": {"model": "whisper-large-v3"},
|
||||
}
|
||||
with patch("tools.transcription_tools._validate_audio_file", return_value=None), \
|
||||
patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \
|
||||
patch("tools.transcription_tools.is_stt_enabled", return_value=True), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openrouter"):
|
||||
transcription_tools.transcribe_audio("/tmp/audio.mp3")
|
||||
transcription_tools.transcribe_audio(sample_audio_file)
|
||||
|
||||
assert provider.last_call["kwargs"]["model"] == "whisper-large-v3"
|
||||
|
||||
def test_caller_model_overrides_config_model(self):
|
||||
def test_caller_model_overrides_config_model(self, sample_audio_file):
|
||||
"""An explicit ``model`` arg to transcribe_audio wins over
|
||||
``stt.<provider>.model`` in config."""
|
||||
from unittest.mock import patch
|
||||
|
|
@ -415,33 +435,31 @@ class TestLanguageForwardingFromConfig:
|
|||
"provider": "openrouter",
|
||||
"openrouter": {"model": "config-model"},
|
||||
}
|
||||
with patch("tools.transcription_tools._validate_audio_file", return_value=None), \
|
||||
patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \
|
||||
patch("tools.transcription_tools.is_stt_enabled", return_value=True), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openrouter"):
|
||||
transcription_tools.transcribe_audio(
|
||||
"/tmp/audio.mp3", model="explicit-arg-model",
|
||||
sample_audio_file, model="explicit-arg-model",
|
||||
)
|
||||
|
||||
assert provider.last_call["kwargs"]["model"] == "explicit-arg-model"
|
||||
|
||||
def test_missing_provider_namespace_passes_none(self):
|
||||
def test_missing_provider_namespace_passes_none(self, sample_audio_file):
|
||||
"""No ``stt.<provider>`` subsection → language is None,
|
||||
model falls back to caller arg or None. No crash."""
|
||||
from unittest.mock import patch
|
||||
provider = _FakeProvider(name="openrouter")
|
||||
transcription_registry.register_provider(provider)
|
||||
|
||||
with patch("tools.transcription_tools._validate_audio_file", return_value=None), \
|
||||
patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openrouter"}), \
|
||||
patch("tools.transcription_tools.is_stt_enabled", return_value=True), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openrouter"):
|
||||
transcription_tools.transcribe_audio("/tmp/audio.mp3")
|
||||
transcription_tools.transcribe_audio(sample_audio_file)
|
||||
|
||||
assert provider.last_call["kwargs"]["language"] is None
|
||||
assert provider.last_call["kwargs"]["model"] is None
|
||||
|
||||
def test_non_dict_provider_namespace_does_not_crash(self):
|
||||
def test_non_dict_provider_namespace_does_not_crash(self, sample_audio_file):
|
||||
"""If someone accidentally writes ``stt.openrouter: "foo"`` (a
|
||||
string instead of a dict), we should not crash — treat as
|
||||
empty config."""
|
||||
|
|
@ -450,11 +468,10 @@ class TestLanguageForwardingFromConfig:
|
|||
transcription_registry.register_provider(provider)
|
||||
|
||||
stt_config = {"provider": "openrouter", "openrouter": "garbage"}
|
||||
with patch("tools.transcription_tools._validate_audio_file", return_value=None), \
|
||||
patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value=stt_config), \
|
||||
patch("tools.transcription_tools.is_stt_enabled", return_value=True), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openrouter"):
|
||||
result = transcription_tools.transcribe_audio("/tmp/audio.mp3")
|
||||
result = transcription_tools.transcribe_audio(sample_audio_file)
|
||||
|
||||
# Should still dispatch successfully (config is just ignored)
|
||||
assert result["success"] is True
|
||||
|
|
|
|||
|
|
@ -54,6 +54,18 @@ def sample_ogg(tmp_path):
|
|||
return str(ogg_path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def oversized_wav(tmp_path):
|
||||
"""Create a sparse WAV-shaped file just above the remote upload cap."""
|
||||
from tools.transcription_tools import MAX_FILE_SIZE
|
||||
|
||||
wav_path = tmp_path / "oversized.wav"
|
||||
with wav_path.open("wb") as audio_file:
|
||||
audio_file.seek(MAX_FILE_SIZE)
|
||||
audio_file.write(b"\0")
|
||||
return str(wav_path)
|
||||
|
||||
|
||||
pytestmark = pytest.mark.usefixtures("disable_lazy_stt_install")
|
||||
|
||||
|
||||
|
|
@ -1110,6 +1122,39 @@ class TestTranscribeAudioDispatch:
|
|||
assert result["success"] is True
|
||||
mock_local.assert_called_once()
|
||||
|
||||
def test_oversized_local_file_reaches_dispatcher(self, oversized_wav):
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "local"}), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="local"), \
|
||||
patch("tools.transcription_tools._transcribe_local",
|
||||
return_value={"success": True, "transcript": "hi"}) as mock_local:
|
||||
from tools.transcription_tools import transcribe_audio
|
||||
result = transcribe_audio(oversized_wav)
|
||||
|
||||
assert result["success"] is True
|
||||
mock_local.assert_called_once()
|
||||
|
||||
def test_oversized_local_command_file_reaches_dispatcher(self, oversized_wav):
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "local_command"}), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="local_command"), \
|
||||
patch("tools.transcription_tools._transcribe_local_command",
|
||||
return_value={"success": True, "transcript": "hi"}) as mock_command:
|
||||
from tools.transcription_tools import transcribe_audio
|
||||
result = transcribe_audio(oversized_wav)
|
||||
|
||||
assert result["success"] is True
|
||||
mock_command.assert_called_once()
|
||||
|
||||
def test_oversized_remote_file_is_rejected_before_dispatch(self, oversized_wav):
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openai"}), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openai"), \
|
||||
patch("tools.transcription_tools._transcribe_openai") as mock_openai:
|
||||
from tools.transcription_tools import transcribe_audio
|
||||
result = transcribe_audio(oversized_wav)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "File too large" in result["error"]
|
||||
mock_openai.assert_not_called()
|
||||
|
||||
def test_dispatches_to_openai(self, sample_ogg):
|
||||
with patch("tools.transcription_tools._load_stt_config", return_value={"provider": "openai"}), \
|
||||
patch("tools.transcription_tools._get_provider", return_value="openai"), \
|
||||
|
|
|
|||
|
|
@ -837,9 +837,20 @@ class TestTranscribeRecording:
|
|||
monkeypatch.setattr("tools.voice_mode._TEMP_DIR", str(temp_dir))
|
||||
monkeypatch.setattr("tools.transcription_tools.MAX_FILE_SIZE", 70 * 1024)
|
||||
|
||||
call_count = 0
|
||||
seen_paths = []
|
||||
|
||||
def fake_transcribe(path, model=None):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
# First call is on the original file — simulate remote provider
|
||||
# rejecting it as too large so chunking kicks in.
|
||||
if call_count == 1:
|
||||
return {
|
||||
"success": False,
|
||||
"transcript": "",
|
||||
"error": "File too large: 0.1MB (max 0.1MB)",
|
||||
}
|
||||
seen_paths.append(path)
|
||||
assert model == "base"
|
||||
assert path != str(wav_path)
|
||||
|
|
@ -877,7 +888,17 @@ class TestTranscribeRecording:
|
|||
monkeypatch.setattr("tools.voice_mode._TEMP_DIR", str(temp_dir))
|
||||
monkeypatch.setattr("tools.transcription_tools.MAX_FILE_SIZE", 70 * 1024)
|
||||
|
||||
call_count = 0
|
||||
|
||||
def fake_transcribe(path, model=None):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count == 1:
|
||||
return {
|
||||
"success": False,
|
||||
"transcript": "",
|
||||
"error": "File too large: 0.1MB (max 0.1MB)",
|
||||
}
|
||||
return {"success": False, "transcript": "", "error": "provider rejected audio"}
|
||||
|
||||
with patch("tools.transcription_tools.transcribe_audio", side_effect=fake_transcribe):
|
||||
|
|
@ -889,6 +910,97 @@ class TestTranscribeRecording:
|
|||
assert "provider rejected audio" in result["error"]
|
||||
assert list(temp_dir.iterdir()) == []
|
||||
|
||||
def test_trusts_transcribe_audio_skip_chunk_for_local(self, tmp_path, monkeypatch):
|
||||
"""Local providers never return 'File too large' — no chunking."""
|
||||
wav_path = tmp_path / "record.wav"
|
||||
n_frames = 50000
|
||||
audio = struct.pack(f"<{n_frames}h", *([1000] * n_frames))
|
||||
with wave.open(str(wav_path), "wb") as wf:
|
||||
wf.setnchannels(1)
|
||||
wf.setsampwidth(2)
|
||||
wf.setframerate(16000)
|
||||
wf.writeframes(audio)
|
||||
|
||||
mock_transcribe = MagicMock(return_value={
|
||||
"success": True,
|
||||
"transcript": "local whisper result",
|
||||
"provider": "local",
|
||||
})
|
||||
|
||||
with patch("tools.transcription_tools.transcribe_audio", mock_transcribe):
|
||||
from tools.voice_mode import transcribe_recording
|
||||
result = transcribe_recording(str(wav_path), model="base")
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["transcript"] == "local whisper result"
|
||||
assert "chunks" not in result
|
||||
mock_transcribe.assert_called_once_with(str(wav_path), model="base")
|
||||
|
||||
def test_chunks_when_transcribe_audio_returns_file_too_large(self, tmp_path, monkeypatch):
|
||||
"""Remote provider rejects large file → chunking fallback."""
|
||||
wav_path = tmp_path / "record.wav"
|
||||
n_frames = 50000
|
||||
audio = struct.pack(f"<{n_frames}h", *([1000] * n_frames))
|
||||
with wave.open(str(wav_path), "wb") as wf:
|
||||
wf.setnchannels(1)
|
||||
wf.setsampwidth(2)
|
||||
wf.setframerate(16000)
|
||||
wf.writeframes(audio)
|
||||
|
||||
temp_dir = tmp_path / "chunks"
|
||||
temp_dir.mkdir()
|
||||
monkeypatch.setattr("tools.voice_mode._TEMP_DIR", str(temp_dir))
|
||||
monkeypatch.setattr("tools.transcription_tools.MAX_FILE_SIZE", 70 * 1024)
|
||||
|
||||
call_count = 0
|
||||
|
||||
def fake_transcribe(path, model=None):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count == 1:
|
||||
return {
|
||||
"success": False,
|
||||
"transcript": "",
|
||||
"error": "File too large: 30.0MB (max 25MB)",
|
||||
}
|
||||
return {
|
||||
"success": True,
|
||||
"transcript": f"chunk {call_count - 1}",
|
||||
"provider": "openai",
|
||||
}
|
||||
|
||||
with patch("tools.transcription_tools.transcribe_audio", side_effect=fake_transcribe):
|
||||
from tools.voice_mode import transcribe_recording
|
||||
result = transcribe_recording(str(wav_path), model="whisper-1")
|
||||
|
||||
assert result["success"] is True
|
||||
assert result.get("chunks", 0) > 1
|
||||
|
||||
def test_other_error_does_not_trigger_chunk(self, tmp_path, monkeypatch):
|
||||
"""Non-size errors from transcribe_audio are returned as-is."""
|
||||
wav_path = tmp_path / "record.wav"
|
||||
n_frames = 50000
|
||||
audio = struct.pack(f"<{n_frames}h", *([1000] * n_frames))
|
||||
with wave.open(str(wav_path), "wb") as wf:
|
||||
wf.setnchannels(1)
|
||||
wf.setsampwidth(2)
|
||||
wf.setframerate(16000)
|
||||
wf.writeframes(audio)
|
||||
|
||||
mock_transcribe = MagicMock(return_value={
|
||||
"success": False,
|
||||
"transcript": "",
|
||||
"error": "STT is disabled in config.yaml",
|
||||
})
|
||||
|
||||
with patch("tools.transcription_tools.transcribe_audio", mock_transcribe):
|
||||
from tools.voice_mode import transcribe_recording
|
||||
result = transcribe_recording(str(wav_path), model="base")
|
||||
|
||||
assert result["success"] is False
|
||||
assert "STT is disabled" in result["error"]
|
||||
mock_transcribe.assert_called_once()
|
||||
|
||||
|
||||
class TestWhisperHallucinationFilter:
|
||||
def test_known_hallucinations(self):
|
||||
|
|
|
|||
|
|
@ -393,6 +393,14 @@ def _resolve_command_stt_provider_config(
|
|||
return None
|
||||
|
||||
|
||||
def _is_local_stt_provider(provider: str, stt_config: Dict[str, Any]) -> bool:
|
||||
"""Return whether *provider* is exempt from Hermes's remote upload cap."""
|
||||
key = (provider or "").lower().strip()
|
||||
if key in {"local", "local_command"}:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _iter_command_stt_providers(stt_config: Dict[str, Any]):
|
||||
"""Yield (name, config) pairs for every declared command-type STT provider."""
|
||||
if not isinstance(stt_config, dict):
|
||||
|
|
@ -1094,7 +1102,26 @@ def _dispatch_to_plugin_provider(
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _validate_audio_file(file_path: str) -> Optional[Dict[str, Any]]:
|
||||
def _validate_audio_file_size(audio_path: Path) -> Optional[Dict[str, Any]]:
|
||||
"""Return an error when *audio_path* exceeds the remote upload cap."""
|
||||
try:
|
||||
file_size = audio_path.stat().st_size
|
||||
except OSError as e:
|
||||
return {"success": False, "transcript": "", "error": f"Failed to access file: {e}"}
|
||||
if file_size > MAX_FILE_SIZE:
|
||||
return {
|
||||
"success": False,
|
||||
"transcript": "",
|
||||
"error": f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)",
|
||||
}
|
||||
return None
|
||||
|
||||
|
||||
def _validate_audio_file(
|
||||
file_path: str,
|
||||
*,
|
||||
enforce_size_limit: bool = True,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Validate the audio file. Returns an error dict or None if OK."""
|
||||
audio_path = Path(file_path)
|
||||
|
||||
|
|
@ -1110,17 +1137,12 @@ def _validate_audio_file(file_path: str) -> Optional[Dict[str, Any]]:
|
|||
"transcript": "",
|
||||
"error": f"Unsupported format: {audio_path.suffix}. Supported: {', '.join(sorted(SUPPORTED_FORMATS))}",
|
||||
}
|
||||
if enforce_size_limit:
|
||||
return _validate_audio_file_size(audio_path)
|
||||
try:
|
||||
file_size = audio_path.stat().st_size
|
||||
if file_size > MAX_FILE_SIZE:
|
||||
return {
|
||||
"success": False,
|
||||
"transcript": "",
|
||||
"error": f"File too large: {file_size / (1024*1024):.1f}MB (max {MAX_FILE_SIZE / (1024*1024):.0f}MB)",
|
||||
}
|
||||
audio_path.stat()
|
||||
except OSError as e:
|
||||
return {"success": False, "transcript": "", "error": f"Failed to access file: {e}"}
|
||||
|
||||
return None
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -1874,8 +1896,9 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A
|
|||
- "error" (str, optional): Error message if success is False
|
||||
- "provider" (str, optional): Which provider was used
|
||||
"""
|
||||
# Validate input
|
||||
error = _validate_audio_file(file_path)
|
||||
# Apply common path validation before provider resolution so invalid files
|
||||
# cannot trigger provider setup or lazy installation.
|
||||
error = _validate_audio_file(file_path, enforce_size_limit=False)
|
||||
if error:
|
||||
return error
|
||||
|
||||
|
|
@ -1889,6 +1912,10 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A
|
|||
}
|
||||
|
||||
provider = _get_provider(stt_config)
|
||||
if not _is_local_stt_provider(provider, stt_config):
|
||||
error = _validate_audio_file_size(Path(file_path))
|
||||
if error:
|
||||
return error
|
||||
|
||||
if provider == "local":
|
||||
local_cfg = stt_config.get("local") or {}
|
||||
|
|
|
|||
|
|
@ -941,10 +941,13 @@ def transcribe_recording(wav_path: str, model: Optional[str] = None) -> Dict[str
|
|||
"""
|
||||
from tools.transcription_tools import MAX_FILE_SIZE, transcribe_audio
|
||||
|
||||
if _should_chunk_for_transcription(wav_path, MAX_FILE_SIZE):
|
||||
result = transcribe_audio(wav_path, model=model)
|
||||
|
||||
# Only chunk when the provider itself reports "File too large" —
|
||||
# local providers (faster-whisper, whisper.cpp, etc.) have no upload
|
||||
# cap so ``transcribe_audio`` will never return this error for them.
|
||||
if not result.get("success") and "File too large" in result.get("error", ""):
|
||||
result = _transcribe_wav_in_chunks(wav_path, model=model, max_file_size=MAX_FILE_SIZE)
|
||||
else:
|
||||
result = transcribe_audio(wav_path, model=model)
|
||||
|
||||
# Filter out Whisper hallucinations (common on silent/near-silent audio)
|
||||
if result.get("success") and is_whisper_hallucination(result.get("transcript", "")):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue