feat(hermes): ship full-duplex voice release

This commit is contained in:
jenkins 2026-08-23 22:13:52 -03:00
parent d3cbeb06c3
commit d254931a14
36 changed files with 3603 additions and 140 deletions

View File

@ -114,7 +114,7 @@ spec:
string( string(
name: 'EXPECTED_SOURCE_REVISION', name: 'EXPECTED_SOURCE_REVISION',
defaultValue: '', defaultValue: '',
description: 'Full reviewed commit that must be contained by atlas/titan-iac main.' description: 'Full reviewed commit already contained by atlas/titan-iac main.'
) )
string( string(
name: 'CONFIRM_PUBLISH', name: 'CONFIRM_PUBLISH',
@ -152,9 +152,11 @@ spec:
;; ;;
esac esac
test "${#EXPECTED_SOURCE_REVISION}" -eq 40 test "${#EXPECTED_SOURCE_REVISION}" -eq 40
main_revision="$(git rev-parse origin/main)"
git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}" "${main_revision}"
git checkout --detach "${EXPECTED_SOURCE_REVISION}"
actual_revision="$(git rev-parse HEAD)" actual_revision="$(git rev-parse HEAD)"
test "${actual_revision}" = "$(git rev-parse origin/main)" test "${actual_revision}" = "${EXPECTED_SOURCE_REVISION}"
git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}" "${actual_revision}"
test -z "$(git status --porcelain)" test -z "$(git status --porcelain)"
test -f dockerfiles/Dockerfile.hermes-agent test -f dockerfiles/Dockerfile.hermes-agent
case "${BUILD_NUMBER}" in case "${BUILD_NUMBER}" in

View File

@ -126,9 +126,12 @@ spec:
*[!0-9a-f]*|'') echo 'EXPECTED_SOURCE_REVISION must be a lowercase full commit' >&2; exit 2 ;; *[!0-9a-f]*|'') echo 'EXPECTED_SOURCE_REVISION must be a lowercase full commit' >&2; exit 2 ;;
esac esac
test "${#EXPECTED_SOURCE_REVISION}" -eq 40 test "${#EXPECTED_SOURCE_REVISION}" -eq 40
main_revision="$(git rev-parse HEAD)"
test "${main_revision}" = "$(git rev-parse origin/main)"
git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}" "${main_revision}"
git checkout --detach "${EXPECTED_SOURCE_REVISION}"
actual_revision="$(git rev-parse HEAD)" actual_revision="$(git rev-parse HEAD)"
test "${actual_revision}" = "$(git rev-parse origin/main)" test "${actual_revision}" = "${EXPECTED_SOURCE_REVISION}"
git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}" "${actual_revision}"
test -z "$(git status --porcelain)" test -z "$(git status --porcelain)"
case "${BUILD_NUMBER}" in ''|0*|*[!0-9]*) exit 2 ;; esac case "${BUILD_NUMBER}" in ''|0*|*[!0-9]*) exit 2 ;; esac
image="registry.bstein.dev/bstein/hermes-jetson-${IMAGE_COMPONENT}" image="registry.bstein.dev/bstein/hermes-jetson-${IMAGE_COMPONENT}"
@ -147,7 +150,12 @@ spec:
set -eu set -eu
python3 -m pip install --disable-pip-version-check --no-cache-dir \ python3 -m pip install --disable-pip-version-check --no-cache-dir \
--target=/tmp/hermes-voice-test-deps pytest==8.3.4 PyYAML==6.0.2 --target=/tmp/hermes-voice-test-deps pytest==8.3.4 PyYAML==6.0.2
python3 -m py_compile \
dockerfiles/hermes-jetson-stt-server.py \
dockerfiles/hermes-jetson-tts-server.py \
dockerfiles/hermes_jetson_tts_cues.py
PYTHONPATH=/tmp/hermes-voice-test-deps python3 -m pytest -q \ PYTHONPATH=/tmp/hermes-voice-test-deps python3 -m pytest -q \
testing/tests/test_hermes_stt_streaming.py \
testing/tests/test_hermes_tts_language_routing.py \ testing/tests/test_hermes_tts_language_routing.py \
testing/tests/test_hermes_voice_language_routing.py \ testing/tests/test_hermes_voice_language_routing.py \
testing/tests/test_hermes_oci_promote.py \ testing/tests/test_hermes_oci_promote.py \

View File

@ -143,9 +143,12 @@ spec:
;; ;;
esac esac
test "${#EXPECTED_SOURCE_REVISION}" -eq 40 test "${#EXPECTED_SOURCE_REVISION}" -eq 40
main_revision="$(git rev-parse HEAD)"
test "${main_revision}" = "$(git rev-parse origin/main)"
git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}" "${main_revision}"
git checkout --detach "${EXPECTED_SOURCE_REVISION}"
actual_revision="$(git rev-parse HEAD)" actual_revision="$(git rev-parse HEAD)"
test "${actual_revision}" = "$(git rev-parse origin/main)" test "${actual_revision}" = "${EXPECTED_SOURCE_REVISION}"
git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}" "${actual_revision}"
test -z "$(git status --porcelain)" test -z "$(git status --porcelain)"
test -f dockerfiles/Dockerfile.hermes-webui test -f dockerfiles/Dockerfile.hermes-webui
case "${BUILD_NUMBER}" in case "${BUILD_NUMBER}" in
@ -172,6 +175,14 @@ spec:
pytest==8.3.4 PyYAML==6.0.2 pytest==8.3.4 PyYAML==6.0.2
PYTHONPATH=/tmp/hermes-webui-release-test-deps \ PYTHONPATH=/tmp/hermes-webui-release-test-deps \
python3 -m pytest -q \ python3 -m pytest -q \
testing/tests/test_hermes_chat_quality.py \
testing/tests/test_hermes_handsfree_stt.py \
testing/tests/test_hermes_voice_instrument.py \
testing/tests/test_hermes_voice_full_duplex.py \
testing/tests/test_hermes_voice_route_preflight.py \
testing/tests/test_hermes_voice_preflight_delivery.py \
testing/tests/test_hermes_thinking_voice_cues.py \
testing/tests/test_hermes_voice_language_routing.py \
testing/tests/test_hermes_webui_brand.py \ testing/tests/test_hermes_webui_brand.py \
testing/tests/test_hermes_webui_release.py \ testing/tests/test_hermes_webui_release.py \
testing/tests/test_hermes_oci_promote.py \ testing/tests/test_hermes_oci_promote.py \

View File

@ -49,7 +49,8 @@ ADD --checksum=sha256:1afc81f703c0e4cb3b4d7c0dca096b8b54a98806807f0170cf5eb55577
RUN chmod 0555 /opt/models /opt/models/piper RUN chmod 0555 /opt/models /opt/models/piper
COPY dockerfiles/hermes-jetson-tts-server.py /opt/atlas/hermes-jetson-tts-server.py COPY dockerfiles/hermes-jetson-tts-server.py /opt/atlas/hermes-jetson-tts-server.py
RUN chmod 0555 /opt/atlas/hermes-jetson-tts-server.py COPY dockerfiles/hermes_jetson_tts_cues.py /opt/atlas/hermes_jetson_tts_cues.py
RUN chmod 0555 /opt/atlas/hermes-jetson-tts-server.py /opt/atlas/hermes_jetson_tts_cues.py
# Load every pinned voice during the ARM64 build, including the three baked # Load every pinned voice during the ARM64 build, including the three baked
# for the multilingual chat policy. This catches package or model-format # for the multilingual chat policy. This catches package or model-format

View File

@ -53,6 +53,8 @@ RUN /opt/hermes/.venv/bin/python -c 'import cryptography, yaml' \
&& grep -Fq "'atlas/auto/maximum': 'Automatic · Maximum'" /opt/hermes-webui/static/panels.js \ && grep -Fq "'atlas/auto/maximum': 'Automatic · Maximum'" /opt/hermes-webui/static/panels.js \
&& grep -Fq 'Atlas Jetson (private)' /opt/hermes-webui/static/index.html \ && grep -Fq 'Atlas Jetson (private)' /opt/hermes-webui/static/index.html \
&& grep -Fq 'HERMES_WEBUI_ATLAS_TTS_URL' /opt/hermes-webui/api/routes.py \ && grep -Fq 'HERMES_WEBUI_ATLAS_TTS_URL' /opt/hermes-webui/api/routes.py \
&& grep -Fq 'settings["webui_bundle_version"]' /opt/hermes-webui/api/routes.py \
&& grep -Fq 'settings.webui_bundle_version||settings.webui_version' /opt/hermes-webui/static/panels.js \
&& grep -Fq 'Audio conversion failed: upload is invalid' /opt/hermes/tools/transcription_tools.py \ && grep -Fq 'Audio conversion failed: upload is invalid' /opt/hermes/tools/transcription_tools.py \
&& grep -Fq "capability.provider!=='local_command'" /opt/hermes-webui/static/atlas-voice.js \ && grep -Fq "capability.provider!=='local_command'" /opt/hermes-webui/static/atlas-voice.js \
&& grep -Fq 'prefers-reduced-motion: reduce' /opt/hermes-webui/static/atlas-voice.css \ && grep -Fq 'prefers-reduced-motion: reduce' /opt/hermes-webui/static/atlas-voice.css \

View File

@ -39,10 +39,96 @@ MAX_WS_MESSAGE_BYTES = 256 * 1024
STREAM_IDLE_SECONDS = int(os.getenv("HERMES_STT_STREAM_IDLE_SECONDS", "120")) STREAM_IDLE_SECONDS = int(os.getenv("HERMES_STT_STREAM_IDLE_SECONDS", "120"))
VAD_END_SILENCE_MS = int(os.getenv("HERMES_STT_VAD_END_SILENCE_MS", "650")) VAD_END_SILENCE_MS = int(os.getenv("HERMES_STT_VAD_END_SILENCE_MS", "650"))
VAD_TAIL_MS = int(os.getenv("HERMES_STT_VAD_TAIL_MS", "220")) VAD_TAIL_MS = int(os.getenv("HERMES_STT_VAD_TAIL_MS", "220"))
MODEL_LOCK = threading.Lock() ROLLING_MIN_AUDIO_MS = max(
300,
min(2_000, int(os.getenv("HERMES_STT_ROLLING_MIN_AUDIO_MS", "650"))),
)
ROLLING_INTERVAL_MS = max(
400,
min(5_000, int(os.getenv("HERMES_STT_ROLLING_INTERVAL_MS", "750"))),
)
ROLLING_WINDOW_MS = max(
3_000,
min(30_000, int(os.getenv("HERMES_STT_ROLLING_WINDOW_MS", "12000"))),
)
ROLLING_MAX_RESULT_LAG_MS = max(
500,
min(5_000, int(os.getenv("HERMES_STT_ROLLING_MAX_LAG_MS", "2000"))),
)
_WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" _WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
class InferenceGate:
"""Give authoritative transcripts priority over disposable partial work."""
def __init__(self) -> None:
self._condition = threading.Condition()
self._busy = False
self._final_waiters = 0
self._last_final_wait_ms = 0.0
self._last_final_decode_ms = 0.0
self._last_background_decode_ms = 0.0
self._background_skips = 0
def run_final(self, callback):
"""Wait fairly enough that newly arriving partials cannot steal the GPU."""
wait_started = time.monotonic()
with self._condition:
self._final_waiters += 1
try:
while self._busy:
self._condition.wait()
self._busy = True
finally:
self._final_waiters -= 1
self._last_final_wait_ms = (time.monotonic() - wait_started) * 1000
decode_started = time.monotonic()
try:
return callback()
finally:
with self._condition:
self._last_final_decode_ms = (
time.monotonic() - decode_started
) * 1000
self._busy = False
self._condition.notify_all()
def try_background(self, callback):
"""Run one useful predecode only when no final is active or waiting."""
with self._condition:
if self._busy or self._final_waiters:
self._background_skips += 1
return None
self._busy = True
decode_started = time.monotonic()
try:
return callback()
finally:
with self._condition:
self._last_background_decode_ms = (
time.monotonic() - decode_started
) * 1000
self._busy = False
self._condition.notify_all()
def snapshot(self) -> dict:
"""Expose bounded timing telemetry for live latency verification."""
with self._condition:
return {
"busy": self._busy,
"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
),
"background_skips": self._background_skips,
}
INFERENCE_GATE = InferenceGate()
def _repetitive_token(token: str) -> bool: def _repetitive_token(token: str) -> bool:
"""Identify long periodic Whisper hallucinations caused by steady noise.""" """Identify long periodic Whisper hallucinations caused by steady noise."""
letters = "".join(re.findall(r"[a-z]+", token.lower())) letters = "".join(re.findall(r"[a-z]+", token.lower()))
@ -112,37 +198,47 @@ def _transcription_payload(result: dict) -> dict:
} }
def _decode_path(model: object, path: str, language: str) -> dict:
"""Decode one path with the quality settings shared by partial and final work."""
result = model.transcribe(
path,
language=None if language in {"", "auto"} else language,
task="transcribe",
fp16=torch.cuda.is_available(),
condition_on_previous_text=False,
temperature=0,
compression_ratio_threshold=2.0,
logprob_threshold=-0.8,
no_speech_threshold=0.5,
verbose=False,
)
return _transcription_payload(result)
def _transcribe_path(model: object, path: str, language: str) -> dict: def _transcribe_path(model: object, path: str, language: str) -> dict:
"""Run the one canonical full-utterance Whisper decode configuration.""" """Run the one canonical full-utterance Whisper decode configuration."""
with MODEL_LOCK: return INFERENCE_GATE.run_final(lambda: _decode_path(model, path, language))
result = model.transcribe(
path,
language=None if language in {"", "auto"} else language, def _write_pcm_wav(pcm: bytes) -> str:
task="transcribe", """Write an aligned browser PCM snapshot to a temporary Whisper input."""
fp16=torch.cuda.is_available(), with tempfile.NamedTemporaryFile(
condition_on_previous_text=False, prefix="atlas-stt-stream-", suffix=".wav", delete=False
temperature=0, ) as temp:
compression_ratio_threshold=2.0, temp_path = temp.name
logprob_threshold=-0.8, with wave.open(temp_path, "wb") as wav_file:
no_speech_threshold=0.5, wav_file.setnchannels(STREAM_CHANNELS)
verbose=False, wav_file.setsampwidth(STREAM_SAMPLE_WIDTH)
) wav_file.setframerate(STREAM_SAMPLE_RATE)
return _transcription_payload(result) wav_file.writeframes(pcm)
return temp_path
def _transcribe_pcm(model: object, pcm: bytes, language: str) -> dict: def _transcribe_pcm(model: object, pcm: bytes, language: str) -> dict:
"""Decode one complete 16-kHz mono PCM snapshot through the canonical path.""" """Decode one complete 16-kHz mono PCM snapshot through the canonical path."""
temp_path = "" temp_path = ""
try: try:
with tempfile.NamedTemporaryFile( temp_path = _write_pcm_wav(pcm)
prefix="atlas-stt-stream-", suffix=".wav", delete=False
) as temp:
temp_path = temp.name
with wave.open(temp_path, "wb") as wav_file:
wav_file.setnchannels(STREAM_CHANNELS)
wav_file.setsampwidth(STREAM_SAMPLE_WIDTH)
wav_file.setframerate(STREAM_SAMPLE_RATE)
wav_file.writeframes(pcm)
return _transcribe_path(model, temp_path, language) return _transcribe_path(model, temp_path, language)
finally: finally:
if temp_path: if temp_path:
@ -152,6 +248,76 @@ def _transcribe_pcm(model: object, pcm: bytes, language: str) -> dict:
pass pass
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.
"""
temp_path = ""
try:
temp_path = _write_pcm_wav(pcm)
return INFERENCE_GATE.try_background(
lambda: _decode_path(model, temp_path, language)
)
finally:
if temp_path:
try:
os.unlink(temp_path)
except OSError:
pass
def _transcribe_pcm_speculative(
model: object, pcm: bytes, language: str
) -> dict | None:
"""Predecode one EOS snapshot only when the model is currently idle."""
temp_path = ""
try:
temp_path = _write_pcm_wav(pcm)
return INFERENCE_GATE.try_background(
lambda: _decode_path(model, temp_path, language)
)
finally:
if temp_path:
try:
os.unlink(temp_path)
except OSError:
pass
def _pcm_bytes_for_ms(milliseconds: int) -> int:
"""Convert milliseconds to an aligned byte count for the stream format."""
return (
STREAM_SAMPLE_RATE
* STREAM_SAMPLE_WIDTH
* milliseconds
// 1000
// STREAM_SAMPLE_WIDTH
* STREAM_SAMPLE_WIDTH
)
def _token_identity(token: str) -> str:
"""Normalize punctuation/case drift before comparing consecutive decodes."""
return re.sub(r"[^\w']+", "", token.casefold(), flags=re.UNICODE)
def _stable_token_prefix(previous: str, current: str) -> str:
"""Return only tokens repeated in the same order by consecutive decodes."""
old_tokens = previous.split()
new_tokens = current.split()
agreed = 0
for old, new in zip(old_tokens, new_tokens):
if not _token_identity(old) or _token_identity(old) != _token_identity(new):
break
agreed += 1
return " ".join(new_tokens[:agreed])
class WebSocketError(Exception): class WebSocketError(Exception):
"""Protocol error carrying an RFC 6455 close code safe to expose.""" """Protocol error carrying an RFC 6455 close code safe to expose."""
@ -300,6 +466,12 @@ class StreamingTranscription:
self._pending_key = "" self._pending_key = ""
self._cached_key = "" self._cached_key = ""
self._cached_payload: dict | None = None self._cached_payload: dict | None = None
self._rolling_inflight = False
self._rolling_last_started_at = 0.0
self._rolling_last_audio_bytes = 0
self._rolling_previous = ""
self._rolling_previous_window_start = 0
self._rolling_revision = 0
def _response(self, message_type: str, **values: object) -> dict: def _response(self, message_type: str, **values: object) -> dict:
return {"type": message_type, "turn_id": self.turn_id, **values} return {"type": message_type, "turn_id": self.turn_id, **values}
@ -345,13 +517,125 @@ class StreamingTranscription:
key = f"{self._epoch}:{hashlib.sha256(pcm).hexdigest()}" key = f"{self._epoch}:{hashlib.sha256(pcm).hexdigest()}"
return pcm, key return pcm, key
def _rolling_snapshot_locked(self) -> tuple[bytes, int]:
"""Return a bounded tail window and its absolute start in milliseconds."""
window_bytes = _pcm_bytes_for_ms(ROLLING_WINDOW_MS)
start = max(0, len(self._pcm) - window_bytes)
start -= start % STREAM_SAMPLE_WIDTH
pcm = bytes(self._pcm[start:])
start_ms = start * 1000 // (STREAM_SAMPLE_RATE * STREAM_SAMPLE_WIDTH)
return pcm, start_ms
def _reset_rolling_history_locked(self) -> None:
"""Discard provisional agreement after the client's speech epoch changes."""
self._rolling_previous = ""
self._rolling_previous_window_start = 0
# Audio before a resume belongs to an invalidated provisional epoch and
# must not immediately satisfy the next rolling interval by itself.
self._rolling_last_audio_bytes = len(self._pcm)
self._rolling_last_started_at = 0.0
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)
interval_bytes = _pcm_bytes_for_ms(ROLLING_INTERVAL_MS)
required_new_bytes = (
interval_bytes if self._rolling_last_audio_bytes else minimum_bytes
)
if (
self._rolling_inflight
or total_bytes < minimum_bytes
or total_bytes - self._rolling_last_audio_bytes < required_new_bytes
or (
self._rolling_last_started_at
and (now - self._rolling_last_started_at) * 1000
< ROLLING_INTERVAL_MS
)
):
return None
pcm, window_start_ms = self._rolling_snapshot_locked()
self._rolling_inflight = True
self._rolling_last_started_at = now
self._rolling_last_audio_bytes = total_bytes
return pcm, window_start_ms, total_bytes, self._epoch
def _start_rolling(
self,
pcm: bytes,
window_start_ms: int,
audio_bytes: int,
epoch: int,
) -> None:
"""Run one disposable partial decode and publish only a current result."""
def worker() -> None:
payload: dict | None = None
try:
payload = _transcribe_pcm_rolling(self.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.
print(f"[stt] rolling transcription skipped: {exc}", flush=True)
message: dict | None = None
with self._lock:
self._rolling_inflight = False
lag_bytes = max(0, len(self._pcm) - audio_bytes)
max_lag_bytes = _pcm_bytes_for_ms(ROLLING_MAX_RESULT_LAG_MS)
current = (
payload is not None
and not self._closed
and not self._committing
and not self._at_eos
and self._epoch == epoch
and lag_bytes <= max_lag_bytes
)
if current:
transcript = str(payload.get("text") or "")
stable = ""
if self._rolling_previous_window_start == window_start_ms:
stable = _stable_token_prefix(
self._rolling_previous,
transcript,
)
self._rolling_previous = transcript
self._rolling_previous_window_start = window_start_ms
self._rolling_revision += 1
message = self._response(
"partial",
transcript=transcript,
stable_transcript=stable,
language=payload.get("language") or "",
speculative=True,
rolling=True,
revision=self._rolling_revision,
epoch=epoch,
window_start_ms=window_start_ms,
)
self._lock.notify_all()
if message is not None:
try:
self.connection.send_json(message)
except (BrokenPipeError, ConnectionError, OSError):
self.cancel()
threading.Thread(
target=worker,
name=f"stt-rolling-{self.turn_id}",
daemon=True,
).start()
def append(self, pcm: bytes) -> None: def append(self, pcm: bytes) -> None:
"""Append PCM, track server VAD and begin speculation at probable EOS.""" """Append PCM while overlapping rolling understanding with active speech."""
if not self._started: if not self._started:
raise WebSocketError("start must precede audio", 1008) raise WebSocketError("start must precede audio", 1008)
if not pcm or len(pcm) % STREAM_SAMPLE_WIDTH: if not pcm or len(pcm) % STREAM_SAMPLE_WIDTH:
raise WebSocketError("unaligned PCM audio", 1003) raise WebSocketError("unaligned PCM audio", 1003)
auto_speculate = False auto_speculate = False
rolling: tuple[bytes, int, int, int] | None = None
with self._lock: with self._lock:
if self._closed or self._committing: if self._closed or self._committing:
raise WebSocketError("turn is no longer accepting audio", 1008) raise WebSocketError("turn is no longer accepting audio", 1008)
@ -368,10 +652,12 @@ class StreamingTranscription:
self._epoch += 1 self._epoch += 1
self._cached_key = "" self._cached_key = ""
self._cached_payload = None self._cached_payload = None
self._reset_rolling_history_locked()
self._heard_speech = True self._heard_speech = True
self._at_eos = False self._at_eos = False
self._silence_bytes = 0 self._silence_bytes = 0
self._last_speech_byte = len(self._pcm) self._last_speech_byte = len(self._pcm)
rolling = self._begin_rolling_locked(time.monotonic())
elif self._heard_speech: elif self._heard_speech:
self._silence_bytes += len(pcm) self._silence_bytes += len(pcm)
required = ( required = (
@ -385,6 +671,8 @@ class StreamingTranscription:
auto_speculate = True auto_speculate = True
if auto_speculate: if auto_speculate:
self.speculate() self.speculate()
elif rolling is not None:
self._start_rolling(*rolling)
def resume(self) -> None: def resume(self) -> None:
"""Invalidate an EOS snapshot when client-side VAD hears resumed speech.""" """Invalidate an EOS snapshot when client-side VAD hears resumed speech."""
@ -397,6 +685,7 @@ class StreamingTranscription:
self._client_active = True self._client_active = True
self._cached_key = "" self._cached_key = ""
self._cached_payload = None self._cached_payload = None
self._reset_rolling_history_locked()
def speculate(self) -> None: def speculate(self) -> None:
"""Decode a stable snapshot once; only an unchanged turn may consume it.""" """Decode a stable snapshot once; only an unchanged turn may consume it."""
@ -412,7 +701,7 @@ class StreamingTranscription:
def worker() -> None: def worker() -> None:
try: try:
payload = _transcribe_pcm(self.model, pcm, self.language) payload = _transcribe_pcm_speculative(self.model, pcm, self.language)
except Exception as exc: except Exception as exc:
print(f"[stt] speculative transcription failed: {exc}", flush=True) print(f"[stt] speculative transcription failed: {exc}", flush=True)
with self._lock: with self._lock:
@ -423,6 +712,13 @@ class StreamingTranscription:
self.error("speculative transcription failed") self.error("speculative transcription failed")
return return
if payload is None:
with self._lock:
if self._pending_key == key:
self._pending_key = ""
self._lock.notify_all()
return
should_send = False should_send = False
with self._lock: with self._lock:
if self._pending_key == key: if self._pending_key == key:
@ -531,6 +827,7 @@ class SpeechHandler(BaseHTTPRequestHandler):
"ok": True, "ok": True,
"model": MODEL_NAME, "model": MODEL_NAME,
"device": "cuda" if torch.cuda.is_available() else "cpu", "device": "cuda" if torch.cuda.is_available() else "cpu",
"inference": INFERENCE_GATE.snapshot(),
"streaming": { "streaming": {
"enabled": True, "enabled": True,
"path": "/v1/audio/transcriptions/stream", "path": "/v1/audio/transcriptions/stream",
@ -539,6 +836,12 @@ class SpeechHandler(BaseHTTPRequestHandler):
"max_seconds": MAX_STREAM_SECONDS, "max_seconds": MAX_STREAM_SECONDS,
"server_vad": True, "server_vad": True,
"speculative": True, "speculative": True,
"rolling": {
"enabled": True,
"interval_ms": ROLLING_INTERVAL_MS,
"window_ms": ROLLING_WINDOW_MS,
"stable_prefix": True,
},
}, },
}, },
) )

View File

@ -7,6 +7,7 @@ import io
import json import json
import os import os
import re import re
import struct
import threading import threading
import wave import wave
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@ -15,6 +16,8 @@ from pathlib import Path
import onnxruntime import onnxruntime
from piper import PiperConfig, PiperVoice, SynthesisConfig from piper import PiperConfig, PiperVoice, SynthesisConfig
from hermes_jetson_tts_cues import build_cue_cache, resolve_cached_cue, write_cached_cue
HOST = os.getenv("HERMES_TTS_HOST", "0.0.0.0") HOST = os.getenv("HERMES_TTS_HOST", "0.0.0.0")
PORT = int(os.getenv("HERMES_TTS_PORT", "9001")) PORT = int(os.getenv("HERMES_TTS_PORT", "9001"))
@ -29,6 +32,25 @@ STREAM_WRITE_TIMEOUT_SECONDS = max(
VOICE_LOCK = threading.Lock() VOICE_LOCK = threading.Lock()
TURN_ID_PATTERN = re.compile(r"[A-Za-z0-9._:-]{1,128}\Z") TURN_ID_PATTERN = re.compile(r"[A-Za-z0-9._:-]{1,128}\Z")
# Piper deliberately leaves a generous tail after sentence punctuation. That
# sounds natural when one synthesis result is played in isolation, but hands-
# free mode queues many independently synthesized clauses and compounds those
# tails with the browser's punctuation cadence. Only silence after the last
# audible 5 ms window is shortened; voiced samples are never faded or removed.
SILENCE_WINDOW_MS = 5
SILENCE_ABS_THRESHOLD = 32
SILENCE_TRIM_TOLERANCE_MS = 20
CLAUSE_PAUSE_MS = 90
COLON_PAUSE_MS = 110
QUESTION_PAUSE_MS = 170
PERIOD_PAUSE_MS = 190
ELLIPSIS_PAUSE_MS = 220
PARAGRAPH_PAUSE_MS = 280
UNPUNCTUATED_PAUSE_MS = 70
SENTENCE_BOUNDARY_PATTERN = re.compile(
r"(?P<punct>\.{1,3}|[!?…]+)(?:[\"'”’)}\]]+)?(?P<gap>\s+|$)"
)
# Fixed, allow-listed language -> baked voice mapping. This is the ONLY path # Fixed, allow-listed language -> baked voice mapping. This is the ONLY path
# from a client-supplied string to a model name: client input is looked up # from a client-supplied string to a model name: client input is looked up
# here and never used to build a filesystem path directly. Both "-" and "_" # here and never used to build a filesystem path directly. Both "-" and "_"
@ -79,6 +101,111 @@ def normalize_turn_id(value: object) -> str | None:
return normalized return normalized
def trailing_pause_ms(text: str) -> int:
"""Return the natural trailing pause budget for one spoken text unit."""
if re.search(r"\n\s*\n\s*$", text):
return PARAGRAPH_PAUSE_MS
terminal = text.rstrip().rstrip("\"'”’)}]")
if terminal.endswith(("...", "…")):
return ELLIPSIS_PAUSE_MS
if terminal.endswith("."):
return PERIOD_PAUSE_MS
if terminal.endswith(("?", "!")):
return QUESTION_PAUSE_MS
if terminal.endswith((":", ";")):
return COLON_PAUSE_MS
if terminal.endswith(","):
return CLAUSE_PAUSE_MS
return UNPUNCTUATED_PAUSE_MS
def sentence_pause_targets(text: str) -> list[int]:
"""Map Piper's likely sentence results to punctuation-aware pause budgets."""
targets: list[int] = []
cursor = 0
for match in SENTENCE_BOUNDARY_PATTERN.finditer(text):
unit = text[cursor : match.end()]
targets.append(trailing_pause_ms(unit))
cursor = match.end()
if text[cursor:].strip() or not targets:
targets.append(trailing_pause_ms(text[cursor:] or text))
return targets
def trim_trailing_pcm_silence(
pcm: bytes | bytearray | memoryview,
sample_rate: int,
pause_ms: int,
) -> memoryview:
"""Shorten only a confirmed PCM silence tail to the requested duration.
The backward scan is restricted to the trailing envelope. Returning a
memoryview avoids copying sentence audio on the latency-sensitive stream
path. All-silent input is retained because it carries no safe phoneme
boundary from which to measure a pause.
"""
raw = memoryview(pcm).cast("B")
if len(raw) < 2 or len(raw) % 2 or sample_rate <= 0:
return raw
window_samples = max(1, sample_rate * SILENCE_WINDOW_MS // 1000)
window_bytes = window_samples * 2
active_end: int | None = None
cursor = len(raw)
while cursor > 0:
start = max(0, cursor - window_bytes)
# Piper PCM is explicitly signed 16-bit little endian. A conservative
# threshold preserves quiet word endings while ignoring quantization
# noise in the generated tail.
if any(
abs(struct.unpack_from("<h", raw, offset)[0]) > SILENCE_ABS_THRESHOLD
for offset in range(start, cursor, 2)
):
active_end = cursor
break
cursor = start
if active_end is None:
return raw
desired_tail_bytes = max(0, sample_rate * pause_ms // 1000) * 2
cutoff = min(len(raw), active_end + desired_tail_bytes)
tolerance_bytes = max(0, sample_rate * SILENCE_TRIM_TOLERANCE_MS // 1000) * 2
if len(raw) - cutoff <= tolerance_bytes:
return raw
return raw[:cutoff]
def normalize_wav_trailing_pause(audio: bytes, text: str) -> bytes:
"""Apply the same safe silence-tail policy to the compatibility WAV path."""
source = io.BytesIO(audio)
try:
with wave.open(source, "rb") as reader:
params = reader.getparams()
if (
params.nchannels != 1
or params.sampwidth != 2
or params.comptype != "NONE"
):
return audio
pcm = reader.readframes(params.nframes)
except (EOFError, wave.Error):
return audio
normalized = trim_trailing_pcm_silence(
pcm,
params.framerate,
trailing_pause_ms(text),
)
if len(normalized) == len(pcm):
return audio
output = io.BytesIO()
with wave.open(output, "wb") as writer:
writer.setparams(params)
writer.writeframes(normalized)
return output.getvalue()
def _json(handler: BaseHTTPRequestHandler, status: int, payload: dict) -> None: def _json(handler: BaseHTTPRequestHandler, status: int, payload: dict) -> None:
body = json.dumps(payload).encode("utf-8") body = json.dumps(payload).encode("utf-8")
handler.send_response(status) handler.send_response(status)
@ -114,6 +241,7 @@ class SpeechHandler(BaseHTTPRequestHandler):
"path": "/v1/audio/speech/stream", "path": "/v1/audio/speech/stream",
"format": "pcm_s16le", "format": "pcm_s16le",
}, },
"thinking_cues": len(self.server.cue_cache), # type: ignore[attr-defined]
}, },
) )
@ -146,6 +274,21 @@ class SpeechHandler(BaseHTTPRequestHandler):
return return
speed = min(2.0, max(0.5, speed)) speed = min(2.0, max(0.5, speed))
try:
cue = resolve_cached_cue(
payload,
getattr(self.server, "cue_cache", {}),
)
except ValueError as exc:
_json(self, 400, {"error": str(exc)})
return
if cue is not None:
if self.path != "/v1/audio/speech/stream":
_json(self, 400, {"error": "thinking cues require PCM streaming"})
return
write_cached_cue(self, cue, normalize_turn_id(payload.get("turn_id")))
return
# Policy is driven ONLY by "language". A client-supplied "voice" # Policy is driven ONLY by "language". A client-supplied "voice"
# field is deliberately never read here; it cannot override the # field is deliberately never read here; it cannot override the
# allow-listed mapping. # allow-listed mapping.
@ -175,7 +318,7 @@ class SpeechHandler(BaseHTTPRequestHandler):
wav_file, wav_file,
SynthesisConfig(length_scale=1.0 / speed), SynthesisConfig(length_scale=1.0 / speed),
) )
audio = output.getvalue() audio = normalize_wav_trailing_pause(output.getvalue(), text)
self.send_response(200) self.send_response(200)
self.send_header("Content-Type", "audio/wav") self.send_header("Content-Type", "audio/wav")
self.send_header("Content-Length", str(len(audio))) self.send_header("Content-Length", str(len(audio)))
@ -241,12 +384,14 @@ class SpeechHandler(BaseHTTPRequestHandler):
connected = True connected = True
try: try:
pause_targets = sentence_pause_targets(text)
audio_chunks = iter( audio_chunks = iter(
voice.synthesize( voice.synthesize(
text, text,
SynthesisConfig(length_scale=1.0 / speed), SynthesisConfig(length_scale=1.0 / speed),
) )
) )
chunk_index = 0
while connected: while connected:
# Serialize Piper/ONNX access, but never hold the model lock # Serialize Piper/ONNX access, but never hold the model lock
# while a slow browser applies network back-pressure. # while a slow browser applies network back-pressure.
@ -261,7 +406,13 @@ class SpeechHandler(BaseHTTPRequestHandler):
or audio_chunk.sample_channels != 1 or audio_chunk.sample_channels != 1
): ):
raise RuntimeError("Piper returned an unexpected PCM format") raise RuntimeError("Piper returned an unexpected PCM format")
pcm = memoryview(audio_chunk.audio_int16_bytes) pause_ms = pause_targets[min(chunk_index, len(pause_targets) - 1)]
pcm = trim_trailing_pcm_silence(
audio_chunk.audio_int16_bytes,
sample_rate,
pause_ms,
)
chunk_index += 1
for offset in range(0, len(pcm), STREAM_WRITE_BYTES): for offset in range(0, len(pcm), STREAM_WRITE_BYTES):
if not self._write_stream_chunk(pcm[offset : offset + STREAM_WRITE_BYTES]): if not self._write_stream_chunk(pcm[offset : offset + STREAM_WRITE_BYTES]):
connected = False connected = False
@ -322,6 +473,12 @@ def main() -> None:
f"HERMES_TTS_VOICE must name one of the baked policy voices: {sorted(BAKED_VOICE_NAMES)}" f"HERMES_TTS_VOICE must name one of the baked policy voices: {sorted(BAKED_VOICE_NAMES)}"
) )
voices = load_voices(CACHE_DIR, ONNX_THREADS) voices = load_voices(CACHE_DIR, ONNX_THREADS)
cue_cache = build_cue_cache(
voices,
SynthesisConfig,
trim_trailing_pcm_silence,
trailing_pause_ms,
)
print( print(
f"[tts] loaded {len(voices)} Piper voices on CPU with {ONNX_THREADS} ONNX threads each: " f"[tts] loaded {len(voices)} Piper voices on CPU with {ONNX_THREADS} ONNX threads each: "
+ ", ".join(sorted(voices)), + ", ".join(sorted(voices)),
@ -330,6 +487,7 @@ def main() -> None:
server = ThreadingHTTPServer((HOST, PORT), SpeechHandler) server = ThreadingHTTPServer((HOST, PORT), SpeechHandler)
server.voices = voices # type: ignore[attr-defined] server.voices = voices # type: ignore[attr-defined]
server.default_voice_name = DEFAULT_VOICE_NAME # type: ignore[attr-defined] server.default_voice_name = DEFAULT_VOICE_NAME # type: ignore[attr-defined]
server.cue_cache = cue_cache # type: ignore[attr-defined]
print(f"[tts] ready on {HOST}:{PORT}", flush=True) print(f"[tts] ready on {HOST}:{PORT}", flush=True)
server.serve_forever(poll_interval=0.25) server.serve_forever(poll_interval=0.25)

View File

@ -313,6 +313,7 @@ replace_exact(
routes, routes,
"def _tts_open(req, *, timeout=30, opener_factory=None):", "def _tts_open(req, *, timeout=30, opener_factory=None):",
'''ATLAS_TTS_LANGUAGES = ("en", "ru", "es") '''ATLAS_TTS_LANGUAGES = ("en", "ru", "es")
ATLAS_TTS_CUE_IDS = ("thinking", "let_me_think", "still_working", "one_more_moment")
def _atlas_tts_language(body): def _atlas_tts_language(body):
@ -343,9 +344,12 @@ replace_exact(
"def _tts_open(req, *, timeout=30, opener_factory=None):", "def _tts_open(req, *, timeout=30, opener_factory=None):",
'''ATLAS_TTS_STREAM_URL = "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech/stream" '''ATLAS_TTS_STREAM_URL = "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech/stream"
ATLAS_STT_STREAM_URL = "http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions/stream" ATLAS_STT_STREAM_URL = "http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions/stream"
ATLAS_VOICE_PREFLIGHT_URL = "http://hermes-switchyard.hermes.svc.cluster.local:9009/voice/route-preflight"
ATLAS_VOICE_WS_PROTOCOL = "hermes-voice-v1" ATLAS_VOICE_WS_PROTOCOL = "hermes-voice-v1"
ATLAS_VOICE_MAX_BYTES = 8 * 1024 * 1024 ATLAS_VOICE_MAX_BYTES = 8 * 1024 * 1024
ATLAS_VOICE_DEADLINE_SECONDS = 100 ATLAS_VOICE_DEADLINE_SECONDS = 100
ATLAS_VOICE_PREFLIGHT_TIERS = ("fast", "balanced", "deep", "maximum")
ATLAS_VOICE_PREFLIGHT_MAX_RESPONSE_BYTES = 1024
def _atlas_exact_stream_url(env_name, expected): def _atlas_exact_stream_url(env_name, expected):
@ -358,6 +362,7 @@ def _handle_atlas_streaming_capability(handler):
"""Advertise only transports whose immutable in-cluster URLs are configured.""" """Advertise only transports whose immutable in-cluster URLs are configured."""
tts = bool(_atlas_exact_stream_url("HERMES_WEBUI_ATLAS_TTS_STREAM_URL", ATLAS_TTS_STREAM_URL)) tts = bool(_atlas_exact_stream_url("HERMES_WEBUI_ATLAS_TTS_STREAM_URL", ATLAS_TTS_STREAM_URL))
stt = bool(_atlas_exact_stream_url("HERMES_WEBUI_ATLAS_STT_STREAM_URL", ATLAS_STT_STREAM_URL)) stt = bool(_atlas_exact_stream_url("HERMES_WEBUI_ATLAS_STT_STREAM_URL", ATLAS_STT_STREAM_URL))
preflight = True
j(handler, { j(handler, {
"tts": { "tts": {
"available": tts, "available": tts,
@ -372,10 +377,85 @@ def _handle_atlas_streaming_capability(handler):
"format": "pcm_s16le", "format": "pcm_s16le",
"sample_rate": 16000, "sample_rate": 16000,
}, },
"preflight": {
"available": preflight,
"path": "/api/voice/route-preflight",
"advisory": True,
},
}, extra_headers={"Cache-Control": "no-store"}) }, extra_headers={"Cache-Control": "no-store"})
return True return True
def _atlas_voice_preflight_payload(data):
"""Validate a provisional transcript without accepting routing authority."""
if not isinstance(data, dict):
raise ValueError("invalid request body")
turn_id = data.get("turn_id")
revision = data.get("revision")
transcript = data.get("transcript")
if not isinstance(turn_id, str) or not re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", turn_id):
raise ValueError("invalid turn_id")
if isinstance(revision, bool) or not isinstance(revision, int) or not 1 <= revision <= 1000000:
raise ValueError("invalid revision")
if not isinstance(transcript, str):
raise ValueError("invalid transcript")
transcript = " ".join(transcript.split())
if not 12 <= len(transcript) <= 512:
raise ValueError("invalid transcript")
return {"turn_id": turn_id, "revision": revision, "transcript": transcript}
def _atlas_voice_preflight_open(request, timeout=1.0):
"""Open only the immutable in-cluster advisory endpoint without proxies."""
return build_opener(ProxyHandler({}), _NoRedirectTtsHandler()).open(request, timeout=timeout)
def _handle_atlas_voice_preflight(handler):
"""Relay one same-origin local advisory and expose no classifier output."""
if not _check_same_origin_browser_request(handler):
return bad(handler, "Voice route preflight origin validation failed", 403)
target = ATLAS_VOICE_PREFLIGHT_URL
try:
payload = _atlas_voice_preflight_payload(read_body(handler))
except (TypeError, ValueError) as exc:
return bad(handler, str(exc), 400)
request = Request(
target,
data=json.dumps(payload, separators=(",", ":")).encode("utf-8"),
headers={"Content-Type": "application/json", "Accept": "application/json"},
)
try:
with _atlas_voice_preflight_open(request, timeout=1.0) as upstream:
raw = upstream.read(ATLAS_VOICE_PREFLIGHT_MAX_RESPONSE_BYTES + 1)
if len(raw) > ATLAS_VOICE_PREFLIGHT_MAX_RESPONSE_BYTES:
raise ValueError("oversized advisory response")
result = json.loads(raw)
tier = result.get("tier") if isinstance(result, dict) else None
target_hint = result.get("target") if isinstance(result, dict) else None
if (
result.get("turn_id") != payload["turn_id"]
or result.get("revision") != payload["revision"]
or tier not in ATLAS_VOICE_PREFLIGHT_TIERS
or target_hint != "atlas/auto/" + tier
or result.get("advisory") is not True
):
raise ValueError("invalid advisory response")
except Exception:
# Advisory failure never changes or delays the final Switchyard request.
return bad(handler, "Voice route preflight unavailable", 503)
return j(
handler,
{
"turn_id": payload["turn_id"],
"revision": payload["revision"],
"tier": tier,
"target": target_hint,
"advisory": True,
},
extra_headers={"Cache-Control": "no-store"},
)
def _atlas_tts_stream_payload(data): def _atlas_tts_stream_payload(data):
"""Build the narrow Piper payload used by the raw-PCM stream endpoint.""" """Build the narrow Piper payload used by the raw-PCM stream endpoint."""
if not isinstance(data, dict): if not isinstance(data, dict):
@ -390,8 +470,13 @@ def _atlas_tts_stream_payload(data):
language = _atlas_tts_language(data) language = _atlas_tts_language(data)
if language: if language:
payload["language"] = language payload["language"] = language
cue_id = data.get("cue_id")
if "cue_id" in data:
if not language or cue_id not in ATLAS_TTS_CUE_IDS:
raise ValueError("invalid thinking cue")
payload["cue_id"] = cue_id
turn_id = data.get("turn_id") turn_id = data.get("turn_id")
if isinstance(turn_id, str) and re.fullmatch(r"[A-Za-z0-9._-]{1,64}", turn_id): if isinstance(turn_id, str) and re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", turn_id):
payload["turn_id"] = turn_id payload["turn_id"] = turn_id
return payload return payload
@ -636,6 +721,9 @@ replace_exact(
''' if parsed.path == "/api/transcribe": ''' if parsed.path == "/api/transcribe":
return handle_transcribe(handler) return handle_transcribe(handler)
if parsed.path == "/api/voice/route-preflight":
return _handle_atlas_voice_preflight(handler)
if parsed.path == "/api/tts/stream": if parsed.path == "/api/tts/stream":
return _handle_atlas_tts_stream(handler) return _handle_atlas_tts_stream(handler)

File diff suppressed because it is too large Load Diff

View File

@ -1,5 +1,5 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Stamp Atlas WebUI shell URLs with the immutable image release identity.""" """Bind the WebUI client and server to one immutable release identity."""
from __future__ import annotations from __future__ import annotations
@ -23,3 +23,60 @@ for relative in ("static/index.html", "static/sw.js"):
source.replace("__WEBUI_VERSION__", f"__WEBUI_VERSION__-{RELEASE_ID}"), source.replace("__WEBUI_VERSION__", f"__WEBUI_VERSION__-{RELEASE_ID}"),
encoding="utf-8", encoding="utf-8",
) )
routes = ROOT / "api/routes.py"
routes_source = routes.read_text(encoding="utf-8")
routes_marker = ' settings["webui_version"] = WEBUI_VERSION\n'
if routes_source.count(routes_marker) != 1:
raise SystemExit(f"Hermes release settings context changed in {routes}")
routes.write_text(
routes_source.replace(
routes_marker,
routes_marker
+ " settings[\"webui_bundle_version\"] = "
+ f'f"{{WEBUI_VERSION}}-{RELEASE_ID}"\n',
),
encoding="utf-8",
)
panels = ROOT / "static/panels.js"
panels_source = panels.read_text(encoding="utf-8")
server_marker = (
" const server=_normalizeWebUIVersion(settings.webui_version);\n"
)
server_replacement = (
" const server=_normalizeWebUIVersion("
"settings.webui_bundle_version||settings.webui_version);\n"
)
if panels_source.count(server_marker) != 1:
raise SystemExit(f"Hermes release skew context changed in {panels}")
matching_marker = " if(client===server) return;\n"
matching_replacement = (
" if(client===server){\n"
" const banner=document.getElementById('staleClientBanner');\n"
" if(banner) banner.style.display='none';\n"
" return;\n"
" }\n"
)
if panels_source.count(matching_marker) != 1:
raise SystemExit(f"Hermes release match context changed in {panels}")
poll_marker = " if(_isBannerVisible()) return;\n"
if panels_source.count(poll_marker) != 1:
raise SystemExit(f"Hermes release poll context changed in {panels}")
visible_stop_marker = (
" if(_isBannerVisible()){ clearInterval(_pollTimer); "
"_pollTimer=null; return; }\n"
)
if panels_source.count(visible_stop_marker) != 1:
raise SystemExit(f"Hermes release monitor context changed in {panels}")
panels.write_text(
panels_source.replace(server_marker, server_replacement)
.replace(matching_marker, matching_replacement)
.replace(poll_marker, "")
.replace(visible_stop_marker, ""),
encoding="utf-8",
)

View File

@ -4,6 +4,7 @@
from __future__ import annotations from __future__ import annotations
import json import json
import re
import sys import sys
from collections.abc import Callable from collections.abc import Callable
from typing import Any from typing import Any
@ -95,6 +96,34 @@ def smoke(
raise RuntimeError(f"served / omitted Hermes persona link: {source}") raise RuntimeError(f"served / omitted Hermes persona link: {source}")
if "static/hermes-brand.css" not in root_html: if "static/hermes-brand.css" not in root_html:
raise RuntimeError("served / omitted static/hermes-brand.css") raise RuntimeError("served / omitted static/hermes-brand.css")
version_match = re.search(
r"window\.__HERMES_WEBUI_BUNDLE_VERSION__='([^']+)'",
root_html,
)
if not version_match:
raise RuntimeError("served / omitted the WebUI bundle identity")
client_version = version_match.group(1)
settings_body, _settings_url = _get(
urljoin(base_url, "api/settings"), opener
)
try:
settings = json.loads(settings_body)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise RuntimeError("served /api/settings is not valid JSON") from exc
server_version = settings.get("webui_bundle_version")
if not isinstance(server_version, str) or server_version != client_version:
raise RuntimeError(
"served WebUI client/server release identities do not match"
)
worker_body, _worker_url = _get(urljoin(base_url, "sw.js"), opener)
try:
worker = worker_body.decode("utf-8")
except UnicodeDecodeError as exc:
raise RuntimeError("served /sw.js is not UTF-8 JavaScript") from exc
if f"hermes-shell-{client_version}" not in worker:
raise RuntimeError("served /sw.js uses a different release identity")
direct_icon, _direct_url = _get( direct_icon, _direct_url = _get(
urljoin(base_url, "static/hermes-agent-192.png"), opener urljoin(base_url, "static/hermes-agent-192.png"), opener
@ -107,6 +136,7 @@ def smoke(
"icon_requests": len(icon_sources), "icon_requests": len(icon_sources),
"root": base_url, "root": base_url,
"direct_icon": "static/hermes-agent-192.png", "direct_icon": "static/hermes-agent-192.png",
"release_identity": client_version,
} }
@ -118,6 +148,7 @@ def main() -> int:
"Hermes WebUI smoke passed: " "Hermes WebUI smoke passed: "
f"manifest={result['manifest']} " f"manifest={result['manifest']} "
f"icon_requests={result['icon_requests']} " f"icon_requests={result['icon_requests']} "
f"release_identity={result['release_identity']} "
"root=branded direct_icon=png" "root=branded direct_icon=png"
) )
return 0 return 0

View File

@ -0,0 +1,119 @@
"""Precompute the small, localized Hermes thinking-cue audio set."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable
CUE_TEXT = {
"en": {
"thinking": "I'm thinking.",
"let_me_think": "Let me think.",
"still_working": "Still working on that.",
"one_more_moment": "One more moment.",
},
"ru": {
"thinking": "Я думаю.",
"let_me_think": "Дайте подумать.",
"still_working": "Я всё ещё думаю над этим.",
"one_more_moment": "Ещё мгновение.",
},
"es": {
"thinking": "Estoy pensando.",
"let_me_think": "Déjame pensar.",
"still_working": "Sigo pensando en eso.",
"one_more_moment": "Un momento más.",
},
}
LANGUAGE_VOICE = {
"en": "en_US-amy-medium",
"ru": "ru_RU-irina-medium",
"es": "es_MX-claude-high",
}
@dataclass(frozen=True)
class CachedCue:
"""One immutable PCM response synthesized before the server becomes ready."""
cue_id: str
language: str
voice_name: str
sample_rate: int
pcm: bytes
def build_cue_cache(
voices: dict,
synthesis_config_factory: Callable[..., object],
normalize_pcm: Callable[[object, int, int], object],
pause_for_text: Callable[[str], int],
) -> dict[tuple[str, str], CachedCue]:
"""Synthesize every allow-listed cue once, failing startup on voice drift."""
cache: dict[tuple[str, str], CachedCue] = {}
for language, entries in CUE_TEXT.items():
voice_name = LANGUAGE_VOICE[language]
voice = voices[voice_name]
sample_rate = int(voice.config.sample_rate)
for cue_id, text in entries.items():
chunks: list[bytes] = []
for result in voice.synthesize(text, synthesis_config_factory(length_scale=1.0)):
if (
result.sample_rate != sample_rate
or result.sample_width != 2
or result.sample_channels != 1
):
raise RuntimeError("Piper returned an unexpected cue PCM format")
chunks.append(
bytes(
normalize_pcm(
result.audio_int16_bytes,
sample_rate,
pause_for_text(text),
)
)
)
if not chunks:
raise RuntimeError(f"Piper returned no audio for cue {language}/{cue_id}")
cache[(language, cue_id)] = CachedCue(
cue_id, language, voice_name, sample_rate, b"".join(chunks)
)
return cache
def resolve_cached_cue(payload: object, cache: dict) -> CachedCue | None:
"""Resolve only an exact language/cue ID pair; client text is never trusted."""
if not isinstance(payload, dict) or "cue_id" not in payload:
return None
language = payload.get("language")
cue_id = payload.get("cue_id")
if not isinstance(language, str) or not isinstance(cue_id, str):
raise ValueError("invalid thinking cue")
cue = cache.get((language, cue_id))
if cue is None:
raise ValueError("unknown thinking cue")
return cue
def write_cached_cue(handler, cue: CachedCue, turn_id: str | None) -> None:
"""Return immutable PCM without acquiring the live Piper model lock."""
handler.send_response(200)
handler.send_header("Content-Type", "audio/pcm")
handler.send_header("Content-Length", str(len(cue.pcm)))
handler.send_header("Cache-Control", "private, max-age=86400, immutable")
handler.send_header("X-Audio-Format", "pcm_s16le")
handler.send_header("X-Audio-Sample-Rate", str(cue.sample_rate))
handler.send_header("X-Audio-Channels", "1")
handler.send_header("X-Audio-Sample-Width", "2")
handler.send_header("X-TTS-Voice", cue.voice_name)
handler.send_header("X-TTS-Cue-ID", cue.cue_id)
handler.send_header("X-TTS-Cache", "HIT")
if turn_id is not None:
handler.send_header("X-TTS-Turn-ID", turn_id)
handler.end_headers()
try:
handler.wfile.write(cue.pcm)
handler.wfile.flush()
except (BrokenPipeError, ConnectionResetError, TimeoutError, OSError):
handler.close_connection = True

View File

@ -134,6 +134,7 @@ configMapGenerator:
- configure_agent_clients.py=scripts/configure_agent_clients.py - configure_agent_clients.py=scripts/configure_agent_clients.py
- codex_broker.py=scripts/codex_broker.py - codex_broker.py=scripts/codex_broker.py
- classifier_broker.py=scripts/classifier_broker.py - classifier_broker.py=scripts/classifier_broker.py
- voice_route_preflight.py=scripts/voice_route_preflight.py
- claude_oauth_broker.py=scripts/claude_oauth_broker.py - claude_oauth_broker.py=scripts/claude_oauth_broker.py
- worker_route_broker.py=scripts/worker_route_broker.py - worker_route_broker.py=scripts/worker_route_broker.py
- hermes_coordinator.py=scripts/hermes_coordinator.py - hermes_coordinator.py=scripts/hermes_coordinator.py

View File

@ -405,6 +405,7 @@ spec:
app: hermes-switchyard app: hermes-switchyard
ports: ports:
- {protocol: TCP, port: 9005} - {protocol: TCP, port: 9005}
- {protocol: TCP, port: 9009}
- to: - to:
- podSelector: - podSelector:
matchLabels: matchLabels:
@ -457,6 +458,12 @@ spec:
values: [hermes, hermes-agent, hermes-chat-tenant] values: [hermes, hermes-agent, hermes-chat-tenant]
ports: ports:
- {protocol: TCP, port: 9005} - {protocol: TCP, port: 9005}
- from:
- podSelector:
matchLabels:
app: hermes-chat-tenant
ports:
- {protocol: TCP, port: 9009}
- from: - from:
- namespaceSelector: - namespaceSelector:
matchLabels: matchLabels:
@ -466,6 +473,7 @@ spec:
app: server app: server
ports: ports:
- {protocol: TCP, port: 9005} - {protocol: TCP, port: 9005}
- {protocol: TCP, port: 9009}
egress: egress:
- to: - to:
- namespaceSelector: - namespaceSelector:

View File

@ -6,14 +6,29 @@ from __future__ import annotations
import copy import copy
import json import json
import os import os
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any, Final from typing import Any, Final
import httpx import httpx
try:
from voice_route_preflight import (
COORDINATOR as VOICE_PREFLIGHT,
MAX_BODY_BYTES as VOICE_PREFLIGHT_MAX_BODY_BYTES,
validate_request as validate_voice_preflight_request,
)
except ModuleNotFoundError: # Test imports use the repository package path.
from services.hermes.scripts.voice_route_preflight import (
COORDINATOR as VOICE_PREFLIGHT,
MAX_BODY_BYTES as VOICE_PREFLIGHT_MAX_BODY_BYTES,
validate_request as validate_voice_preflight_request,
)
HOST: Final = os.environ.get("HERMES_CLASSIFIER_BROKER_HOST", "0.0.0.0") HOST: Final = os.environ.get("HERMES_CLASSIFIER_BROKER_HOST", "0.0.0.0")
PORT: Final = int(os.environ.get("HERMES_CLASSIFIER_BROKER_PORT", "9008")) PORT: Final = int(os.environ.get("HERMES_CLASSIFIER_BROKER_PORT", "9008"))
VOICE_PORT: Final = int(os.environ.get("HERMES_VOICE_PREFLIGHT_PORT", "9009"))
UPSTREAM: Final = os.environ.get( UPSTREAM: Final = os.environ.get(
"HERMES_CLASSIFIER_BROKER_UPSTREAM", "HERMES_CLASSIFIER_BROKER_UPSTREAM",
"http://ollama.ai.svc.cluster.local:11434", "http://ollama.ai.svc.cluster.local:11434",
@ -31,6 +46,7 @@ READ_TIMEOUT_SECONDS: Final = float(
os.environ.get("HERMES_CLASSIFIER_BROKER_READ_TIMEOUT", "60") os.environ.get("HERMES_CLASSIFIER_BROKER_READ_TIMEOUT", "60")
) )
ALLOWED_PATHS: Final = {"/v1/chat/completions", "/v1/models"} ALLOWED_PATHS: Final = {"/v1/chat/completions", "/v1/models"}
VOICE_PREFLIGHT_PATH: Final = "/voice/route-preflight"
def _bounded_text(value: str, limit: int) -> str: def _bounded_text(value: str, limit: int) -> str:
@ -177,10 +193,28 @@ class Handler(BaseHTTPRequestHandler):
self.end_headers() self.end_headers()
self.wfile.write(body) self.wfile.write(body)
def _empty(self, status: int) -> None:
self.send_response(status)
self.send_header("Content-Length", "0")
self.send_header("Cache-Control", "no-store")
self.end_headers()
def _text(self, status: int, body: str, content_type: str) -> None:
encoded = body.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(encoded)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(encoded)
def do_GET(self) -> None: # noqa: N802 def do_GET(self) -> None: # noqa: N802
if self.path == "/health": if self.path == "/health":
self._json(200, {"ok": True, "upstream": "ollama"}) self._json(200, {"ok": True, "upstream": "ollama"})
return return
if self.path == "/metrics":
self._text(200, VOICE_PREFLIGHT.metrics(), "text/plain; version=0.0.4")
return
if self.path not in ALLOWED_PATHS: if self.path not in ALLOWED_PATHS:
self._json(404, {"error": "not found"}) self._json(404, {"error": "not found"})
return return
@ -190,28 +224,81 @@ class Handler(BaseHTTPRequestHandler):
if self.path not in ALLOWED_PATHS: if self.path not in ALLOWED_PATHS:
self._json(404, {"error": "not found"}) self._json(404, {"error": "not found"})
return return
authoritative = self.path == "/v1/chat/completions"
if authoritative:
# Mark priority as soon as the authoritative path arrives, before
# even validating or reading its local body, so an
# in-flight disposable decode is closed at authoritative arrival.
VOICE_PREFLIGHT.begin_authoritative()
try:
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
length = -1
if length <= 0 or length > MAX_BODY_BYTES:
self._json(413, {"error": "request too large"})
return
try:
payload = json.loads(self.rfile.read(length))
if not isinstance(payload, dict):
raise ValueError("request must be a JSON object")
body = json.dumps(
compact_payload(payload), separators=(",", ":")
).encode("utf-8")
except (ValueError, TypeError, json.JSONDecodeError) as exc:
self._json(400, {"error": str(exc)})
return
print(
f"classifier-broker request_bytes={length} compacted_bytes={len(body)}",
flush=True,
)
if authoritative:
self._proxy(body)
return
self._proxy(body)
finally:
if authoritative:
VOICE_PREFLIGHT.end_authoritative()
def _voice_preflight(self) -> None:
"""Return one disposable local tier without starting a hosted turn."""
try: try:
length = int(self.headers.get("Content-Length", "0")) length = int(self.headers.get("Content-Length", "0"))
except ValueError: except ValueError:
length = -1 length = -1
if length <= 0 or length > MAX_BODY_BYTES: if length <= 0 or length > VOICE_PREFLIGHT_MAX_BODY_BYTES:
self._json(413, {"error": "request too large"}) self._json(413, {"error": "request too large"})
return return
try: try:
payload = json.loads(self.rfile.read(length)) payload = json.loads(self.rfile.read(length))
if not isinstance(payload, dict): if not isinstance(payload, dict):
raise ValueError("request must be a JSON object") raise ValueError("request must be a JSON object")
body = json.dumps( turn_id, revision, transcript = validate_voice_preflight_request(payload)
compact_payload(payload), separators=(",", ":")
).encode("utf-8")
except (ValueError, TypeError, json.JSONDecodeError) as exc: except (ValueError, TypeError, json.JSONDecodeError) as exc:
self._json(400, {"error": str(exc)}) self._json(400, {"error": str(exc)})
return return
print(
f"classifier-broker request_bytes={length} compacted_bytes={len(body)}", cancel = VOICE_PREFLIGHT.begin(turn_id)
flush=True, if cancel is None:
self._empty(204)
return
try:
tier = VOICE_PREFLIGHT.stream(transcript, cancel)
finally:
VOICE_PREFLIGHT.end()
if not tier:
self._empty(204)
return
self._json(
200,
{
"turn_id": turn_id,
"revision": revision,
"tier": tier,
"target": f"atlas/auto/{tier}",
"advisory": True,
},
) )
self._proxy(body)
def _proxy(self, body: bytes | None) -> None: def _proxy(self, body: bytes | None) -> None:
try: try:
@ -235,9 +322,37 @@ class Handler(BaseHTTPRequestHandler):
self._json(503, {"error": f"classifier unavailable: {exc}"}) self._json(503, {"error": f"classifier unavailable: {exc}"})
class VoiceHandler(Handler):
"""Expose only advisory, health and telemetry routes to chat tenants."""
server_version = "HermesVoicePreflight/1"
def do_GET(self) -> None: # noqa: N802
if self.path == "/health":
self._json(200, {"ok": True, "upstream": "ollama"})
return
if self.path == "/metrics":
self._text(200, VOICE_PREFLIGHT.metrics(), "text/plain; version=0.0.4")
return
self._json(404, {"error": "not found"})
def do_POST(self) -> None: # noqa: N802
if self.path == VOICE_PREFLIGHT_PATH:
self._voice_preflight()
return
self._json(404, {"error": "not found"})
def main() -> None: def main() -> None:
voice_server = ThreadingHTTPServer((HOST, VOICE_PORT), VoiceHandler)
voice_thread = threading.Thread(target=voice_server.serve_forever, daemon=True)
voice_thread.start()
server = ThreadingHTTPServer((HOST, PORT), Handler) server = ThreadingHTTPServer((HOST, PORT), Handler)
server.serve_forever() try:
server.serve_forever()
finally:
voice_server.shutdown()
voice_server.server_close()
if __name__ == "__main__": if __name__ == "__main__":

View File

@ -0,0 +1,305 @@
#!/usr/bin/env python3
"""Bounded local route advice for provisional Hermes voice transcripts."""
from __future__ import annotations
import json
import os
import re
import threading
import time
from typing import Any, Final
import httpx
UPSTREAM: Final = os.environ.get(
"HERMES_CLASSIFIER_BROKER_UPSTREAM",
"http://ollama.ai.svc.cluster.local:11434",
).rstrip("/")
MODEL: Final = os.environ.get(
"HERMES_VOICE_PREFLIGHT_MODEL", "qwen2.5:14b-instruct-q4_0"
)
MAX_BODY_BYTES: Final = 2 * 1024
MAX_TEXT_CHARS: Final = 512
TIMEOUT_SECONDS: Final = max(
0.1, min(0.75, float(os.environ.get("HERMES_VOICE_PREFLIGHT_TIMEOUT", "0.7")))
)
TIERS: Final = {"fast", "balanced", "deep", "maximum"}
TURN_PATTERN: Final = re.compile(r"[A-Za-z0-9_.:-]{1,128}\Z")
SEEN_TTL_SECONDS: Final = 120.0
MAX_SEEN_TURNS: Final = 256
def validate_request(payload: dict[str, Any]) -> tuple[str, int, str]:
"""Validate one bounded provisional transcript without identity coercion."""
turn_id = payload.get("turn_id")
revision = payload.get("revision")
transcript = payload.get("transcript")
if not isinstance(turn_id, str) or not TURN_PATTERN.fullmatch(turn_id):
raise ValueError("invalid turn_id")
if isinstance(revision, bool) or not isinstance(revision, int):
raise ValueError("invalid revision")
if revision < 1 or revision > 1_000_000:
raise ValueError("invalid revision")
if not isinstance(transcript, str):
raise ValueError("invalid transcript")
transcript = " ".join(transcript.split())
if len(transcript) < 12 or len(transcript) > MAX_TEXT_CHARS:
raise ValueError("invalid transcript")
return turn_id, revision, transcript
def inference_payload(transcript: str) -> dict[str, Any]:
"""Build the tiny local-only classification request used during speech."""
return {
"model": MODEL,
"messages": [
{
"role": "system",
"content": (
"Classify this incomplete spoken request into one advisory effort "
"tier. Return JSON only as {\"tier\":\"fast|balanced|deep|maximum\"}. "
"Use fast for simple conversation or facts, balanced for ordinary "
"assistance, deep for multi-step work, and maximum only for explicit "
"high-stakes or unusually complex work. This is not authoritative."
),
},
{"role": "user", "content": transcript},
],
"temperature": 0,
"max_tokens": 16,
"stream": True,
"response_format": {"type": "json_object"},
}
class Coordinator:
"""Serialize disposable inference and give final classification priority."""
def __init__(self) -> None:
self.condition = threading.Condition()
self.authoritative = 0
self.active = False
self.cancel: threading.Event | None = None
self.client: Any = None
self.response: Any = None
self.seen: dict[str, float] = {}
self.stats = {
"admitted": 0,
"rejected": 0,
"success": 0,
"timeout": 0,
"cancelled": 0,
"failure": 0,
"preempted": 0,
}
self.duration_count = 0
self.duration_sum = 0.0
def _prune(self, now: float) -> None:
self.seen = {
turn: created
for turn, created in self.seen.items()
if now - created <= SEEN_TTL_SECONDS
}
if len(self.seen) > MAX_SEEN_TURNS:
self.seen = dict(sorted(self.seen.items(), key=lambda item: item[1])[-MAX_SEEN_TURNS:])
def begin_authoritative(self) -> None:
"""Cancel advisory I/O as soon as a real classifier request arrives."""
with self.condition:
self.authoritative += 1
if self.cancel is not None:
self.cancel.set()
self.stats["preempted"] += 1
resources = (self.response, self.client)
for resource in resources:
if resource is not None:
try:
resource.close()
except Exception:
pass
def end_authoritative(self) -> None:
"""Release the real-classifier priority marker."""
with self.condition:
self.authoritative = max(0, self.authoritative - 1)
self.condition.notify_all()
def begin(self, turn_id: str) -> threading.Event | None:
"""Admit at most one advisory globally and once per browser turn."""
now = time.monotonic()
with self.condition:
self._prune(now)
if self.active or self.authoritative or turn_id in self.seen:
self.stats["rejected"] += 1
return None
self.seen[turn_id] = now
self.active = True
self.cancel = threading.Event()
self.stats["admitted"] += 1
return self.cancel
def register(
self, cancel: threading.Event, *, client: Any = None, response: Any = None
) -> bool:
"""Publish cancellable resources unless authority already preempted them."""
with self.condition:
if (
not self.active
or self.cancel is not cancel
or cancel.is_set()
or self.authoritative
):
cancel.set()
return False
if client is not None:
self.client = client
if response is not None:
self.response = response
return True
def end(self) -> None:
"""Release the disposable local lane after its bounded request."""
with self.condition:
self.active = False
self.cancel = None
self.client = None
self.response = None
self.condition.notify_all()
def _record(self, outcome: str, started: float) -> None:
"""Record bounded low-cardinality timing without transcript or turn labels."""
elapsed = max(0.0, time.monotonic() - started)
with self.condition:
self.stats[outcome] += 1
self.duration_count += 1
self.duration_sum += elapsed
print(
f"voice-route-preflight outcome={outcome} duration_ms={elapsed * 1000:.1f}",
flush=True,
)
def metrics(self) -> str:
"""Render low-cardinality Prometheus telemetry with no user content."""
with self.condition:
stats = dict(self.stats)
duration_count = self.duration_count
duration_sum = self.duration_sum
authority = self.authoritative
active = int(self.active)
lines = [
"# HELP hermes_voice_route_preflight_total Local voice route preflight outcomes.",
"# TYPE hermes_voice_route_preflight_total counter",
]
lines.extend(
f'hermes_voice_route_preflight_total{{outcome="{name}"}} {value}'
for name, value in sorted(stats.items())
)
lines.extend(
[
"# TYPE hermes_voice_route_preflight_duration_seconds summary",
f"hermes_voice_route_preflight_duration_seconds_count {duration_count}",
f"hermes_voice_route_preflight_duration_seconds_sum {duration_sum:.6f}",
"# TYPE hermes_voice_route_preflight_active gauge",
f"hermes_voice_route_preflight_active {active}",
"# TYPE hermes_classifier_authoritative_active gauge",
f"hermes_classifier_authoritative_active {authority}",
]
)
return "\n".join(lines) + "\n"
def stream(self, transcript: str, cancel: threading.Event) -> str:
"""Run one hard-bounded cancellable local Qwen advisory decode."""
started = time.monotonic()
outcome = "failure"
deadline = time.monotonic() + TIMEOUT_SECONDS
timeout = httpx.Timeout(
min(0.35, TIMEOUT_SECONDS),
connect=min(0.25, TIMEOUT_SECONDS),
read=min(0.35, TIMEOUT_SECONDS),
write=min(0.25, TIMEOUT_SECONDS),
pool=min(0.1, TIMEOUT_SECONDS),
)
client = httpx.Client(timeout=timeout)
if not self.register(cancel, client=client):
client.close()
self._record("cancelled", started)
return ""
response = None
def expire() -> None:
cancel.set()
for resource in (response, client):
if resource is not None:
try:
resource.close()
except Exception:
pass
timer = threading.Timer(max(0.01, deadline - time.monotonic()), expire)
timer.daemon = True
timer.start()
content = ""
try:
request = client.build_request(
"POST",
f"{UPSTREAM}/v1/chat/completions",
headers={"Content-Type": "application/json"},
json=inference_payload(transcript),
)
response = client.send(request, stream=True)
if not self.register(cancel, response=response):
outcome = "cancelled"
response.close()
return ""
response.raise_for_status()
for line in response.iter_lines():
if cancel.is_set() or time.monotonic() >= deadline:
break
if not line.startswith("data:"):
continue
event = line[5:].strip()
if event == "[DONE]":
break
try:
delta = (json.loads(event)["choices"][0].get("delta") or {}).get(
"content", ""
)
except (KeyError, IndexError, TypeError, json.JSONDecodeError):
continue
if isinstance(delta, str):
content += delta
if len(content) > 128:
cancel.set()
break
if cancel.is_set() or time.monotonic() >= deadline:
outcome = "timeout" if time.monotonic() >= deadline else "cancelled"
return ""
decoded = json.loads(content)
tier = decoded.get("tier") if isinstance(decoded, dict) else None
if tier in TIERS:
outcome = "success"
return tier
return ""
except Exception:
outcome = "cancelled" if cancel.is_set() else "failure"
return ""
finally:
cancel.set()
timer.cancel()
if response is not None:
try:
response.close()
except Exception:
pass
try:
client.close()
except Exception:
pass
self._record(outcome, started)
COORDINATOR = Coordinator()

View File

@ -22,7 +22,7 @@ spec:
labels: labels:
app: hermes-switchyard app: hermes-switchyard
annotations: annotations:
ai.bstein.dev/config-rev: "20260823-provider-neutral-pools" ai.bstein.dev/config-rev: "20260823-voice-route-preflight-v1"
prometheus.io/scrape: "true" prometheus.io/scrape: "true"
prometheus.io/port: "9005" prometheus.io/port: "9005"
prometheus.io/path: /metrics prometheus.io/path: /metrics
@ -182,11 +182,16 @@ spec:
- name: classifier - name: classifier
containerPort: 9008 containerPort: 9008
protocol: TCP protocol: TCP
- name: voice-preflight
containerPort: 9009
protocol: TCP
env: env:
- name: HERMES_CLASSIFIER_BROKER_UPSTREAM - name: HERMES_CLASSIFIER_BROKER_UPSTREAM
value: http://ollama.ai.svc.cluster.local:11434 value: http://ollama.ai.svc.cluster.local:11434
- name: HERMES_CLASSIFIER_BROKER_READ_TIMEOUT - name: HERMES_CLASSIFIER_BROKER_READ_TIMEOUT
value: "60" value: "60"
- name: HERMES_VOICE_PREFLIGHT_MODEL
value: qwen2.5:14b-instruct-q4_0
readinessProbe: readinessProbe:
httpGet: httpGet:
path: /health path: /health

View File

@ -4,6 +4,10 @@ kind: Service
metadata: metadata:
name: hermes-switchyard name: hermes-switchyard
namespace: hermes namespace: hermes
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "9009"
prometheus.io/path: /metrics
labels: labels:
app: hermes-switchyard app: hermes-switchyard
spec: spec:
@ -15,3 +19,7 @@ spec:
port: 9005 port: 9005
targetPort: http targetPort: http
protocol: TCP protocol: TCP
- name: voice-preflight
port: 9009
targetPort: voice-preflight
protocol: TCP

View File

@ -654,11 +654,11 @@ data:
} }
pipelineJob('hermes-agent-image') { pipelineJob('hermes-agent-image') {
disabled(false) disabled(false)
description('Bounded daemonless Kaniko release for the latest atlas/titan-iac main containing a reviewed commit. Archives exact evidence, then publishes an immutable tag for Flux deployment.') description('Bounded daemonless Kaniko release for one exact reviewed atlas/titan-iac revision already contained by main. Archives exact evidence, then publishes an immutable tag for Flux deployment.')
authenticationToken(System.getenv('HERMES_AGENT_IMAGE_BUILD_TOKEN')) authenticationToken(System.getenv('HERMES_AGENT_IMAGE_BUILD_TOKEN'))
parameters { parameters {
booleanParam('PUBLISH_IMAGE', false, 'Publish the reviewed Hermes agent image.') booleanParam('PUBLISH_IMAGE', false, 'Publish the reviewed Hermes agent image.')
stringParam('EXPECTED_SOURCE_REVISION', '', 'Full reviewed commit that must be contained by atlas/titan-iac main.') stringParam('EXPECTED_SOURCE_REVISION', '', 'Full reviewed commit already contained by atlas/titan-iac main.')
stringParam('CONFIRM_PUBLISH', '', 'Exact confirmation: PUBLISH HERMES AGENT') stringParam('CONFIRM_PUBLISH', '', 'Exact confirmation: PUBLISH HERMES AGENT')
} }
definition { definition {
@ -678,7 +678,7 @@ data:
} }
pipelineJob('hermes-webui-image') { pipelineJob('hermes-webui-image') {
disabled(false) disabled(false)
description('Bounded daemonless Kaniko release for the latest atlas/titan-iac main containing a reviewed commit. Archives exact evidence, then publishes an immutable tag for Flux deployment.') description('Bounded daemonless Kaniko release for one exact reviewed atlas/titan-iac revision already contained by main. Archives exact evidence, then publishes an immutable tag for Flux deployment.')
authenticationToken(System.getenv('HERMES_AGENT_IMAGE_BUILD_TOKEN')) authenticationToken(System.getenv('HERMES_AGENT_IMAGE_BUILD_TOKEN'))
parameters { parameters {
booleanParam('PUBLISH_IMAGE', false, 'Publish the reviewed Hermes WebUI image.') booleanParam('PUBLISH_IMAGE', false, 'Publish the reviewed Hermes WebUI image.')

View File

@ -7,6 +7,8 @@ Hermes WebUI image pinned by `dockerfiles/Dockerfile.hermes-webui`:
- Version: `0.52.181` - Version: `0.52.181`
- OCI source revision: `7a94e34a6d639576576baa9131acf6765f6d2b98` - OCI source revision: `7a94e34a6d639576576baa9131acf6765f6d2b98`
- Full upstream `static/index.html` SHA-256: `6e218d42f6e047168a774c59aa9fc98a55b608cdc070cf1049ad414a597a722c` - Full upstream `static/index.html` SHA-256: `6e218d42f6e047168a774c59aa9fc98a55b608cdc070cf1049ad414a597a722c`
- Full upstream `api/routes.py` SHA-256: `a09c3a236a323b0d1ad22409f6576fa21a0f3ca413ad8eeec06abb87067dc2ab`
- Full upstream `static/panels.js` SHA-256: `276e5dfc2cac6eee4a1820e4fab9c4da52743efed4a1b1e9d58d6dd15f6fd5c9`
The fixture stays intentionally narrow, but tests execute the shipped patchers The fixture stays intentionally narrow, but tests execute the shipped patchers
against it; large inline artwork bodies are reduced while their exact unique against it; large inline artwork bodies are reduced while their exact unique

View File

@ -44,6 +44,17 @@ def _handle_tts(handler, data, text, rate_str, engine):
def handle_get(handler, parsed) -> bool: def handle_get(handler, parsed) -> bool:
"""Handle all GET routes. Returns True if handled, False for 404.""" """Handle all GET routes. Returns True if handled, False for 404."""
if parsed.path == "/api/settings":
settings = {}
# Inject the running version so the UI badge stays in sync with git tags
# without any manual release step.
try:
from api.updates import AGENT_VERSION, WEBUI_VERSION
settings["webui_version"] = WEBUI_VERSION
settings["agent_version"] = AGENT_VERSION
except Exception:
pass
return j(handler, settings)
return False return False

View File

@ -5,6 +5,67 @@ const _SETTINGS_SPEECH_STORAGE_KEYS={
}; };
let _settingsSpeechChangedKeys=new Set(); let _settingsSpeechChangedKeys=new Set();
function _normalizeWebUIVersion(value){
if(!value) return '';
const s=String(value).trim();
if(!s) return '';
const lower=s.toLowerCase();
if(lower==='__webui_version__'||lower==='not detected'||lower==='unknown') return '';
return s;
}
function _currentWebUIBundleVersion(){
try{
const raw=window.__HERMES_WEBUI_BUNDLE_VERSION__;
if(!raw) return '';
let s=String(raw);
try{ s=decodeURIComponent(s.replace(/\+/g,' ')); }catch(_){}
return _normalizeWebUIVersion(s);
}catch(_){ return ''; }
}
function _showStaleWebUIClientBanner(clientVersion,serverVersion){
const banner=document.getElementById('staleClientBanner');
if(!banner) return;
const msg=document.getElementById('staleClientMessage');
const versions=document.getElementById('staleClientVersions');
if(msg) msg.textContent='This tab is running a different WebUI version. Hard refresh to restore full functionality.';
if(versions) versions.textContent='Running: '+clientVersion+' → Server: '+serverVersion;
banner.style.display='flex';
}
function checkWebUIVersionSkew(settings){
try{
if(!settings) return;
const client=_currentWebUIBundleVersion();
const server=_normalizeWebUIVersion(settings.webui_version);
if(!client||!server) return;
if(client===server) return;
_showStaleWebUIClientBanner(client,server);
}catch(_){}
}
window.checkWebUIVersionSkew=checkWebUIVersionSkew;
function _startWebUIVersionSkewMonitor(){
let _pollTimer=null;
function _isBannerVisible(){
const banner=document.getElementById('staleClientBanner');
return !!(banner&&banner.style.display==='flex');
}
function _check(){
if(_isBannerVisible()) return;
Promise.resolve().then(function(){ return api('/api/settings'); }).then(function(s){ checkWebUIVersionSkew(s); }).catch(function(){});
}
function _startPoll(){
if(_pollTimer||document.hidden) return;
_pollTimer=setInterval(function(){
if(document.hidden){ clearInterval(_pollTimer); _pollTimer=null; return; }
if(_isBannerVisible()){ clearInterval(_pollTimer); _pollTimer=null; return; }
_check();
},60000);
}
}
function _speechPreferencesPayloadFromUi(){ function _speechPreferencesPayloadFromUi(){
const payload={}; const payload={};
const ttsVoiceSel=$('settingsTtsVoice'); const ttsVoiceSel=$('settingsTtsVoice');

View File

@ -18,6 +18,7 @@
"dockerfiles/hermes-kaniko-heredoc-runner.py", "dockerfiles/hermes-kaniko-heredoc-runner.py",
"services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py", "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py",
"services/hermes/scripts/hermes_image_release_status.py", "services/hermes/scripts/hermes_image_release_status.py",
"services/hermes/scripts/voice_route_preflight.py",
"services/hermes/scripts/jenkins_image_build_trigger.py", "services/hermes/scripts/jenkins_image_build_trigger.py",
"ci/scripts/publish_test_metrics.py", "ci/scripts/publish_test_metrics.py",
"ci/scripts/publish_test_metrics_quality.py", "ci/scripts/publish_test_metrics_quality.py",
@ -109,6 +110,7 @@
"dockerfiles/hermes-kaniko-heredoc-runner.py", "dockerfiles/hermes-kaniko-heredoc-runner.py",
"services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py", "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py",
"services/hermes/scripts/hermes_image_release_status.py", "services/hermes/scripts/hermes_image_release_status.py",
"services/hermes/scripts/voice_route_preflight.py",
"services/hermes/scripts/jenkins_image_build_trigger.py", "services/hermes/scripts/jenkins_image_build_trigger.py",
"ci/scripts/publish_test_metrics.py", "ci/scripts/publish_test_metrics.py",
"ci/scripts/publish_test_metrics_quality.py", "ci/scripts/publish_test_metrics_quality.py",
@ -319,6 +321,7 @@
"minimum_percent": 95.0, "minimum_percent": 95.0,
"minimum_branch_percent": 95.0, "minimum_branch_percent": 95.0,
"branch_tracked_files": [ "branch_tracked_files": [
"services/hermes/scripts/voice_route_preflight.py",
"scripts/ops/hermes_handoff_acceptance.py", "scripts/ops/hermes_handoff_acceptance.py",
"scripts/ops/hermes_handoff_arming.py", "scripts/ops/hermes_handoff_arming.py",
"scripts/ops/hermes_handoff_catalog.py", "scripts/ops/hermes_handoff_catalog.py",
@ -340,6 +343,7 @@
"testing/quality_handoff_mutation.py" "testing/quality_handoff_mutation.py"
], ],
"tracked_files": [ "tracked_files": [
"services/hermes/scripts/voice_route_preflight.py",
"ci/scripts/hermes_image_release.py", "ci/scripts/hermes_image_release.py",
"dockerfiles/hermes-kaniko-heredoc-runner.py", "dockerfiles/hermes-kaniko-heredoc-runner.py",
"services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py", "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py",

View File

@ -492,6 +492,71 @@ scenarios.adaptive_chunks_are_sentence_gated = async () => {
return { partial, complete }; return { partial, complete };
}; };
scenarios.spoken_urls_are_skipped_without_damaging_text = async () => {
const harness = makeHarness();
await harness.flush();
const clean = harness.context._atlasStripHttpUrlsForSpeech;
return {
sentence: clean('Read this https://example.com/docs. Then continue.'),
wrapped: clean('Open (https://example.com/a_(b)). Next.'),
punctuated: clean('Try https://example.com/a?q=1, or https://example.org/x!'),
domains: clean('Keep example.com and sub.example.org exactly as written.'),
prose: clean('No links here; keep this sentence exactly as written.'),
};
};
scenarios.spoken_urls_are_elided_before_all_voice_routes = async () => {
const results = [];
for (const language of ['en', 'ru', 'es']) {
const harness = makeHarness();
await startVoiceMode(harness);
await runTurn(harness, {
transcript: 'Read the answer.',
language,
reply: 'Keep example.com, but skip https://private.example/path. Done.',
});
results.push({ language, tts: harness.ttsRequests });
}
return { results };
};
scenarios.canonical_pcm_fallback_builds_a_valid_wav = async () => {
const harness = makeHarness();
await harness.flush();
const pcm = new Uint8Array([0x00, 0x00, 0xff, 0x7f, 0x00, 0x80]);
const blob = harness.context._atlasPcm16WavBlob([pcm.buffer], 16000);
const bytes = new Uint8Array(blob.parts[0]);
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
const text = (start, length) => String.fromCharCode(...bytes.slice(start, start + length));
return {
type: blob.type,
size: bytes.length,
riff: text(0, 4),
wave: text(8, 4),
format: view.getUint16(20, true),
channels: view.getUint16(22, true),
rate: view.getUint32(24, true),
bits: view.getUint16(34, true),
data: view.getUint32(40, true),
pcm: Array.from(bytes.slice(44)),
};
};
scenarios.applied_aec_settings_are_fail_safe = async () => {
const harness = makeHarness();
await harness.flush();
const check = harness.context._atlasCaptureAecIsUsable;
const stream = (value) => ({
getAudioTracks: () => [{ getSettings: () => ({ echoCancellation: value }) }],
});
return {
applied: check(stream(true)),
rejected: check(stream(false)),
unknown: check({ getAudioTracks: () => [{ getSettings: () => ({}) }] }),
unsupported: check({}),
};
};
scenarios.first_sentence_speaks_before_stream_completion = async () => { scenarios.first_sentence_speaks_before_stream_completion = async () => {
const harness = makeHarness(); const harness = makeHarness();
await startVoiceMode(harness); await startVoiceMode(harness);

View File

@ -101,6 +101,8 @@ def test_pipeline_requires_reviewed_main_and_runtime_credentials() -> None:
assert "git rev-parse origin/main" in source assert "git rev-parse origin/main" in source
assert "git fetch --no-tags origin main" not in source assert "git fetch --no-tags origin main" not in source
assert 'git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}"' in source assert 'git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}"' in source
assert 'git checkout --detach "${EXPECTED_SOURCE_REVISION}"' in source
assert 'test "${actual_revision}" = "${EXPECTED_SOURCE_REVISION}"' in source
assert "credentialsId: 'harbor-robot'" in source assert "credentialsId: 'harbor-robot'" in source
assert "set +x" in source assert "set +x" in source
assert "umask 077" in source assert "umask 077" in source

View File

@ -8,6 +8,7 @@ import struct
import sys import sys
import threading import threading
import time import time
import wave
from pathlib import Path from pathlib import Path
from types import SimpleNamespace from types import SimpleNamespace
@ -63,6 +64,25 @@ class _Connection:
self.condition.wait(deadline - time.monotonic()) self.condition.wait(deadline - time.monotonic())
raise AssertionError(f"no {message_type} response: {self.messages}") raise AssertionError(f"no {message_type} response: {self.messages}")
def wait_for_revision(self, revision: int, timeout: float = 2.0) -> dict:
"""Wait for one particular rolling partial rather than an older one."""
deadline = time.monotonic() + timeout
with self.condition:
while time.monotonic() < deadline:
match = next(
(
message
for message in self.messages
if message.get("type") == "partial"
and message.get("revision") == revision
),
None,
)
if match:
return match
self.condition.wait(deadline - time.monotonic())
raise AssertionError(f"no rolling revision {revision}: {self.messages}")
class _Model: class _Model:
"""Return a deterministic Whisper-shaped result and retain decode options.""" """Return a deterministic Whisper-shaped result and retain decode options."""
@ -85,6 +105,46 @@ class _Model:
} }
class _SequencedModel(_Model):
"""Return evolving text and record how much PCM each decode consumed."""
def __init__(self, texts: list[str]) -> None:
super().__init__()
self.texts = texts
self.frame_counts: list[int] = []
def transcribe(self, path: str, **options: object) -> dict:
with wave.open(path, "rb") as wav_file:
self.frame_counts.append(wav_file.getnframes())
text = self.texts[min(len(self.calls), len(self.texts) - 1)]
self.calls.append((path, options))
return {
"text": f" {text}",
"language": "en",
"segments": [
{
"text": f" {text}",
"no_speech_prob": 0.01,
"avg_logprob": -0.1,
}
],
}
class _BlockingModel(_Model):
"""Hold inference so epoch/cancel behavior can be tested deterministically."""
def __init__(self) -> None:
super().__init__()
self.entered = threading.Event()
self.release = threading.Event()
def transcribe(self, path: str, **options: object) -> dict:
self.entered.set()
assert self.release.wait(2.0)
return super().transcribe(path, **options)
def _start(session) -> None: def _start(session) -> None:
session.start( session.start(
{ {
@ -105,6 +165,22 @@ def _silence(samples: int = 12_000) -> bytes:
return bytes(samples * 2) return bytes(samples * 2)
def _configure_fast_rolling(module, monkeypatch) -> None:
"""Make rolling thresholds deterministic without adding test sleeps."""
monkeypatch.setattr(module, "ROLLING_MIN_AUDIO_MS", 100)
monkeypatch.setattr(module, "ROLLING_INTERVAL_MS", 100)
monkeypatch.setattr(module, "ROLLING_MAX_RESULT_LAG_MS", 5_000)
def _wait_rolling_idle(session, timeout: float = 1.0) -> None:
"""Wait until an inference worker has applied or discarded its result."""
deadline = time.monotonic() + timeout
with session._lock:
while session._rolling_inflight and time.monotonic() < deadline:
session._lock.wait(deadline - time.monotonic())
assert not session._rolling_inflight
def test_websocket_handshake_matches_rfc_example(monkeypatch): def test_websocket_handshake_matches_rfc_example(monkeypatch):
module = _load_server(monkeypatch) module = _load_server(monkeypatch)
@ -158,6 +234,161 @@ def test_stream_speculation_is_reused_for_unchanged_commit(monkeypatch):
assert model.calls[0][1]["condition_on_previous_text"] is False assert model.calls[0][1]["condition_on_previous_text"] is False
def test_active_speech_emits_rate_limited_stable_rolling_partials(monkeypatch):
module = _load_server(monkeypatch)
_configure_fast_rolling(module, monkeypatch)
connection = _Connection()
model = _SequencedModel(["please open", "please open my calendar"])
session = module.StreamingTranscription(connection, model)
_start(session)
session.append(_speech())
first = connection.wait_for_revision(1)
assert first == {
"type": "partial",
"turn_id": "turn-10",
"transcript": "please open",
"stable_transcript": "",
"language": "en",
"speculative": True,
"rolling": True,
"revision": 1,
"epoch": 0,
"window_start_ms": 0,
}
# Audio progress is required in addition to wall-clock rate limiting.
session._rolling_last_started_at = 0.0
session.append(_speech())
second = connection.wait_for_revision(2)
assert second["transcript"] == "please open my calendar"
assert second["stable_transcript"] == "please open"
assert len(model.calls) == 2
def test_rolling_window_is_bounded_but_final_decode_uses_full_utterance(monkeypatch):
module = _load_server(monkeypatch)
_configure_fast_rolling(module, monkeypatch)
monkeypatch.setattr(module, "ROLLING_WINDOW_MS", 300)
connection = _Connection()
model = _SequencedModel(["partial", "complete request"])
session = module.StreamingTranscription(connection, model)
_start(session)
utterance = _speech(samples=16_000)
session.append(utterance)
partial = connection.wait_for_revision(1)
assert partial["window_start_ms"] == 700
assert model.frame_counts == [4_800]
session.commit()
assert connection.wait_for("final")["transcript"] == "complete request"
assert model.frame_counts == [4_800, 16_000]
def test_cancel_drops_inflight_rolling_result(monkeypatch):
module = _load_server(monkeypatch)
_configure_fast_rolling(module, monkeypatch)
connection = _Connection()
model = _BlockingModel()
session = module.StreamingTranscription(connection, model)
_start(session)
session.append(_speech())
assert model.entered.wait(1.0)
session.cancel()
model.release.set()
_wait_rolling_idle(session)
assert connection.messages == []
def test_resume_epoch_drops_old_rolling_result(monkeypatch):
module = _load_server(monkeypatch)
_configure_fast_rolling(module, monkeypatch)
connection = _Connection()
model = _BlockingModel()
session = module.StreamingTranscription(connection, model)
_start(session)
session.append(_speech())
assert model.entered.wait(1.0)
session.resume()
model.release.set()
_wait_rolling_idle(session)
assert connection.messages == []
assert session._epoch == 1
def test_busy_gpu_skips_partial_instead_of_queuing_work(monkeypatch):
module = _load_server(monkeypatch)
_configure_fast_rolling(module, monkeypatch)
connection = _Connection()
model = _Model()
session = module.StreamingTranscription(connection, model)
_start(session)
release = threading.Event()
entered = threading.Event()
def occupy_gpu():
module.INFERENCE_GATE.try_background(
lambda: (entered.set(), release.wait(2.0))
)
worker = threading.Thread(target=occupy_gpu)
worker.start()
assert entered.wait(1.0)
try:
session.append(_speech())
_wait_rolling_idle(session)
assert model.calls == []
assert connection.messages == []
finally:
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.
session._rolling_last_started_at = 0.0
session.append(_speech())
assert connection.wait_for_revision(1)["transcript"] == "hello"
assert len(model.calls) == 1
def test_waiting_final_prevents_new_rolling_work_from_stealing_gpu(monkeypatch):
module = _load_server(monkeypatch)
first_entered = threading.Event()
release_first = threading.Event()
final_ran = threading.Event()
first = threading.Thread(
target=lambda: module.INFERENCE_GATE.try_background(
lambda: (first_entered.set(), release_first.wait(2.0))
)
)
first.start()
assert first_entered.wait(1.0)
final = threading.Thread(
target=lambda: module.INFERENCE_GATE.run_final(final_ran.set)
)
final.start()
deadline = time.monotonic() + 1.0
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
release_first.set()
first.join(timeout=1.0)
final.join(timeout=1.0)
assert final_ran.is_set()
telemetry = module.INFERENCE_GATE.snapshot()
assert telemetry["background_skips"] == 1
assert telemetry["last_final_wait_ms"] >= 0
assert telemetry["last_final_decode_ms"] >= 0
def test_resume_invalidates_speculative_snapshot(monkeypatch): def test_resume_invalidates_speculative_snapshot(monkeypatch):
module = _load_server(monkeypatch) module = _load_server(monkeypatch)
connection = _Connection() connection = _Connection()
@ -238,4 +469,5 @@ def test_stream_health_contract_is_declared_in_source():
assert '"format": "pcm_s16le"' in source assert '"format": "pcm_s16le"' in source
assert '"server_vad": True' in source assert '"server_vad": True' in source
assert '"speculative": True' in source assert '"speculative": True' in source
assert '"inference": INFERENCE_GATE.snapshot()' in source
assert "MAX_STREAM_AUDIO_BYTES" in source assert "MAX_STREAM_AUDIO_BYTES" in source

View File

@ -0,0 +1,65 @@
"""Release contracts for localized, interruptible thinking voice cues."""
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
VOICE = ROOT / "dockerfiles/hermes-webui-atlas-voice.js"
def _source() -> str:
return VOICE.read_text(encoding="utf-8")
def test_fast_answers_suppress_cues_and_long_waits_are_bounded():
source = _source()
assert "const THINKING_CUE_FIRST_MS=1900" in source
assert "const THINKING_CUE_INTERVAL_MS=6500" in source
assert "cue.issued>=cue.pool.length" in source
assert "scheduleNextThinkingCue(cue,THINKING_CUE_FIRST_MS)" in source
assert "scheduleNextThinkingCue(cue,THINKING_CUE_INTERVAL_MS)" in source
assert "cancelThinkingCues();\n const turn=ensureSpeechTurn(token)" in source
def test_cues_are_localized_deterministic_and_non_repeating():
source = _source()
assert "{id:'thinking',text:\"I'm thinking.\"}" in source
assert "{id:'let_me_think',text:'Дайте подумать.'}" in source
assert "{id:'still_working',text:'Sigo pensando en eso.'}" in source
assert "offset:cuePoolOffset(turnId,pool.length)" in source
assert "cue.pool[(cue.offset+cue.issued)%cue.pool.length]" in source
assert "cue.issued+=1" in source
assert "const localized=normalizeSttLanguage(language)||'en'" in source
def test_cues_are_turn_owned_audio_only_and_abortable():
source = _source()
cue_region = source.split("async function issueThinkingCue(cue){", 1)[1].split(
"function scheduleThinkingCues", 1
)[0]
assert "cue.token===generation" in source
assert "state==='thinking'" in source
assert "cue.turnId+':thinking-cue:'" in cue_region
assert "request.cue_id=entry.id" in cue_region
assert "fetch(TTS_STREAM_URL" in cue_region
assert "fetch('/api/tts'" not in cue_region
assert "atlas-pcm-playback" in cue_region
assert "signal:controller.signal" in cue_region
assert "cue.controller.abort()" in source
assert "cancelThinkingCues();\n cancelSpeechTurn();" in source
assert "composer.value" not in cue_region
assert "window.send" not in cue_region
def test_thinking_barge_in_arms_quickly_but_guards_playback_echo():
source = _source()
assert "let speechArmAt=Date.now()+100" in source
assert "speechArmAt=now+(state==='thinking'?150:450)" in source
assert "playbackActive&&!playbackWasActive" in source
assert "const BARGE_TRIGGER_FRAMES=4" in source
assert "},50);" in source
assert "lookback:monitor.lookback" in source

View File

@ -5,7 +5,9 @@ from __future__ import annotations
import importlib.util import importlib.util
import io import io
import json import json
import struct
import sys import sys
import wave
from types import SimpleNamespace from types import SimpleNamespace
import pytest import pytest
@ -19,6 +21,7 @@ CLAUDE = "es_MX-claude-high"
def _load_tts_server(monkeypatch): def _load_tts_server(monkeypatch):
server_path = ROOT / "dockerfiles" / "hermes-jetson-tts-server.py" server_path = ROOT / "dockerfiles" / "hermes-jetson-tts-server.py"
monkeypatch.syspath_prepend(str(server_path.parent))
spec = importlib.util.spec_from_file_location("hermes_jetson_tts_server", server_path) spec = importlib.util.spec_from_file_location("hermes_jetson_tts_server", server_path)
assert spec and spec.loader assert spec and spec.loader
module = importlib.util.module_from_spec(spec) module = importlib.util.module_from_spec(spec)
@ -239,6 +242,72 @@ def test_turn_ids_are_safe_to_echo_in_response_headers(tts, value, expected):
assert tts.normalize_turn_id(value) == expected assert tts.normalize_turn_id(value) == expected
@pytest.mark.parametrize(
"text,expected",
[
("A short clause,", 90),
("A continued thought:", 110),
("Is this natural?", 170),
("This is natural.", 190),
("Thinking...", 220),
("First paragraph.\n\n", 280),
("A provisional phrase", 70),
('She said, "done."', 190),
],
)
def test_trailing_pause_budget_tracks_punctuation(tts, text, expected):
assert tts.trailing_pause_ms(text) == expected
def test_sentence_pause_targets_keep_paragraph_and_final_clause_cadence(tts):
assert tts.sentence_pause_targets("First.\n\nSecond? A tail,") == [280, 170, 90]
def _pcm(*runs: tuple[int, int]) -> bytes:
return b"".join(struct.pack(f"<{count}h", *([value] * count)) for value, count in runs)
def test_pcm_tail_trim_preserves_voice_and_keeps_period_pause(tts):
# A deliberately low-amplitude final phoneme remains above the very
# conservative silence threshold and must remain intact.
pcm = _pcm((1200, 50), (40, 50), (0, 600))
normalized = tts.trim_trailing_pcm_silence(pcm, sample_rate=1000, pause_ms=190)
assert bytes(normalized[:200]) == pcm[:200]
assert len(normalized) == (100 + 190) * 2
assert bytes(normalized[-190 * 2 :]) == b"\x00" * (190 * 2)
def test_pcm_tail_trim_leaves_short_and_all_silent_audio_unchanged(tts):
near_target = _pcm((900, 100), (0, 205))
all_silent = _pcm((0, 800))
assert bytes(tts.trim_trailing_pcm_silence(near_target, 1000, 190)) == near_target
assert bytes(tts.trim_trailing_pcm_silence(all_silent, 1000, 190)) == all_silent
def test_pcm_tail_trim_rejects_malformed_or_unknown_format_without_mutation(tts):
malformed = b"\x01\x02\x03"
assert bytes(tts.trim_trailing_pcm_silence(malformed, 22_050, 190)) == malformed
assert bytes(tts.trim_trailing_pcm_silence(b"\x01\x00", 0, 190)) == b"\x01\x00"
def test_compatibility_wav_uses_the_same_period_pause_budget(tts):
source = io.BytesIO()
with wave.open(source, "wb") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(1000)
wav_file.writeframes(_pcm((1000, 100), (0, 600)))
normalized = tts.normalize_wav_trailing_pause(source.getvalue(), "Finished.")
with wave.open(io.BytesIO(normalized), "rb") as wav_file:
assert wav_file.getparams().nchannels == 1
assert wav_file.getparams().sampwidth == 2
assert wav_file.getparams().framerate == 1000
assert wav_file.getnframes() == 100 + 190
def _decode_chunked_response(data: bytes) -> bytes: def _decode_chunked_response(data: bytes) -> bytes:
decoded = bytearray() decoded = bytearray()
cursor = 0 cursor = 0
@ -328,6 +397,154 @@ def test_stream_endpoint_returns_progressive_pcm_and_preserves_voice_policy(tts)
assert _decode_chunked_response(handler.wfile.getvalue()) == first_pcm + second_pcm assert _decode_chunked_response(handler.wfile.getvalue()) == first_pcm + second_pcm
def test_cached_thinking_cue_is_lock_free_and_ignores_client_text(tts):
cached = SimpleNamespace(
cue_id="thinking",
language="ru",
voice_name=IRINA,
sample_rate=22_050,
pcm=b"\x01\x00\x02\x00",
)
class _CachedHandler(tts.SpeechHandler):
def __init__(self):
payload = json.dumps(
{
"text": "attacker-controlled text is ignored",
"language": "ru",
"cue_id": "thinking",
"turn_id": "turn-9:thinking-cue:1",
}
).encode()
self.path = "/v1/audio/speech/stream"
self.headers = {"Content-Length": str(len(payload))}
self.rfile = io.BytesIO(payload)
self.wfile = io.BytesIO()
self.status = None
self.response_headers = {}
self.close_connection = False
self.server = SimpleNamespace(
voices={}, default_voice_name=AMY, cue_cache={("ru", "thinking"): cached}
)
def send_response(self, status, message=None):
self.status = status
def send_header(self, name, value):
self.response_headers[name] = value
def end_headers(self):
return None
handler = _CachedHandler()
handler.do_POST()
assert handler.status == 200
assert handler.wfile.getvalue() == cached.pcm
assert handler.response_headers["X-TTS-Voice"] == IRINA
assert handler.response_headers["X-TTS-Cue-ID"] == "thinking"
assert handler.response_headers["X-TTS-Cache"] == "HIT"
assert handler.response_headers["X-TTS-Turn-ID"] == "turn-9:thinking-cue:1"
def test_unknown_cached_thinking_cue_fails_closed(tts):
with pytest.raises(ValueError, match="unknown thinking cue"):
tts.resolve_cached_cue(
{"language": "en", "cue_id": "invented"},
{},
)
def test_thinking_cue_cache_precomputes_all_localized_voice_pairs(tts):
class _CueVoice:
config = SimpleNamespace(sample_rate=22_050)
def synthesize(self, text, syn_config):
assert text
assert syn_config.length_scale == 1.0
yield SimpleNamespace(
sample_rate=22_050,
sample_width=2,
sample_channels=1,
audio_int16_bytes=b"\x01\x00\x02\x00",
)
voices = {name: _CueVoice() for name in (AMY, IRINA, CLAUDE)}
cache = tts.build_cue_cache(
voices,
tts.SynthesisConfig,
lambda pcm, _rate, _pause: memoryview(pcm),
lambda _text: 190,
)
assert len(cache) == 12
assert {key[0] for key in cache} == {"en", "ru", "es"}
assert {cue.voice_name for (language, _), cue in cache.items() if language == "en"} == {AMY}
assert {cue.voice_name for (language, _), cue in cache.items() if language == "ru"} == {IRINA}
assert {cue.voice_name for (language, _), cue in cache.items() if language == "es"} == {CLAUDE}
assert all(cue.pcm and len(cue.pcm) % 2 == 0 for cue in cache.values())
def test_thinking_cue_cache_fails_startup_on_missing_or_malformed_audio(tts):
class _EmptyVoice:
config = SimpleNamespace(sample_rate=22_050)
def synthesize(self, _text, _syn_config):
return iter(())
voices = {name: _EmptyVoice() for name in (AMY, IRINA, CLAUDE)}
with pytest.raises(RuntimeError, match="returned no audio"):
tts.build_cue_cache(
voices,
tts.SynthesisConfig,
lambda pcm, _rate, _pause: memoryview(pcm),
lambda _text: 190,
)
def test_stream_endpoint_trims_each_sentence_with_its_own_pause_budget(tts):
sentence_pcm = _pcm((1000, 100), (0, 600))
class _StreamingVoice:
config = SimpleNamespace(sample_rate=1000)
def synthesize(self, text, syn_config):
assert text == "First. Final clause,"
for _ in range(2):
yield SimpleNamespace(
sample_rate=1000,
sample_width=2,
sample_channels=1,
audio_int16_bytes=sentence_pcm,
)
class _StreamingHandler(tts.SpeechHandler):
def __init__(self):
self.wfile = io.BytesIO()
self.response_headers = {}
self.close_connection = False
def send_response(self, status, message=None):
return None
def send_header(self, name, value):
self.response_headers[name] = value
def end_headers(self):
return None
handler = _StreamingHandler()
handler._stream_pcm("First. Final clause,", 1.0, AMY, _StreamingVoice(), "turn-2")
decoded = _decode_chunked_response(handler.wfile.getvalue())
# 100 ms of voiced audio plus 190 ms after the period and 90 ms after the
# final clause. The audio itself remains byte-for-byte identical.
assert len(decoded) == ((100 + 190) + (100 + 90)) * 2
assert decoded[: 100 * 2] == sentence_pcm[: 100 * 2]
second_start = (100 + 190) * 2
assert decoded[second_start : second_start + 100 * 2] == sentence_pcm[: 100 * 2]
def test_stream_disconnect_stops_before_synthesizing_another_sentence(tts): def test_stream_disconnect_stops_before_synthesizing_another_sentence(tts):
generated: list[int] = [] generated: list[int] = []

View File

@ -0,0 +1,120 @@
"""Contracts for low-latency, interruptible Hermes hands-free playback."""
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
VOICE = ROOT / "dockerfiles/hermes-webui-atlas-voice.js"
WORKLET = ROOT / "dockerfiles/hermes-webui-atlas-voice-worklet.js"
def test_barge_in_keeps_aec_capture_and_requires_sustained_speech():
source = VOICE.read_text(encoding="utf-8")
assert "echoCancellation:true" in source
assert "noiseSuppression:true" in source
assert "autoGainControl:true" in source
assert "const BARGE_DUCK_FRAMES=2" in source
assert "const BARGE_TRIGGER_FRAMES=4" in source
assert "let speechArmAt=Date.now()+100" in source
assert "speechArmAt=now+(state==='thinking'?150:450)" in source
assert "speechArmAt=now+100" in source
assert "playbackActive&&!playbackWasActive" in source
assert "const threshold=Math.max(0.05,(noiseFloor*3)+0.008)" in source
assert "setPlaybackDucked(true)" in source
assert "setPlaybackDucked(false)" in source
assert "settings.echoCancellation!==false" in source
assert "const automaticBargeAllowed=!playbackActive||monitor.aecUsable" in source
def test_barge_in_cancels_owned_turn_and_reuses_pcm_lookback():
source = VOICE.read_text(encoding="utf-8")
assert "api/chat/cancel?stream_id=" in source
assert "controller.abort();},1800" in source
assert "const deadline=Date.now()+10000" in source
assert "stopResponseObserver();" in source
assert "cancelSpeechTurn();" in source
assert "stopPlayback();" in source
assert "bargeCancelPromise=cancelActiveModelTurn()" in source
assert "lookback:monitor.lookback" in source
assert "heardSpeech:true" in source
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
def test_streamed_chunks_share_one_audio_timeline_until_final_drain():
source = VOICE.read_text(encoding="utf-8")
worklet = WORKLET.read_text(encoding="utf-8")
assert "if(session.node)" in source
assert "return session;" in source
assert "session.node.port.postMessage({type:'push'" in source
assert "session.node.port.postMessage({type:'end'})" in source
assert "await drainPcmPlayback(session,turn.token)" in source
assert "node.connect(gain)" in source
assert "gain.connect(context.destination)" in source
assert "this.ended&&!this.queue.length" in worklet
assert "this.port.postMessage({type:'drained'})" in worklet
def test_barge_cancel_settles_before_new_turn_baseline_and_observer():
source = VOICE.read_text(encoding="utf-8")
region = source.split("async function sendTranscript", 1)[1].split(
"function audioExtension", 1
)[0]
assert region.index("await settleBargeCancellation(token)") < region.index(
"rememberAssistantBaseline()"
)
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
def test_barge_handoff_never_uses_a_truncated_container_fallback():
source = VOICE.read_text(encoding="utf-8")
assert "monitor.handoff=true" in source
assert "if(!monitor.handoff) trimBargeLookback(monitor)" in source
assert "handoffMonitor:monitor" in source
assert "reusedCapture.lookback.forEach" in source
assert "requireStreamingLookback:true" in source
assert "allowContainerFallback===false" in source
assert "Please repeat your interruption" in source
def test_rolling_partials_are_feedback_only_and_wav_fallback_is_owned():
source = VOICE.read_text(encoding="utf-8")
partial = source.split("payload.type==='partial'", 1)[1].split(
"payload.type==='final'", 1
)[0]
fallback = source.split("async function fetchSpeech", 1)[1].split(
"function cuePoolOffset", 1
)[0]
assert "payload.stable_transcript" in partial
assert "label.textContent='Listening · '+preview" in partial
assert "composer.value" not in partial
assert "window.send" not in partial
assert "signal:controller.signal" in fallback
assert "token!==generation" in fallback
assert "session.turnId!==turnId" in fallback
assert "closePcmBeforeBlob" in source
def test_stt_client_queue_and_canonical_archive_are_strictly_bounded():
source = VOICE.read_text(encoding="utf-8")
assert "const STT_MAX_QUEUED_BYTES=1048576" in source
assert "const STT_MAX_ARCHIVE_BYTES=2880000" in source
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 "function clearQueue(){queue=[];queuedBytes=0;clearFlushTimer();}" in source
assert "takeFallbackBlob:function()" in source
assert "pcm16WavBlob(archive,16000)" in source
assert "if(pcmFallback)" in source

View File

@ -175,7 +175,8 @@ def test_release_candidate_streaming_keeps_one_ahead_and_safe_fallbacks():
assert "At most one later synthesis request exists" in script assert "At most one later synthesis request exists" in script
assert "const nextPrepared=next.then" in script assert "const nextPrepared=next.then" in script
assert "await playPrepared(asset,turn.token)" in script assert "await playPrepared(asset,turn.token)" in script
assert "playBlob(await fetchSpeech(asset.chunk,asset.language,asset.turnId),token)" in script assert "playBlob(await fetchSpeech(asset.chunk,asset.language,asset.turnId,token),token)" in script
assert "await closePcmBeforeBlob(playbackSession,token)" in script
assert "if(asset.started) throw error" in script assert "if(asset.started) throw error" in script
assert "playbackSession.controllers.forEach(function(controller){controller.abort();})" in script assert "playbackSession.controllers.forEach(function(controller){controller.abort();})" in script
assert "sendJson({type:'cancel',turn_id:turnId})" in script assert "sendJson({type:'cancel',turn_id:turnId})" in script

View File

@ -674,6 +674,53 @@ def test_adaptive_chunks_wait_for_sentence_then_change_size(voice_probe):
assert "".join(chunks).replace(" ", "") assert "".join(chunks).replace(" ", "")
def test_spoken_http_urls_are_skipped_without_mangling_prose(voice_probe):
result = voice_probe["spoken_urls_are_skipped_without_damaging_text"]
assert result["sentence"] == "Read this. Then continue."
assert result["wrapped"] == "Open. Next."
assert result["punctuated"] == "Try, or!"
assert result["domains"] == "Keep example.com and sub.example.org exactly as written."
assert result["prose"] == "No links here; keep this sentence exactly as written."
def test_url_elision_happens_before_every_localized_voice_route(voice_probe):
cases = voice_probe["spoken_urls_are_elided_before_all_voice_routes"]["results"]
assert [case["language"] for case in cases] == ["en", "ru", "es"]
for case in cases:
assert case["tts"]
spoken = " ".join(request["text"] for request in case["tts"])
assert "http://" not in spoken
assert "https://" not in spoken
assert "private.example" not in spoken
assert "example.com" in spoken
assert all(request["language"] == case["language"] for request in case["tts"])
def test_canonical_pcm_fallback_wav_is_runtime_valid(voice_probe):
result = voice_probe["canonical_pcm_fallback_builds_a_valid_wav"]
assert result == {
"type": "audio/wav",
"size": 50,
"riff": "RIFF",
"wave": "WAVE",
"format": 1,
"channels": 1,
"rate": 16000,
"bits": 16,
"data": 6,
"pcm": [0, 0, 255, 127, 0, 128],
}
def test_applied_aec_settings_are_runtime_fail_safe(voice_probe):
assert voice_probe["applied_aec_settings_are_fail_safe"] == {
"applied": True,
"rejected": False,
"unknown": True,
"unsupported": True,
}
def test_first_complete_sentence_reaches_tts_before_completion_callback(voice_probe): def test_first_complete_sentence_reaches_tts_before_completion_callback(voice_probe):
result = voice_probe["first_sentence_speaks_before_stream_completion"] result = voice_probe["first_sentence_speaks_before_stream_completion"]
assert result["beforeBoundary"] == 0 assert result["beforeBoundary"] == 0
@ -756,6 +803,34 @@ def test_streaming_tts_payload_is_narrow_and_turn_bound(patched_webui):
} }
def test_streaming_tts_payload_forwards_only_allowlisted_localized_cues(patched_webui):
routes = patched_webui.routes
assert routes._atlas_tts_stream_payload(
{
"text": "client text is ignored by the cue cache",
"language": "es",
"cue_id": "still_working",
"turn_id": "voice-turn-7:thinking-cue:2",
}
) == {
"model": "piper",
"input": "client text is ignored by the cue cache",
"speed": 1.0,
"language": "es",
"cue_id": "still_working",
"turn_id": "voice-turn-7:thinking-cue:2",
}
for hostile in ("invented", "../thinking", 7, None):
with pytest.raises(ValueError, match="invalid thinking cue"):
routes._atlas_tts_stream_payload(
{"text": "ignored", "language": "en", "cue_id": hostile}
)
with pytest.raises(ValueError, match="invalid thinking cue"):
routes._atlas_tts_stream_payload(
{"text": "ignored", "language": "fr", "cue_id": "thinking"}
)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# 6. Build-time enforcement and documented semantics # 6. Build-time enforcement and documented semantics
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------

View File

@ -0,0 +1,171 @@
"""Browser, proxy, and Flux delivery contracts for voice route preflight."""
from __future__ import annotations
import json
from types import SimpleNamespace
import yaml
from testing.tests.test_hermes_chat_support import ROOT, _documents
from testing.tests.test_hermes_voice_language_routing import patched_webui # noqa: F401
HERMES = ROOT / "services" / "hermes"
VOICE = ROOT / "dockerfiles" / "hermes-webui-atlas-voice.js"
def test_webui_proxy_enforces_origin_exact_echo_and_no_store(
patched_webui, # noqa: F811
monkeypatch,
):
routes = patched_webui.routes
request_payload = {
"turn_id": "ef" * 16 + "-4-1",
"revision": 2,
"transcript": "Explain the safest non disruptive option",
}
monkeypatch.setattr(routes, "read_body", lambda _handler: request_payload, raising=False)
monkeypatch.setattr(
routes,
"j",
lambda _handler, payload, status=200, extra_headers=None: {
"status": status,
"payload": payload,
"headers": extra_headers,
},
raising=False,
)
trusted = {"value": True}
monkeypatch.setattr(
routes,
"_check_same_origin_browser_request",
lambda _handler: trusted["value"],
raising=False,
)
monkeypatch.setattr(
routes,
"bad",
lambda _handler, message, status=400: {"status": status, "error": message},
raising=False,
)
observed = {}
class Upstream:
def read(self, size):
observed["size"] = size
return json.dumps(
{
"turn_id": request_payload["turn_id"],
"revision": 2,
"tier": "deep",
"target": "atlas/auto/deep",
"advisory": True,
"private": "must be removed",
}
).encode()
def __enter__(self):
return self
def __exit__(self, *args):
return False
def open_preflight(request, timeout):
observed["url"] = request.full_url
observed["body"] = json.loads(request.data)
observed["timeout"] = timeout
return Upstream()
monkeypatch.setattr(routes, "_atlas_voice_preflight_open", open_preflight)
result = routes._handle_atlas_voice_preflight(SimpleNamespace())
assert observed["url"] == routes.ATLAS_VOICE_PREFLIGHT_URL
assert observed["body"] == request_payload
assert observed["timeout"] == 1.0
assert result["headers"] == {"Cache-Control": "no-store"}
assert result["payload"] == {
"turn_id": request_payload["turn_id"],
"revision": 2,
"tier": "deep",
"target": "atlas/auto/deep",
"advisory": True,
}
trusted["value"] = False
assert routes._handle_atlas_voice_preflight(SimpleNamespace())["status"] == 403
def test_browser_debounces_stable_partials_and_never_forces_final_route():
source = VOICE.read_text(encoding="utf-8")
partial = source.split("payload.type==='partial'", 1)[1].split(
"payload.type==='final'", 1
)[0]
scheduler = source.split("function scheduleVoicePreflight", 1)[1].split(
"function cancelThinkingCues", 1
)[0]
send = source.split("async function sendTranscript", 1)[1].split(
"function audioExtension", 1
)[0]
assert "VOICE_PREFLIGHT_DEBOUNCE_MS=200" in source
assert "window.crypto.getRandomValues(bytes)" in source
assert "scheduleVoicePreflight(turnId,revision,stable)" in partial
assert "cancelVoicePreflight(turnId)" in source
assert "controller.abort()" in scheduler
assert "advisory.turn_id!==turnId" in scheduler
assert "advisory.revision!==revision" in scheduler
assert "label.textContent='Listening · '+preview+' · '+tier" in scheduler
assert "composer.value" not in scheduler
assert "window.send" not in scheduler
assert "advisory" not in send
assert "atlas/auto/" not in send
def test_flux_wires_the_sibling_runtime_rollout_service_and_narrow_policy():
kustomization = yaml.safe_load((HERMES / "kustomization.yaml").read_text())
generator = next(
item
for item in kustomization["configMapGenerator"]
if item["name"] == "hermes-coordinator"
)
assert "voice_route_preflight.py=scripts/voice_route_preflight.py" in generator["files"]
deployment = _documents(HERMES / "switchyard-deployment.yaml")[0]
template = deployment["spec"]["template"]
assert template["metadata"]["annotations"]["ai.bstein.dev/config-rev"].endswith(
"voice-route-preflight-v1"
)
classifier = next(
item for item in template["spec"]["containers"] if item["name"] == "classifier-broker"
)
env = {item["name"]: item["value"] for item in classifier["env"]}
assert env["HERMES_VOICE_PREFLIGHT_MODEL"] == "qwen2.5:14b-instruct-q4_0"
service = _documents(HERMES / "switchyard-service.yaml")[0]
ports = {item["name"]: item["port"] for item in service["spec"]["ports"]}
assert ports["voice-preflight"] == 9009
voice_port = next(
item for item in service["spec"]["ports"] if item["name"] == "voice-preflight"
)
assert voice_port["targetPort"] == "voice-preflight"
assert service["metadata"]["annotations"]["prometheus.io/port"] == "9009"
policies = {
item["metadata"]["name"]: item
for item in _documents(HERMES / "networkpolicy.yaml")
}
ingress = policies["hermes-switchyard-isolation"]["spec"]["ingress"]
preflight_rules = [
rule
for rule in ingress
if any(port["port"] == 9009 for port in rule.get("ports", []))
]
assert len(preflight_rules) == 2
assert preflight_rules[0]["from"][0]["podSelector"]["matchLabels"] == {
"app": "hermes-chat-tenant"
}
assert preflight_rules[1]["from"][0]["namespaceSelector"]["matchLabels"] == {
"kubernetes.io/metadata.name": "monitoring"
}
chat = (HERMES / "chat-statefulset.yaml").read_text()
assert "HERMES_WEBUI_VOICE_PREFLIGHT_URL" not in chat

View File

@ -13,6 +13,8 @@ def test_voice_pipeline_binds_component_source_digest_and_release() -> None:
pipeline = PIPELINE.read_text(encoding="utf-8") pipeline = PIPELINE.read_text(encoding="utf-8")
assert "IMAGE_COMPONENT must be stt or tts" in pipeline assert "IMAGE_COMPONENT must be stt or tts" in pipeline
assert "git merge-base --is-ancestor" in pipeline assert "git merge-base --is-ancestor" in pipeline
assert 'git checkout --detach "${EXPECTED_SOURCE_REVISION}"' in pipeline
assert 'test "${actual_revision}" = "${EXPECTED_SOURCE_REVISION}"' in pipeline
assert "git status --porcelain" in pipeline assert "git status --porcelain" in pipeline
assert "Dockerfile.hermes-jetson-${component}" in pipeline assert "Dockerfile.hermes-jetson-${component}" in pipeline
assert "--digest-file=" in pipeline assert "--digest-file=" in pipeline
@ -33,6 +35,8 @@ def test_voice_pipeline_binds_component_source_digest_and_release() -> None:
assert "--compressed-caching=false" in pipeline assert "--compressed-caching=false" in pipeline
assert "--snapshot-mode=redo" in pipeline assert "--snapshot-mode=redo" in pipeline
assert "privileged: true" not in pipeline assert "privileged: true" not in pipeline
assert "testing/tests/test_hermes_stt_streaming.py" in pipeline
assert "testing/tests/test_hermes_tts_language_routing.py" in pipeline
def test_jenkins_declares_one_token_guarded_voice_job() -> None: def test_jenkins_declares_one_token_guarded_voice_job() -> None:

View File

@ -0,0 +1,415 @@
"""Release gates for bounded, advisory-only voice route preflight."""
from __future__ import annotations
import importlib.util
import json
import sys
import threading
import time
from http.server import ThreadingHTTPServer
from types import SimpleNamespace
from urllib.error import HTTPError
from urllib.request import Request, urlopen
import pytest
from testing.tests.test_hermes_chat_support import ROOT
HERMES = ROOT / "services" / "hermes"
def _load_runtime(monkeypatch):
"""Load the sibling runtime modules without requiring httpx in image CI."""
monkeypatch.setitem(sys.modules, "httpx", SimpleNamespace())
preflight_path = HERMES / "scripts" / "voice_route_preflight.py"
spec = importlib.util.spec_from_file_location("voice_route_preflight", preflight_path)
assert spec and spec.loader
preflight = importlib.util.module_from_spec(spec)
spec.loader.exec_module(preflight)
monkeypatch.setitem(sys.modules, "voice_route_preflight", preflight)
broker_path = HERMES / "scripts" / "classifier_broker.py"
spec = importlib.util.spec_from_file_location("classifier_broker", broker_path)
assert spec and spec.loader
broker = importlib.util.module_from_spec(spec)
spec.loader.exec_module(broker)
return preflight, broker
@pytest.mark.parametrize(
"payload",
[
{},
{"turn_id": "bad/id", "revision": 1, "transcript": "long enough text"},
{"turn_id": "nonce-1", "revision": True, "transcript": "long enough text"},
{"turn_id": "nonce-1", "revision": 0, "transcript": "long enough text"},
{"turn_id": "nonce-1", "revision": 1, "transcript": None},
{"turn_id": "nonce-1", "revision": 1, "transcript": "short"},
{"turn_id": "nonce-1", "revision": 1, "transcript": "x" * 513},
],
)
def test_preflight_input_bounds_fail_closed(monkeypatch, payload):
preflight, _ = _load_runtime(monkeypatch)
with pytest.raises(ValueError):
preflight.validate_request(payload)
def test_preflight_is_same_14b_model_and_never_hosts_a_foreground_turn(monkeypatch):
preflight, _ = _load_runtime(monkeypatch)
payload = preflight.inference_payload("Please compare these two safe approaches")
assert preflight.MODEL == "qwen2.5:14b-instruct-q4_0"
assert payload["model"] == preflight.MODEL
assert payload["stream"] is True
assert payload["max_tokens"] <= 16
assert payload["temperature"] == 0
assert "tools" not in payload
assert "provider" not in payload
assert preflight.TIMEOUT_SECONDS <= 0.75
def test_preflight_is_once_per_random_browser_turn_and_authority_preempts(monkeypatch):
preflight, _ = _load_runtime(monkeypatch)
coordinator = preflight.Coordinator()
cancel = coordinator.begin("8f" * 16 + "-1-1")
assert cancel is not None
class Resource:
closed = False
def close(self):
self.closed = True
client = Resource()
response = Resource()
assert coordinator.register(cancel, client=client, response=response)
started = time.monotonic()
coordinator.begin_authoritative()
assert time.monotonic() - started < 0.1
assert cancel.is_set()
assert client.closed and response.closed
assert coordinator.begin("9e" * 16 + "-1-1") is None
coordinator.end_authoritative()
coordinator.end()
# A completed browser turn cannot spend another 750 ms on a later partial.
assert coordinator.begin("8f" * 16 + "-1-1") is None
def test_coordinator_prunes_seen_turns_and_tolerates_close_failures(monkeypatch):
preflight, _ = _load_runtime(monkeypatch)
coordinator = preflight.Coordinator()
now = time.monotonic()
coordinator.seen = {f"turn-{index}": now for index in range(300)}
coordinator._prune(now)
assert len(coordinator.seen) == preflight.MAX_SEEN_TURNS
class BrokenResource:
def close(self):
raise RuntimeError("close failed")
cancel = coordinator.begin("fresh-turn")
assert cancel is not None
assert coordinator.register(cancel, client=BrokenResource(), response=BrokenResource())
coordinator.begin_authoritative()
assert cancel.is_set()
coordinator.end_authoritative()
coordinator.end_authoritative()
coordinator.end()
unrelated = threading.Event()
assert coordinator.register(unrelated) is False
assert unrelated.is_set()
idle = preflight.Coordinator()
idle.begin_authoritative()
idle.end_authoritative()
def _stream_with_fake_httpx(
monkeypatch,
preflight,
coordinator,
lines,
*,
fail_build=False,
close_raises=False,
preempt_on_send=False,
):
"""Drive the streamed parser and cleanup branches without network I/O."""
class Response:
def raise_for_status(self):
return None
def iter_lines(self):
yield from lines
def close(self):
if close_raises:
raise RuntimeError("response close failed")
response = Response()
class Client:
def __init__(self, timeout):
self.timeout = timeout
def build_request(self, *args, **kwargs):
if fail_build:
raise RuntimeError("build failed")
return (args, kwargs)
def send(self, request, stream=False):
assert stream is True
if preempt_on_send:
coordinator.begin_authoritative()
return response
def close(self):
if close_raises:
raise RuntimeError("client close failed")
monkeypatch.setattr(
preflight,
"httpx",
SimpleNamespace(Timeout=lambda *args, **kwargs: (args, kwargs), Client=Client),
)
cancel = coordinator.begin("stream-" + str(time.monotonic_ns()))
assert cancel is not None
result = coordinator.stream("This provisional request is stable", cancel)
if preempt_on_send:
coordinator.end_authoritative()
coordinator.end()
return result
def test_stream_parser_accepts_only_valid_bounded_json(monkeypatch):
preflight, _ = _load_runtime(monkeypatch)
coordinator = preflight.Coordinator()
content = json.dumps({"tier": "deep"})
lines = [
"ignored",
"data: not-json",
'data: {"choices":[]}',
'data: {"choices":[{"delta":{"content":7}}]}',
"data: "
+ json.dumps({"choices": [{"delta": {"content": content}}]}),
"data: [DONE]",
]
assert _stream_with_fake_httpx(monkeypatch, preflight, coordinator, lines) == "deep"
assert 'outcome="success"} 1' in coordinator.metrics()
invalid = preflight.Coordinator()
lines = [
"data: "
+ json.dumps(
{"choices": [{"delta": {"content": json.dumps({"tier": "hosted"})}}]}
)
]
assert _stream_with_fake_httpx(monkeypatch, preflight, invalid, lines) == ""
assert 'outcome="failure"} 1' in invalid.metrics()
def test_stream_fails_closed_on_oversize_preemption_and_cleanup_errors(monkeypatch):
preflight, _ = _load_runtime(monkeypatch)
oversized = preflight.Coordinator()
line = "data: " + json.dumps({"choices": [{"delta": {"content": "x" * 129}}]})
assert _stream_with_fake_httpx(monkeypatch, preflight, oversized, [line]) == ""
assert 'outcome="cancelled"} 1' in oversized.metrics()
preempted = preflight.Coordinator()
assert (
_stream_with_fake_httpx(
monkeypatch, preflight, preempted, [], preempt_on_send=True
)
== ""
)
assert 'outcome="cancelled"} 1' in preempted.metrics()
failed = preflight.Coordinator()
assert (
_stream_with_fake_httpx(
monkeypatch,
preflight,
failed,
[],
fail_build=True,
close_raises=True,
)
== ""
)
assert 'outcome="failure"} 1' in failed.metrics()
def test_stream_rejects_a_cancelled_reservation_before_network(monkeypatch):
preflight, _ = _load_runtime(monkeypatch)
coordinator = preflight.Coordinator()
cancel = coordinator.begin("cancel-before-network")
assert cancel is not None
cancel.set()
closed = {"value": False}
class Client:
def __init__(self, timeout):
self.timeout = timeout
def close(self):
closed["value"] = True
monkeypatch.setattr(
preflight,
"httpx",
SimpleNamespace(Timeout=lambda *args, **kwargs: None, Client=Client),
)
assert coordinator.stream("This request will be cancelled", cancel) == ""
assert closed["value"] is True
assert 'outcome="cancelled"} 1' in coordinator.metrics()
coordinator.end()
def test_absolute_deadline_closes_a_stalled_stream(monkeypatch):
preflight, _ = _load_runtime(monkeypatch)
monkeypatch.setattr(preflight, "TIMEOUT_SECONDS", 0.05)
closed = threading.Event()
class Response:
def raise_for_status(self):
return None
def iter_lines(self):
closed.wait(1)
if False:
yield ""
def close(self):
closed.set()
response = Response()
class Client:
def __init__(self, timeout):
self.timeout = timeout
def build_request(self, *args, **kwargs):
return (args, kwargs)
def send(self, request, stream=False):
assert stream is True
return response
def close(self):
response.close()
monkeypatch.setattr(
preflight,
"httpx",
SimpleNamespace(Timeout=lambda *args, **kwargs: (args, kwargs), Client=Client),
)
coordinator = preflight.Coordinator()
cancel = coordinator.begin("ab" * 16 + "-2-1")
assert cancel is not None
started = time.monotonic()
assert coordinator.stream("This provisional request is stable", cancel) == ""
elapsed = time.monotonic() - started
coordinator.end()
assert elapsed < 0.25
assert 'outcome="timeout"} 1' in coordinator.metrics()
def test_preflight_is_warm_only_and_final_payload_semantics_never_change(monkeypatch):
preflight, broker = _load_runtime(monkeypatch)
coordinator = preflight.Coordinator()
authoritative = {
"model": "qwen2.5:14b-instruct-q4_0",
"messages": [
{"role": "system", "content": "normal routing contract"},
{"role": "user", "content": "Please compare the two deployments"},
],
"response_format": {"type": "json_object"},
}
before = broker.compact_payload(authoritative)
cancel = coordinator.begin("aa" * 16 + "-2-1")
assert cancel is not None
coordinator.end()
after = broker.compact_payload(authoritative)
assert before == after
assert not hasattr(coordinator, "cache")
assert not hasattr(coordinator, "remember")
assert not hasattr(coordinator, "take_hint")
assert not hasattr(preflight, "add_advisory_hint")
def _post(server, payload):
request = Request(
f"http://127.0.0.1:{server.server_port}/voice/route-preflight",
data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
with urlopen(request, timeout=2) as response:
return response.status, response.headers, response.read()
def test_handler_returns_exact_sanitized_binding_and_204_when_busy(monkeypatch):
preflight, broker = _load_runtime(monkeypatch)
coordinator = preflight.Coordinator()
monkeypatch.setattr(broker, "VOICE_PREFLIGHT", coordinator)
monkeypatch.setattr(coordinator, "stream", lambda transcript, cancel: "balanced")
server = ThreadingHTTPServer(("127.0.0.1", 0), broker.VoiceHandler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
payload = {
"turn_id": "cd" * 16 + "-3-1",
"revision": 4,
"transcript": "Private words that must never leave this boundary",
}
try:
status, headers, raw = _post(server, payload)
result = json.loads(raw)
assert status == 200
assert headers["Cache-Control"] == "no-store"
assert result == {
"turn_id": payload["turn_id"],
"revision": 4,
"tier": "balanced",
"target": "atlas/auto/balanced",
"advisory": True,
}
assert b"Private words" not in raw
status, headers, raw = _post(server, payload)
assert status == 204
assert headers["Cache-Control"] == "no-store"
assert raw == b""
finally:
server.shutdown()
server.server_close()
thread.join(timeout=2)
def test_public_and_authoritative_listeners_have_disjoint_route_surfaces(monkeypatch):
_, broker = _load_runtime(monkeypatch)
def post_status(handler, path):
server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
request = Request(
f"http://127.0.0.1:{server.server_port}{path}",
data=b"{}",
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urlopen(request, timeout=2) as response:
return response.status
except HTTPError as error:
return error.code
finally:
server.shutdown()
server.server_close()
thread.join(timeout=2)
assert post_status(broker.Handler, "/voice/route-preflight") == 404
assert post_status(broker.VoiceHandler, "/v1/chat/completions") == 404

View File

@ -239,11 +239,16 @@ def test_server_smoke_resolves_served_manifest_icons_and_checks_root() -> None:
"""The image gate validates HTTP responses, not its own source files.""" """The image gate validates HTTP responses, not its own source files."""
module = _load(SMOKE, "hermes_webui_smoke_contract") module = _load(SMOKE, "hermes_webui_smoke_contract")
manifest = json.loads(MANIFEST.read_text(encoding="utf-8")) manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
release_identity = "exp-v0.52.181-git-0123456789abcdef-build-13"
root = ( root = (
b'<link href="static/hermes-agent.ico">' b'<link href="static/hermes-agent.ico">'
b'<link href="static/hermes-agent-192.png">' b'<link href="static/hermes-agent-192.png">'
b'<link href="static/hermes-agent-512.png">' b'<link href="static/hermes-agent-512.png">'
b'<link href="static/hermes-brand.css?v=reviewed">' b'<link href="static/hermes-brand.css?v=reviewed">'
+ (
"<script>window.__HERMES_WEBUI_BUNDLE_VERSION__='"
f"{release_identity}';</script>"
).encode()
) )
calls = [] calls = []
@ -259,6 +264,18 @@ def test_server_smoke_resolves_served_manifest_icons_and_checks_root() -> None:
) )
if path == "/": if path == "/":
return _SmokeResponse(root, url) return _SmokeResponse(root, url)
if path == "/api/settings":
return _SmokeResponse(
json.dumps(
{"webui_bundle_version": release_identity}
).encode(),
url,
)
if path == "/sw.js":
return _SmokeResponse(
f"const CACHE_NAME='hermes-shell-{release_identity}';".encode(),
url,
)
if path in ( if path in (
"/pwa/static/hermes-agent-192.png", "/pwa/static/hermes-agent-192.png",
"/pwa/static/hermes-agent-512.png", "/pwa/static/hermes-agent-512.png",
@ -270,11 +287,51 @@ def test_server_smoke_resolves_served_manifest_icons_and_checks_root() -> None:
result = module.smoke("http://hermes.test/", opener=open_fixture) result = module.smoke("http://hermes.test/", opener=open_fixture)
assert result["manifest"] == "http://assets.hermes.test/pwa/manifest.json" assert result["manifest"] == "http://assets.hermes.test/pwa/manifest.json"
assert result["icon_requests"] == 3 assert result["icon_requests"] == 3
assert result["release_identity"] == release_identity
assert calls.count("http://assets.hermes.test/pwa/static/hermes-agent-192.png") == 2 assert calls.count("http://assets.hermes.test/pwa/static/hermes-agent-192.png") == 2
assert "http://assets.hermes.test/pwa/static/hermes-agent-512.png" in calls assert "http://assets.hermes.test/pwa/static/hermes-agent-512.png" in calls
assert calls.count("http://hermes.test/static/hermes-agent-192.png") == 1 assert calls.count("http://hermes.test/static/hermes-agent-192.png") == 1
def test_server_smoke_rejects_a_client_server_release_mismatch() -> None:
"""A build cannot publish the permanently stale banner regression."""
module = _load(SMOKE, "hermes_webui_smoke_release_mismatch")
manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
client = "exp-v0.52.181-git-0123456789abcdef-build-13"
root = (
b'<link href="static/hermes-agent.ico">'
b'<link href="static/hermes-agent-192.png">'
b'<link href="static/hermes-agent-512.png">'
b'<link href="static/hermes-brand.css?v=reviewed">'
+ (
"<script>window.__HERMES_WEBUI_BUNDLE_VERSION__='"
f"{client}';</script>"
).encode()
)
def open_mismatch(request, *, timeout):
assert timeout == 5
url = request.full_url
path = urlsplit(url).path
if path == "/manifest.json":
return _SmokeResponse(json.dumps(manifest).encode(), url)
if path == "/":
return _SmokeResponse(root, url)
if path == "/api/settings":
return _SmokeResponse(
json.dumps(
{"webui_bundle_version": client + "-different"}
).encode(),
url,
)
if path.endswith(".png"):
return _SmokeResponse(module.PNG_MAGIC + b"fixture", url)
raise AssertionError(f"unexpected smoke URL: {url}")
with pytest.raises(RuntimeError, match="identities do not match"):
module.smoke("http://hermes.test/", opener=open_mismatch)
def test_server_smoke_rejects_unbranded_served_manifest() -> None: def test_server_smoke_rejects_unbranded_served_manifest() -> None:
"""A successful health check cannot mask the upstream PWA identity.""" """A successful health check cannot mask the upstream PWA identity."""
module = _load(SMOKE, "hermes_webui_smoke_unbranded") module = _load(SMOKE, "hermes_webui_smoke_unbranded")
@ -330,7 +387,7 @@ def test_empty_state_uses_canonical_character_without_inline_staff(
def test_release_patch_gives_every_shell_url_an_immutable_build_token( def test_release_patch_gives_every_shell_url_an_immutable_build_token(
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
"""Atlas image releases never share a browser cache key.""" """Atlas releases share one server/client identity and browser cache key."""
target = _patched_fixture(tmp_path) target = _patched_fixture(tmp_path)
env = os.environ.copy() env = os.environ.copy()
env["HERMES_WEBUI_PATCH_ROOT"] = str(target) env["HERMES_WEBUI_PATCH_ROOT"] = str(target)
@ -345,11 +402,27 @@ def test_release_patch_gives_every_shell_url_an_immutable_build_token(
) )
index = (target / "static/index.html").read_text(encoding="utf-8") index = (target / "static/index.html").read_text(encoding="utf-8")
worker = (target / "static/sw.js").read_text(encoding="utf-8") worker = (target / "static/sw.js").read_text(encoding="utf-8")
routes = (target / "api/routes.py").read_text(encoding="utf-8")
panels = (target / "static/panels.js").read_text(encoding="utf-8")
token = "__WEBUI_VERSION__-git-0123456789abcdef-build-11" token = "__WEBUI_VERSION__-git-0123456789abcdef-build-11"
assert token in index assert token in index
assert token in worker assert token in worker
assert not re.search(r"__WEBUI_VERSION__(?!-git-0123456789abcdef-build-11)", index) assert not re.search(r"__WEBUI_VERSION__(?!-git-0123456789abcdef-build-11)", index)
assert not re.search(r"__WEBUI_VERSION__(?!-git-0123456789abcdef-build-11)", worker) assert not re.search(r"__WEBUI_VERSION__(?!-git-0123456789abcdef-build-11)", worker)
assert (
'settings["webui_bundle_version"] = '
'f"{WEBUI_VERSION}-git-0123456789abcdef-build-11"'
) in routes
assert (
"settings.webui_bundle_version||settings.webui_version"
) in panels
assert "if(client===server) return;" not in panels
assert "if(banner) banner.style.display='none';" in panels
assert "if(_isBannerVisible()) return;" not in panels
assert (
"if(_isBannerVisible()){ clearInterval(_pollTimer);"
not in panels
)
@pytest.mark.parametrize("release_id", ["", "../escape", "UPPERCASE", "a" * 129]) @pytest.mark.parametrize("release_id", ["", "../escape", "UPPERCASE", "a" * 129])
@ -406,6 +479,11 @@ def test_dockerfile_copies_and_verifies_every_tracked_brand_asset() -> None:
assert "python /tmp/hermes-webui-manifest-patch.py" in dockerfile assert "python /tmp/hermes-webui-manifest-patch.py" in dockerfile
assert "COPY dockerfiles/hermes-webui-release-patch.py" in dockerfile assert "COPY dockerfiles/hermes-webui-release-patch.py" in dockerfile
assert "python /tmp/hermes-webui-release-patch.py" in dockerfile assert "python /tmp/hermes-webui-release-patch.py" in dockerfile
assert 'settings["webui_bundle_version"]' in dockerfile
assert (
"settings.webui_bundle_version||settings.webui_version"
in dockerfile
)
assert ( assert (
"COPY dockerfiles/hermes-webui-manifest.json /tmp/hermes-webui-manifest.json" "COPY dockerfiles/hermes-webui-manifest.json /tmp/hermes-webui-manifest.json"
in dockerfile in dockerfile

View File

@ -91,13 +91,15 @@ def test_webui_job_is_independent_bounded_and_main_only() -> None:
assert "stringParam('CONFIRM_PUBLISH', ''" in block assert "stringParam('CONFIRM_PUBLISH', ''" in block
def test_pipeline_builds_latest_main_containing_reviewed_anchor() -> None: def test_pipeline_builds_exact_reviewed_anchor_from_main() -> None:
"""Publish is explicit, evidence-bound, and handed only to Flux.""" """Publish uses one reviewed source revision even when main later advances."""
source = PIPELINE.read_text(encoding="utf-8") source = PIPELINE.read_text(encoding="utf-8")
assert 'test "${CONFIRM_PUBLISH}" = "PUBLISH HERMES WEBUI"' in source assert 'test "${CONFIRM_PUBLISH}" = "PUBLISH HERMES WEBUI"' in source
assert 'test "${actual_revision}" = "$(git rev-parse origin/main)"' in source assert 'test "${main_revision}" = "$(git rev-parse origin/main)"' in source
assert "git fetch --no-tags origin main" not in source assert "git fetch --no-tags origin main" not in source
assert 'git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}"' in source assert 'git merge-base --is-ancestor "${EXPECTED_SOURCE_REVISION}"' in source
assert 'git checkout --detach "${EXPECTED_SOURCE_REVISION}"' in source
assert 'test "${actual_revision}" = "${EXPECTED_SOURCE_REVISION}"' in source
assert "dockerfiles/Dockerfile.hermes-webui" in source assert "dockerfiles/Dockerfile.hermes-webui" in source
assert "ci/scripts/hermes_webui_release.py" in source assert "ci/scripts/hermes_webui_release.py" in source
assert "registry.bstein.dev/bstein/hermes-webui" in source assert "registry.bstein.dev/bstein/hermes-webui" in source
@ -111,6 +113,15 @@ def test_pipeline_builds_latest_main_containing_reviewed_anchor() -> None:
assert "ci/scripts/hermes_oci_promote.py" in source assert "ci/scripts/hermes_oci_promote.py" in source
assert "test_hermes_webui_brand.py" in source assert "test_hermes_webui_brand.py" in source
assert "test_hermes_webui_release.py" in source assert "test_hermes_webui_release.py" in source
for voice_suite in (
"test_hermes_chat_quality.py",
"test_hermes_handsfree_stt.py",
"test_hermes_voice_full_duplex.py",
"test_hermes_thinking_voice_cues.py",
"test_hermes_voice_instrument.py",
"test_hermes_voice_language_routing.py",
):
assert voice_suite in source
for forbidden in ("kubectl ", "flux reconcile", "git push", "git commit"): for forbidden in ("kubectl ", "flux reconcile", "git push", "git commit"):
assert forbidden not in source assert forbidden not in source