diff --git a/ci/Jenkinsfile.hermes-voice-image b/ci/Jenkinsfile.hermes-voice-image index f652c0e8..7db21c0c 100644 --- a/ci/Jenkinsfile.hermes-voice-image +++ b/ci/Jenkinsfile.hermes-voice-image @@ -156,6 +156,7 @@ spec: dockerfiles/hermes_jetson_tts_cues.py PYTHONPATH=/tmp/hermes-voice-test-deps python3 -m pytest -q \ testing/tests/test_hermes_stt_streaming.py \ + testing/tests/test_hermes_stt_rolling_model.py \ testing/tests/test_hermes_tts_language_routing.py \ testing/tests/test_hermes_voice_language_routing.py \ testing/tests/test_hermes_oci_promote.py \ diff --git a/dockerfiles/Dockerfile.hermes-jetson-stt b/dockerfiles/Dockerfile.hermes-jetson-stt index 63da6b9a..f98b13f6 100644 --- a/dockerfiles/Dockerfile.hermes-jetson-stt +++ b/dockerfiles/Dockerfile.hermes-jetson-stt @@ -17,6 +17,9 @@ ADD --checksum=sha256:aff26ae408abcba5fbf8813c21e62b0941638c5f6eebfb145be0c98392 ADD --checksum=sha256:9ecf779972d90ba49c06d968637d720dd632c55bbf19d441fb42bf17a411e794 --chmod=0444 \ https://openaipublic.azureedge.net/main/whisper/models/9ecf779972d90ba49c06d968637d720dd632c55bbf19d441fb42bf17a411e794/small.pt \ /opt/models/whisper/small.pt +ADD --checksum=sha256:65147644a518d12f04e32d6f3b26facc3f8dd46e5390956a9424a650c0ce22b9 --chmod=0444 \ + https://openaipublic.azureedge.net/main/whisper/models/65147644a518d12f04e32d6f3b26facc3f8dd46e5390956a9424a650c0ce22b9/tiny.pt \ + /opt/models/whisper/tiny.pt RUN chmod 0555 /opt/models /opt/models/whisper COPY dockerfiles/hermes-jetson-stt-server.py /opt/atlas/hermes-jetson-stt-server.py @@ -28,11 +31,12 @@ WORKDIR /opt/atlas # Import the Xavier CUDA stack and confirm Whisper resolves the baked artifact. # Full GPU warm-up is covered by the Kubernetes startup probe on titan-21. -RUN python3 -c "import stat; from pathlib import Path; import torch, whisper; p=Path('/opt/models/whisper'); print(whisper.__file__, whisper.__version__, whisper.available_models()); assert {'large-v3-turbo','small'} <= set(whisper.available_models()); assert stat.S_IMODE(p.stat().st_mode)==0o555; assert all(stat.S_IMODE((p/name).stat().st_mode)==0o444 for name in ('large-v3-turbo.pt','small.pt')); print(torch.__version__)" +RUN python3 -c "import stat; from pathlib import Path; import torch, whisper; p=Path('/opt/models/whisper'); print(whisper.__file__, whisper.__version__, whisper.available_models()); assert {'large-v3-turbo','small','tiny'} <= set(whisper.available_models()); assert stat.S_IMODE(p.stat().st_mode)==0o555; assert all(stat.S_IMODE((p/name).stat().st_mode)==0o444 for name in ('large-v3-turbo.pt','small.pt','tiny.pt')); print(torch.__version__)" ENV HERMES_STT_HOST=0.0.0.0 \ HERMES_STT_PORT=9000 \ HERMES_STT_MODEL=small \ + HERMES_STT_ROLLING_MODEL=tiny \ HERMES_STT_CACHE=/opt/models/whisper \ PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 diff --git a/dockerfiles/hermes-jetson-stt-server.py b/dockerfiles/hermes-jetson-stt-server.py index 79084ad2..6fa79bad 100644 --- a/dockerfiles/hermes-jetson-stt-server.py +++ b/dockerfiles/hermes-jetson-stt-server.py @@ -25,6 +25,7 @@ import whisper HOST = os.getenv("HERMES_STT_HOST", "0.0.0.0") PORT = int(os.getenv("HERMES_STT_PORT", "9000")) MODEL_NAME = os.getenv("HERMES_STT_MODEL", "small") +ROLLING_MODEL_NAME = os.getenv("HERMES_STT_ROLLING_MODEL", "tiny") CACHE_DIR = Path(os.getenv("HERMES_STT_CACHE", "/cache/whisper")) MAX_AUDIO_BYTES = 30 * 1024 * 1024 STREAM_SAMPLE_RATE = 16_000 @@ -68,6 +69,8 @@ class InferenceGate: self._last_final_wait_ms = 0.0 self._last_final_decode_ms = 0.0 self._last_background_decode_ms = 0.0 + self._last_rolling_decode_ms = 0.0 + self._last_speculative_decode_ms = 0.0 self._background_skips = 0 def run_final(self, callback): @@ -87,14 +90,14 @@ class InferenceGate: return callback() finally: with self._condition: - self._last_final_decode_ms = ( - time.monotonic() - decode_started - ) * 1000 + self._last_final_decode_ms = (time.monotonic() - decode_started) * 1000 self._busy = False self._condition.notify_all() - def try_background(self, callback): + def try_background(self, callback, *, kind: str): """Run one useful predecode only when no final is active or waiting.""" + if kind not in {"rolling", "speculative"}: + raise ValueError("unknown background inference kind") with self._condition: if self._busy or self._final_waiters: self._background_skips += 1 @@ -105,9 +108,12 @@ class InferenceGate: return callback() finally: with self._condition: - self._last_background_decode_ms = ( - time.monotonic() - decode_started - ) * 1000 + elapsed_ms = (time.monotonic() - decode_started) * 1000 + self._last_background_decode_ms = elapsed_ms + if kind == "rolling": + self._last_rolling_decode_ms = elapsed_ms + else: + self._last_speculative_decode_ms = elapsed_ms self._busy = False self._condition.notify_all() @@ -119,8 +125,10 @@ class InferenceGate: "final_waiters": self._final_waiters, "last_final_wait_ms": round(self._last_final_wait_ms, 1), "last_final_decode_ms": round(self._last_final_decode_ms, 1), - "last_background_decode_ms": round( - self._last_background_decode_ms, 1 + "last_background_decode_ms": round(self._last_background_decode_ms, 1), + "last_rolling_decode_ms": round(self._last_rolling_decode_ms, 1), + "last_speculative_decode_ms": round( + self._last_speculative_decode_ms, 1 ), "background_skips": self._background_skips, } @@ -128,6 +136,7 @@ class InferenceGate: INFERENCE_GATE = InferenceGate() STARTUP_WARMUP_MS: float | None = None +STARTUP_ROLLING_WARMUP_MS: float | None = None def _repetitive_token(token: str) -> bool: @@ -264,20 +273,19 @@ def _warm_model(model: object) -> float: return (time.monotonic() - started) * 1000 -def _transcribe_pcm_rolling( - model: object, pcm: bytes, language: str -) -> dict | None: +def _transcribe_pcm_rolling(model: object, pcm: bytes, language: str) -> dict | None: """Decode a rolling window only when the single GPU worker is immediately free. Partial work is disposable. It must never form an inference backlog ahead of a final utterance or another user's request, so a busy model means this - snapshot is skipped and a later audio frame may try again. + speech epoch is skipped rather than retried. """ temp_path = "" try: temp_path = _write_pcm_wav(pcm) return INFERENCE_GATE.try_background( - lambda: _decode_path(model, temp_path, language) + lambda: _decode_path(model, temp_path, language), + kind="rolling", ) finally: if temp_path: @@ -295,7 +303,8 @@ def _transcribe_pcm_speculative( try: temp_path = _write_pcm_wav(pcm) return INFERENCE_GATE.try_background( - lambda: _decode_path(model, temp_path, language) + lambda: _decode_path(model, temp_path, language), + kind="speculative", ) finally: if temp_path: @@ -462,9 +471,15 @@ def _pcm_rms(pcm: bytes) -> float: class StreamingTranscription: """Own one turn's bounded PCM, VAD state and speculative Whisper work.""" - def __init__(self, connection: WebSocketConnection, model: object) -> None: + def __init__( + self, + connection: WebSocketConnection, + model: object, + rolling_model: object | None = None, + ) -> None: self.connection = connection self.model = model + self.rolling_model = rolling_model if rolling_model is not None else model self.turn_id = "" self.language = "auto" self._pcm = bytearray() @@ -489,6 +504,7 @@ class StreamingTranscription: self._rolling_previous = "" self._rolling_previous_window_start = 0 self._rolling_revision = 0 + self._rolling_attempted = False def _response(self, message_type: str, **values: object) -> dict: return {"type": message_type, "turn_id": self.turn_id, **values} @@ -543,7 +559,7 @@ class StreamingTranscription: start_ms = start * 1000 // (STREAM_SAMPLE_RATE * STREAM_SAMPLE_WIDTH) return pcm, start_ms - def _reset_rolling_history_locked(self) -> None: + def _reset_rolling_history_locked(self, *, rearm: bool) -> None: """Discard provisional agreement after the client's speech epoch changes.""" self._rolling_previous = "" self._rolling_previous_window_start = 0 @@ -551,10 +567,10 @@ class StreamingTranscription: self._rolling_speech_bytes = 0 self._rolling_last_speech_bytes = 0 self._rolling_last_started_at = 0.0 + if rearm: + self._rolling_attempted = False - def _begin_rolling_locked( - self, now: float - ) -> tuple[bytes, int, int, int] | None: + def _begin_rolling_locked(self, now: float) -> tuple[bytes, int, int, int] | None: """Reserve one due rolling decode without ever queuing a second one.""" total_bytes = len(self._pcm) minimum_bytes = _pcm_bytes_for_ms(ROLLING_MIN_AUDIO_MS) @@ -564,17 +580,20 @@ class StreamingTranscription: ) if ( self._rolling_inflight + or self._rolling_attempted or self._rolling_speech_bytes < minimum_bytes or self._rolling_speech_bytes - self._rolling_last_speech_bytes < required_new_bytes or ( self._rolling_last_started_at - and (now - self._rolling_last_started_at) * 1000 - < ROLLING_INTERVAL_MS + and (now - self._rolling_last_started_at) * 1000 < ROLLING_INTERVAL_MS ) ): return None pcm, window_start_ms = self._rolling_snapshot_locked() + # Reserve the epoch before spawning. A busy gate, decode error or stale + # result must not create a retry loop that can collide with the final. + self._rolling_attempted = True self._rolling_inflight = True self._rolling_last_started_at = now self._rolling_last_speech_bytes = self._rolling_speech_bytes @@ -592,7 +611,11 @@ class StreamingTranscription: def worker() -> None: payload: dict | None = None try: - payload = _transcribe_pcm_rolling(self.model, pcm, self.language) + payload = _transcribe_pcm_rolling( + self.rolling_model, + pcm, + self.language, + ) except Exception as exc: # Partial inference is best-effort. The canonical commit path # remains authoritative and reports its own actionable errors. @@ -670,7 +693,7 @@ class StreamingTranscription: self._epoch += 1 self._cached_key = "" self._cached_payload = None - self._reset_rolling_history_locked() + self._reset_rolling_history_locked(rearm=True) self._heard_speech = True self._at_eos = False self._silence_bytes = 0 @@ -687,7 +710,7 @@ class StreamingTranscription: ) if self._silence_bytes >= required and not self._at_eos: self._at_eos = True - self._reset_rolling_history_locked() + self._reset_rolling_history_locked(rearm=False) auto_speculate = True if auto_speculate: self.speculate() @@ -705,7 +728,7 @@ class StreamingTranscription: self._client_active = True self._cached_key = "" self._cached_payload = None - self._reset_rolling_history_locked() + self._reset_rolling_history_locked(rearm=True) def speculate(self) -> None: """Decode a stable snapshot once; only an unchanged turn may consume it.""" @@ -846,11 +869,19 @@ class SpeechHandler(BaseHTTPRequestHandler): { "ok": True, "model": MODEL_NAME, + "rolling_model": ROLLING_MODEL_NAME, "device": "cuda" if torch.cuda.is_available() else "cpu", "inference": INFERENCE_GATE.snapshot(), "startup": { - "warmed": STARTUP_WARMUP_MS is not None, + "warmed": ( + STARTUP_WARMUP_MS is not None + and STARTUP_ROLLING_WARMUP_MS is not None + ), "warmup_ms": round(STARTUP_WARMUP_MS or 0.0, 1), + "rolling_warmup_ms": round( + STARTUP_ROLLING_WARMUP_MS or 0.0, + 1, + ), }, "streaming": { "enabled": True, @@ -909,6 +940,7 @@ class SpeechHandler(BaseHTTPRequestHandler): session = StreamingTranscription( websocket, self.server.model, # type: ignore[attr-defined] + self.server.rolling_model, # type: ignore[attr-defined] ) close_code = 1000 close_reason = "" @@ -1018,16 +1050,32 @@ class SpeechHandler(BaseHTTPRequestHandler): def main() -> None: """Warm Whisper once, then serve concurrent clients through one GPU lock.""" - global STARTUP_WARMUP_MS + global STARTUP_ROLLING_WARMUP_MS, STARTUP_WARMUP_MS CACHE_DIR.mkdir(parents=True, exist_ok=True) if not torch.cuda.is_available(): raise RuntimeError("CUDA is required for the Atlas Whisper service") - print(f"[stt] loading Whisper {MODEL_NAME} into CUDA", flush=True) + print(f"[stt] loading Whisper {MODEL_NAME} for final CUDA", flush=True) model = whisper.load_model(MODEL_NAME, device="cuda", download_root=str(CACHE_DIR)) + print(f"[stt] loading Whisper {ROLLING_MODEL_NAME} for rolling CUDA", flush=True) + rolling_model = whisper.load_model( + ROLLING_MODEL_NAME, + device="cuda", + download_root=str(CACHE_DIR), + ) + STARTUP_ROLLING_WARMUP_MS = _warm_model(rolling_model) + print( + "[stt] rolling CUDA warm-up completed in " f"{STARTUP_ROLLING_WARMUP_MS:.1f}ms", + flush=True, + ) + # Warm the authoritative model last so the first real final never inherits + # lazy CUDA work from either model's startup path. STARTUP_WARMUP_MS = _warm_model(model) - print(f"[stt] CUDA warm-up completed in {STARTUP_WARMUP_MS:.1f}ms", flush=True) + print( + f"[stt] final CUDA warm-up completed in {STARTUP_WARMUP_MS:.1f}ms", flush=True + ) server = ThreadingHTTPServer((HOST, PORT), SpeechHandler) server.model = model # type: ignore[attr-defined] + server.rolling_model = rolling_model # type: ignore[attr-defined] print(f"[stt] ready on {HOST}:{PORT}", flush=True) server.serve_forever(poll_interval=0.25) diff --git a/dockerfiles/hermes-webui-atlas-voice.js b/dockerfiles/hermes-webui-atlas-voice.js index d93a6224..f0e0b06e 100644 --- a/dockerfiles/hermes-webui-atlas-voice.js +++ b/dockerfiles/hermes-webui-atlas-voice.js @@ -638,12 +638,16 @@ if(Number.isFinite(revision)&&revision>partialRevision){ partialRevision=revision; const stable=String(payload.stable_transcript||'').trim(); - if(stable&&active&&captureTurnId===turnId&&state==='listening'){ - const preview=stable.length>72?stable.slice(0,69)+'…':stable; - label.textContent='Listening · '+preview; + const provisional=String(payload.transcript||'').trim(); + const visible=stable||provisional; + if(visible&&active&&captureTurnId===turnId&&state==='listening'){ + const preview=visible.length>72?visible.slice(0,69)+'…':visible; + label.textContent='Listening · '+preview+(stable?'':' · provisional'); if(stable!==lastPreflightText){ - lastPreflightText=stable; - scheduleVoicePreflight(turnId,revision,stable); + if(stable){ + lastPreflightText=stable; + scheduleVoicePreflight(turnId,revision,stable); + } } } } diff --git a/testing/tests/test_hermes_stt_rolling_model.py b/testing/tests/test_hermes_stt_rolling_model.py new file mode 100644 index 00000000..e7a5be8d --- /dev/null +++ b/testing/tests/test_hermes_stt_rolling_model.py @@ -0,0 +1,198 @@ +"""Release gates for Hermes' disposable tiny rolling transcription model.""" + +from __future__ import annotations + +import importlib.util +import struct +import sys +import threading +import time +from pathlib import Path +from types import SimpleNamespace + + +ROOT = Path(__file__).resolve().parents[2] +SERVER = ROOT / "dockerfiles" / "hermes-jetson-stt-server.py" +DOCKERFILE = ROOT / "dockerfiles" / "Dockerfile.hermes-jetson-stt" +PIPELINE = ROOT / "ci" / "Jenkinsfile.hermes-voice-image" + + +def _load_server(monkeypatch): + """Import the server without requiring image-only CUDA dependencies.""" + spec = importlib.util.spec_from_file_location("hermes_stt_dual_model", SERVER) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, "cgi", SimpleNamespace()) + monkeypatch.setitem( + sys.modules, + "torch", + SimpleNamespace( + cuda=SimpleNamespace(is_available=lambda: False, synchronize=lambda: None) + ), + ) + monkeypatch.setitem(sys.modules, "whisper", SimpleNamespace()) + spec.loader.exec_module(module) + return module + + +class _Connection: + """Capture asynchronous stream responses.""" + + def __init__(self) -> None: + self.messages: list[dict] = [] + self.condition = threading.Condition() + + def send_json(self, payload: dict) -> None: + with self.condition: + self.messages.append(payload) + self.condition.notify_all() + + def wait_for(self, message_type: str, timeout: float = 1.0) -> dict: + deadline = time.monotonic() + timeout + with self.condition: + while time.monotonic() < deadline: + for message in self.messages: + if message.get("type") == message_type: + return message + self.condition.wait(deadline - time.monotonic()) + raise AssertionError(f"missing {message_type}: {self.messages}") + + +class _Model: + """Return deterministic Whisper output while recording model ownership.""" + + def __init__(self, text: str) -> None: + self.text = text + self.calls: list[dict] = [] + + def transcribe(self, path: str, **options: object) -> dict: + self.calls.append({"path": path, "options": options}) + return { + "text": self.text, + "language": "en", + "segments": [ + { + "text": self.text, + "no_speech_prob": 0.01, + "avg_logprob": -0.1, + } + ], + } + + +def _start(session) -> None: + session.start( + { + "type": "start", + "turn_id": "dual-model-turn", + "format": "pcm_s16le", + "sample_rate": 16_000, + "language": "auto", + } + ) + + +def _speech(samples: int = 1_600) -> bytes: + return struct.pack(f"<{samples}h", *([12_000] * samples)) + + +def _wait_idle(session) -> None: + deadline = time.monotonic() + 1.0 + with session._lock: + while session._rolling_inflight and time.monotonic() < deadline: + session._lock.wait(deadline - time.monotonic()) + assert not session._rolling_inflight + + +def test_rolling_uses_tiny_once_while_final_remains_small(monkeypatch): + module = _load_server(monkeypatch) + monkeypatch.setattr(module, "ROLLING_MIN_AUDIO_MS", 100) + monkeypatch.setattr(module, "ROLLING_MAX_RESULT_LAG_MS", 5_000) + final_model = _Model("authoritative small") + rolling_model = _Model("provisional tiny") + connection = _Connection() + session = module.StreamingTranscription(connection, final_model, rolling_model) + _start(session) + + session.append(_speech()) + partial = connection.wait_for("partial") + assert partial["transcript"] == "provisional tiny" + assert partial["stable_transcript"] == "" + assert len(rolling_model.calls) == 1 + assert final_model.calls == [] + + session.append(_speech()) + _wait_idle(session) + assert len(rolling_model.calls) == 1 + session.commit() + assert connection.wait_for("final")["transcript"] == "authoritative small" + assert len(final_model.calls) == 1 + + +def test_eos_does_not_rearm_but_resumed_speech_does(monkeypatch): + module = _load_server(monkeypatch) + monkeypatch.setattr(module, "ROLLING_MIN_AUDIO_MS", 100) + monkeypatch.setattr(module, "ROLLING_MAX_RESULT_LAG_MS", 5_000) + final_model = _Model("final") + rolling_model = _Model("rolling") + session = module.StreamingTranscription(_Connection(), final_model, rolling_model) + _start(session) + + session.append(_speech()) + _wait_idle(session) + assert session._rolling_attempted is True + session.append(bytes(24_000)) + assert session._at_eos is True + assert session._rolling_attempted is True + + session.resume() + assert session._rolling_attempted is False + session.append(_speech()) + _wait_idle(session) + assert len(rolling_model.calls) == 2 + + +def test_dual_warmup_and_separate_timing_are_declared(monkeypatch): + module = _load_server(monkeypatch) + model = _Model("warm") + synchronized: list[bool] = [] + monkeypatch.setattr( + module.torch.cuda, "synchronize", lambda: synchronized.append(True) + ) + + elapsed = module._warm_model(model) + + assert elapsed >= 0 + assert module.ROLLING_MIN_AUDIO_MS == 1_800 + assert len(model.calls) == 1 + assert model.calls[0]["options"]["language"] is None + assert synchronized == [True] + assert not Path(model.calls[0]["path"]).exists() + source = SERVER.read_text() + rolling_warm = "STARTUP_ROLLING_WARMUP_MS = _warm_model(rolling_model)" + final_warm = "STARTUP_WARMUP_MS = _warm_model(model)" + assert source.index(rolling_warm) < source.index(final_warm) + assert source.index(final_warm) < source.index("server = ThreadingHTTPServer") + assert 'kind="rolling"' in source + assert 'kind="speculative"' in source + assert '"last_rolling_decode_ms"' in source + assert '"last_speculative_decode_ms"' in source + + +def test_tiny_model_is_checksum_pinned_read_only_in_the_runtime_image(): + dockerfile = DOCKERFILE.read_text() + digest = "65147644a518d12f04e32d6f3b26facc3f8dd46e5390956a9424a650c0ce22b9" + assert f"--checksum=sha256:{digest} --chmod=0444" in dockerfile + assert f"/models/{digest}/tiny.pt" in dockerfile + assert "HERMES_STT_ROLLING_MODEL=tiny" in dockerfile + assert "('large-v3-turbo.pt','small.pt','tiny.pt')" in dockerfile + + +def test_dual_model_contract_is_an_exact_voice_release_gate(): + pipeline = PIPELINE.read_text() + assert "testing/tests/test_hermes_stt_rolling_model.py" in pipeline + source = SERVER.read_text() + assert '"path": "/v1/audio/transcriptions/stream"' in source + assert '"inference": INFERENCE_GATE.snapshot()' in source + assert '"rolling_model": ROLLING_MODEL_NAME' in source + assert "STARTUP_ROLLING_WARMUP_MS is not None" in source diff --git a/testing/tests/test_hermes_stt_streaming.py b/testing/tests/test_hermes_stt_streaming.py index e5a3fbcf..355a2bd1 100644 --- a/testing/tests/test_hermes_stt_streaming.py +++ b/testing/tests/test_hermes_stt_streaming.py @@ -239,7 +239,7 @@ def test_stream_speculation_is_reused_for_unchanged_commit(monkeypatch): assert model.calls[0][1]["condition_on_previous_text"] is False -def test_active_speech_emits_rate_limited_stable_rolling_partials(monkeypatch): +def test_active_speech_emits_only_one_disposable_rolling_partial(monkeypatch): module = _load_server(monkeypatch) connection = _Connection() model = _SequencedModel(["please open", "please open my calendar"]) @@ -261,13 +261,12 @@ def test_active_speech_emits_rate_limited_stable_rolling_partials(monkeypatch): "window_start_ms": 0, } - # Audio progress is required in addition to wall-clock rate limiting. + # More PCM in the same epoch must not launch disposable GPU work again. session._rolling_last_started_at = 0.0 session.append(_speech(samples=12_000)) - second = connection.wait_for_revision(2) - assert second["transcript"] == "please open my calendar" - assert second["stable_transcript"] == "please open" - assert len(model.calls) == 2 + time.sleep(0.02) + assert len(model.calls) == 1 + assert not any(message.get("revision") == 2 for message in connection.messages) def test_rolling_window_is_bounded_but_final_decode_uses_full_utterance(monkeypatch): @@ -338,7 +337,8 @@ def test_busy_gpu_skips_partial_instead_of_queuing_work(monkeypatch): def occupy_gpu(): module.INFERENCE_GATE.try_background( - lambda: (entered.set(), release.wait(2.0)) + lambda: (entered.set(), release.wait(2.0)), + kind="rolling", ) worker = threading.Thread(target=occupy_gpu) @@ -353,12 +353,12 @@ def test_busy_gpu_skips_partial_instead_of_queuing_work(monkeypatch): release.set() worker.join(timeout=1.0) - # A later audio interval gets another opportunity once the final-priority - # GPU lock is free; the skipped snapshot did not leave a queued worker. + # A skipped epoch is never retried, so it cannot race the final later. session._rolling_last_started_at = 0.0 session.append(_speech()) - assert connection.wait_for_revision(1)["transcript"] == "hello" - assert len(model.calls) == 1 + time.sleep(0.02) + assert connection.messages == [] + assert model.calls == [] def test_waiting_final_prevents_new_rolling_work_from_stealing_gpu(monkeypatch): @@ -369,7 +369,8 @@ def test_waiting_final_prevents_new_rolling_work_from_stealing_gpu(monkeypatch): first = threading.Thread( target=lambda: module.INFERENCE_GATE.try_background( - lambda: (first_entered.set(), release_first.wait(2.0)) + lambda: (first_entered.set(), release_first.wait(2.0)), + kind="rolling", ) ) first.start() @@ -382,7 +383,7 @@ def test_waiting_final_prevents_new_rolling_work_from_stealing_gpu(monkeypatch): while module.INFERENCE_GATE._final_waiters < 1 and time.monotonic() < deadline: time.sleep(0.005) assert module.INFERENCE_GATE._final_waiters == 1 - assert module.INFERENCE_GATE.try_background(lambda: "ran") is None + assert module.INFERENCE_GATE.try_background(lambda: "ran", kind="rolling") is None release_first.set() first.join(timeout=1.0) final.join(timeout=1.0) @@ -464,36 +465,3 @@ def test_stream_contract_rejects_wrong_format_and_oversize(monkeypatch): monkeypatch.setattr(module, "MAX_STREAM_AUDIO_BYTES", 4) with pytest.raises(module.WebSocketError, match="too long"): session.append(bytes(6)) - - -def test_stream_health_contract_is_declared_in_source(): - source = SERVER.read_text() - - assert '"path": "/v1/audio/transcriptions/stream"' in source - assert '"format": "pcm_s16le"' in source - assert '"server_vad": True' in source - assert '"speculative": True' in source - assert '"inference": INFERENCE_GATE.snapshot()' in source - assert '"minimum_audio_ms": ROLLING_MIN_AUDIO_MS' in source - assert '"warmed": STARTUP_WARMUP_MS is not None' in source - assert "MAX_STREAM_AUDIO_BYTES" in source - - -def test_startup_warms_canonical_decode_before_serving(monkeypatch): - module = _load_server(monkeypatch) - model = _Model() - synchronized: list[bool] = [] - monkeypatch.setattr(module.torch.cuda, "synchronize", lambda: synchronized.append(True)) - - elapsed = module._warm_model(model) - - assert elapsed >= 0 - assert module.ROLLING_MIN_AUDIO_MS == 1_800 - assert len(model.calls) == 1 - assert model.calls[0][1]["language"] is None - assert synchronized == [True] - assert not Path(model.calls[0][0]).exists() - source = SERVER.read_text() - assert source.index("STARTUP_WARMUP_MS = _warm_model(model)") < source.index( - "server = ThreadingHTTPServer" - ) diff --git a/testing/tests/test_hermes_voice_full_duplex.py b/testing/tests/test_hermes_voice_full_duplex.py index 2f70124e..862d73d3 100644 --- a/testing/tests/test_hermes_voice_full_duplex.py +++ b/testing/tests/test_hermes_voice_full_duplex.py @@ -42,7 +42,10 @@ def test_barge_in_cancels_owned_turn_and_reuses_pcm_lookback(): assert "reusedCapture.lookback.forEach" in source assert "const bytes=new Uint8Array(16)" in source assert "window.crypto.getRandomValues(bytes)" in source - assert "captureTurnId=(voiceTabNonce?voiceTabNonce+'-':'')+String(token)+'-'+String(++turnSequence)" in source + assert ( + "captureTurnId=(voiceTabNonce?voiceTabNonce+'-':'')+String(token)+'-'+String(++turnSequence)" + in source + ) def test_streamed_chunks_share_one_audio_timeline_until_final_drain(): @@ -71,7 +74,10 @@ def test_barge_cancel_settles_before_new_turn_baseline_and_observer(): ) assert region.index("rememberAssistantBaseline()") < region.index("window.send()") assert region.index("window.send()") < region.index("startResponseObserver(token)") - assert "suppressAutoRead||bargeCancelPromise||state==='listening'||state==='transcribing'" in source + assert ( + "suppressAutoRead||bargeCancelPromise||state==='listening'||state==='transcribing'" + in source + ) def test_barge_handoff_never_uses_a_truncated_container_fallback(): @@ -96,7 +102,14 @@ def test_rolling_partials_are_feedback_only_and_wav_fallback_is_owned(): )[0] assert "payload.stable_transcript" in partial - assert "label.textContent='Listening · '+preview" in partial + assert "const provisional=String(payload.transcript||'').trim()" in partial + assert "const visible=stable||provisional" in partial + assert ( + "label.textContent='Listening · '+preview+(stable?'':' · provisional')" + in partial + ) + assert "if(stable){" in partial + assert "scheduleVoicePreflight(turnId,revision,stable)" in partial assert "composer.value" not in partial assert "window.send" not in partial assert "signal:controller.signal" in fallback @@ -113,7 +126,10 @@ def test_stt_client_queue_and_canonical_archive_are_strictly_bounded(): assert "queuedBytes+bytes.byteLength>STT_MAX_QUEUED_BYTES" in source assert "Streaming transcription backpressure limit exceeded" in source assert "if(flushTimer||settled||!queue.length" in source - assert "flushTimer=window.setTimeout(function(){flushTimer=null;flush();},20)" in source + assert ( + "flushTimer=window.setTimeout(function(){flushTimer=null;flush();},20)" + in source + ) assert "function clearQueue(){queue=[];queuedBytes=0;clearFlushTimer();}" in source assert "takeFallbackBlob:function()" in source assert "pcm16WavBlob(archive,16000)" in source