diff --git a/src/engine/audio.py b/src/engine/audio.py new file mode 100644 index 00000000..cf650800 --- /dev/null +++ b/src/engine/audio.py @@ -0,0 +1,5 @@ +"""Audio helpers shared by the speech engines.""" + + +class AudioDecodeError(ValueError): + """The request carried audio that cannot be decoded: a client error, not a model failure.""" diff --git a/src/engine/openvino/qwen3_asr/qwen3_asr_utils.py b/src/engine/openvino/qwen3_asr/qwen3_asr_utils.py index c60c3e1d..6f89f559 100644 --- a/src/engine/openvino/qwen3_asr/qwen3_asr_utils.py +++ b/src/engine/openvino/qwen3_asr/qwen3_asr_utils.py @@ -24,6 +24,8 @@ import numpy as np import soundfile as sf +from src.engine.audio import AudioDecodeError + AudioLike = Union[ str, # wav path / URL / base64 Tuple[np.ndarray, int], # (waveform, sr) @@ -198,6 +200,15 @@ def decode_base64_bytes(b64: str) -> bytes: def load_audio_any(x: str) -> Tuple[np.ndarray, int]: + try: + return _load_audio_any(x) + except AudioDecodeError: + raise + except Exception as e: + raise AudioDecodeError(f"Unreadable audio: {e}") from e + + +def _load_audio_any(x: str) -> Tuple[np.ndarray, int]: if is_url(x): with urllib.request.urlopen(x) as resp: audio_bytes = resp.read() diff --git a/src/engine/ov_genai/whisper.py b/src/engine/ov_genai/whisper.py index db5814c5..b12af9ff 100644 --- a/src/engine/ov_genai/whisper.py +++ b/src/engine/ov_genai/whisper.py @@ -9,6 +9,7 @@ import numpy as np from openvino_genai import WhisperPipeline +from src.engine.audio import AudioDecodeError from src.server.model_registry import ModelRegistry from src.server.schemas.registration import ModelLoadConfig from src.server.schemas.modeling.contract_whisper import OVGenAI_WhisperGenConfig @@ -31,7 +32,10 @@ def prepare_audio(self, gen_config: OVGenAI_WhisperGenConfig) -> list[float]: audio_buffer = io.BytesIO(audio_bytes) - audio, sr = librosa.load(audio_buffer, sr=16000, mono=True) + try: + audio, sr = librosa.load(audio_buffer, sr=16000, mono=True) + except Exception as e: + raise AudioDecodeError(f"Unreadable audio: {e}") from e return audio.astype(np.float32).tolist() diff --git a/src/server/worker_registry.py b/src/server/worker_registry.py index 37de820f..f13638e9 100644 --- a/src/server/worker_registry.py +++ b/src/server/worker_registry.py @@ -12,6 +12,7 @@ from src.engine.ov_genai.llm import OVGenAI_LLM from src.engine.ov_genai.vlm import OVGenAI_VLM from src.engine.ov_genai.whisper import OVGenAI_Whisper +from src.engine.audio import AudioDecodeError from src.engine.openvino.kokoro import OV_Kokoro from src.engine.openvino.qwen3_asr.qwen3_asr import OVQwen3ASR from src.engine.openvino.qwen3_tts.qwen3_tts import OVQwen3TTS @@ -70,6 +71,8 @@ class WorkerPacket: response: Optional[str] = None metrics: Optional[Dict[str, Any]] = None segments: Optional[List[Dict[str, Any]]] = None + # A bad request (e.g. undecodable audio): fail this request, keep the model loaded. + request_error: Optional[BaseException] = None # Orchestration plumbing result_future: Optional[asyncio.Future] = None stream_queue: Optional[asyncio.Queue] = None @@ -96,6 +99,10 @@ def _commit_completed_packet( registry: ModelRegistry, ) -> bool: """Complete the request future. Return True if the worker should exit.""" + if completed.request_error is not None: + if packet.result_future is not None and not packet.result_future.done(): + packet.result_future.set_exception(completed.request_error) + return False if completed.error is not None: logger.error( f"[{model_name}] Inference failed, triggering model unload..." @@ -214,6 +221,9 @@ async def infer_whisper(packet: WorkerPacket, whisper_model: OVGenAI_Whisper) -> packet.response = final_text packet.metrics = metrics + except AudioDecodeError as e: + logger.warning(f"Whisper request rejected: {e}") + packet.request_error = e except Exception as e: logger.error("Whisper inference failed!", exc_info=True) _mark_inference_error(packet, e) @@ -228,6 +238,9 @@ async def infer_qwen3_asr(packet: WorkerPacket, asr_model: OVQwen3ASR) -> Worker try: assert isinstance(packet.gen_config, OV_Qwen3ASRGenConfig), "Expected OV_Qwen3ASRGenConfig for Qwen3 ASR inference" packet.response, packet.metrics, packet.segments = await asr_model.transcribe(packet.gen_config) + except AudioDecodeError as e: + logger.warning(f"Qwen3 ASR request rejected: {e}") + packet.request_error = e except Exception as e: logger.error("Qwen3 ASR inference failed!", exc_info=True) _mark_inference_error(packet, e) diff --git a/tests/unit/test_audio_bad_input_unit.py b/tests/unit/test_audio_bad_input_unit.py new file mode 100644 index 00000000..8b823e80 --- /dev/null +++ b/tests/unit/test_audio_bad_input_unit.py @@ -0,0 +1,16 @@ +import base64 + +import pytest # type: ignore[import] + +from src.engine.audio import AudioDecodeError +from src.engine.openvino.qwen3_asr.qwen3_asr_utils import load_audio_any + + +def test_qwen3_load_audio_rejects_garbage_base64() -> None: + with pytest.raises(AudioDecodeError, match="Unreadable audio"): + load_audio_any(base64.b64encode(b"not audio" * 100).decode()) + + +def test_decode_error_is_a_value_error() -> None: + # routes map ValueError to HTTP 400 + assert issubclass(AudioDecodeError, ValueError) diff --git a/tests/unit/test_ov_genai_whisper_unit.py b/tests/unit/test_ov_genai_whisper_unit.py index ff2d2f41..155caede 100644 --- a/tests/unit/test_ov_genai_whisper_unit.py +++ b/tests/unit/test_ov_genai_whisper_unit.py @@ -152,3 +152,10 @@ def test_unload_model_resets_state(monkeypatch: pytest.MonkeyPatch, load_config: registry.register_unload.assert_called_once_with("model-name") gc_mock.assert_called_once() + +def test_prepare_audio_rejects_garbage(load_config: ModelLoadConfig) -> None: + from src.engine.audio import AudioDecodeError + + whisper = OVGenAI_Whisper(load_config) + with pytest.raises(AudioDecodeError, match="Unreadable audio"): + whisper.prepare_audio(OVGenAI_WhisperGenConfig(audio_base64=base64.b64encode(b"not audio" * 100).decode())) diff --git a/tests/unit/test_worker_registry_unit.py b/tests/unit/test_worker_registry_unit.py index b4c874ba..6333a666 100644 --- a/tests/unit/test_worker_registry_unit.py +++ b/tests/unit/test_worker_registry_unit.py @@ -1,4 +1,5 @@ import asyncio +from unittest.mock import AsyncMock, MagicMock import pytest # type: ignore[import] @@ -404,6 +405,44 @@ def test_missing_model_queue(worker_registry: worker_module.WorkerRegistry) -> N +def test_undecodable_audio_fails_whisper_and_qwen3_requests_not_models() -> None: + from src.engine.audio import AudioDecodeError + + class BadWhisper: + async def transcribe(self, cfg): + raise AudioDecodeError("Unreadable audio: nope") + yield # pragma: no cover + + class BadQwen3: + async def transcribe(self, cfg): + raise AudioDecodeError("Unreadable audio: nope") + + w = worker_module.WorkerPacket(request_id="w", id_model="whisper", gen_config=OVGenAI_WhisperGenConfig(audio_base64="AAA")) + done = asyncio.run(worker_module.InferWorker.infer_whisper(w, BadWhisper())) # type: ignore[arg-type] + assert done.error is None and isinstance(done.request_error, AudioDecodeError) + + q = worker_module.WorkerPacket(request_id="q", id_model="qwen3", gen_config=OV_Qwen3ASRGenConfig(audio_base64="AAA")) + done = asyncio.run(worker_module.InferWorker.infer_qwen3_asr(q, BadQwen3())) # type: ignore[arg-type] + assert done.error is None and isinstance(done.request_error, AudioDecodeError) + + +def test_commit_request_error_fails_future_and_keeps_worker() -> None: + async def _run(): + registry = MagicMock() + registry.register_unload = AsyncMock() + fut = asyncio.get_running_loop().create_future() + packet = worker_module.WorkerPacket(request_id="r", id_model="m", gen_config=OVGenAI_WhisperGenConfig(audio_base64="AAA"), + result_future=fut) + packet.request_error = ValueError("bad audio") + exit_worker = worker_module._commit_completed_packet(packet, packet, "m", registry) + await asyncio.sleep(0) + return exit_worker, fut, registry + + exit_worker, fut, registry = asyncio.run(_run()) + assert exit_worker is False + with pytest.raises(ValueError, match="bad audio"): + fut.result() + registry.register_unload.assert_not_called() def test_infer_kokoro_stream_puts_pcm_per_chunk_then_none() -> None: import numpy as np import torch