ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(tools): repoint tests to moved symbols; add voice_mode_transcript module

Browse Source
This commit is contained in:
Teknium
2026-09-02 13:55:41 -07:00
parent 9fe009f467
commit 8813eba345
16 changed files with 180 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/tools/test_delegate_control_actions.py
+2 -2
View File
@@ -199,7 +199,7 @@ def test_steer_closed_acceptance_is_refused():
def test_stop_interrupts_owned_child(monkeypatch):
import tools.delegate_tool as dt
import tools.delegate_tool_registry as dt
parent = _StubParent()
child = _StubChild(parent)
@@ -219,7 +219,7 @@ def test_stop_interrupts_owned_child(monkeypatch):
def test_stop_foreign_child_is_refused(monkeypatch):
import tools.delegate_tool as dt
import tools.delegate_tool_registry as dt
parent = _StubParent()
foreign = _StubChild(_StubParent())
tests/tools/test_image_source.py
+2 -2
View File
@@ -319,7 +319,7 @@ class TestSvgNormalization:
@pytest.mark.asyncio
async def test_svg_rasterized_when_converter_available(self, tmp_path, monkeypatch):
from tools import vision_tools as vt
from tools import vision_tools_image_prep as vt
isrc = _reload(monkeypatch, tmp_path / "hermes")
monkeypatch.setenv("TERMINAL_ENV", "local")
svg = tmp_path / "art.svg"
@@ -339,7 +339,7 @@ class TestSvgNormalization:
path.unlink()
def test_svg_actionable_error_when_no_converter(self, tmp_path, monkeypatch):
from tools import vision_tools as vt
from tools import vision_tools_image_prep as vt
_reload(monkeypatch, tmp_path / "hermes")
svg = tmp_path / "art.svg"
svg.write_bytes(b'<svg xmlns="http://www.w3.org/2000/svg"/>')
tests/tools/test_process_registry.py
+2 -3
View File
@@ -16,7 +16,6 @@ from tools.process_registry import (
ProcessSession,
FINISHED_TTL_SECONDS,
MAX_PROCESSES,
MAX_ACTIVE_PROCESS_AGE,
)
@@ -2508,11 +2507,11 @@ def _make_delegation_batch_evt(results):
def _patch_delegation_config(
monkeypatch, model="upstage/solar-pro-4", provider="openrouter", **over
):
import tools.process_registry as _pr
import tools.process_registry_notifications as _prn
cfg = {"model": model, "provider": provider}
cfg.update(over)
monkeypatch.setattr(_pr, "_delegation_config", lambda: cfg)
monkeypatch.setattr(_prn, "_delegation_config", lambda: cfg)
return cfg
tests/tools/test_session_search.py
-19
View File
@@ -19,7 +19,6 @@ from tools.session_search_tool import (
SESSION_SEARCH_SCHEMA,
_format_timestamp,
_is_compacted_message,
_is_compression_ended,
_resolve_to_parent,
_session_link,
session_search,
@@ -900,24 +899,6 @@ class TestRewindExclusion:
assert result_rewind["count"] == 0
class TestCompressionEndedHelper:
"""Unit tests for _is_compression_ended."""
def test_compression_ended_session(self, db):
db.create_session("s1", source="cli")
db.end_session("s1", "compression")
assert _is_compression_ended(db, "s1") is True
def test_delegation_child_not_ended(self, db):
"""A delegation child under a compression continuation does NOT have
end_reason='compression' itself."""
db.create_session("s_parent", source="cli")
db.end_session("s_parent", "compression")
db.create_session("s_continuation", source="cli", parent_session_id="s_parent")
db.create_session("s_delegate_child", source="cli", parent_session_id="s_continuation")
assert _is_compression_ended(db, "s_delegate_child") is False
class TestLegacyContinuationPlusDelegation:
"""Regression: a delegation child created under a compression continuation
must stay excluded — its content is still live to the parent agent.
tests/tools/test_skills_hub.py
-82
View File
@@ -1159,88 +1159,6 @@ class TestQuarantineBundleBinaryAssets:
) == "assets/data/sample.wav"
# ---------------------------------------------------------------------------
# GitHubSource._download_directory — tree API + fallback (#2940)
# ---------------------------------------------------------------------------
class TestDownloadDirectoryViaTree:
"""Tests for the Git Trees API path in _download_directory."""
def _source(self):
auth = MagicMock(spec=GitHubAuth)
auth.get_headers.return_value = {}
return GitHubSource(auth=auth)
@patch.object(GitHubSource, "_fetch_file_content")
@patch("tools.skills_hub.httpx.get")
def test_tree_api_downloads_subdirectories(self, mock_get, mock_fetch):
"""Tree API returns files from nested subdirectories."""
repo_resp = MagicMock(status_code=200, json=lambda: {"default_branch": "main"})
tree_resp = MagicMock(status_code=200, json=lambda: {
"truncated": False,
"tree": [
{"type": "blob", "path": "skills/my-skill/SKILL.md"},
{"type": "blob", "path": "skills/my-skill/scripts/run.py"},
{"type": "blob", "path": "skills/my-skill/references/api.md"},
{"type": "tree", "path": "skills/my-skill/scripts"},
{"type": "blob", "path": "other/file.txt"},
],
})
mock_get.side_effect = [repo_resp, tree_resp]
mock_fetch.side_effect = lambda repo, path: f"content-of-{path}"
src = self._source()
files = src._download_directory("owner/repo", "skills/my-skill")
assert "SKILL.md" in files
assert "scripts/run.py" in files
assert "references/api.md" in files
assert "other/file.txt" not in files # outside target path
assert len(files) == 3
@patch.object(GitHubSource, "_download_directory_recursive", return_value={"SKILL.md": "# ok"})
@patch("tools.skills_hub.httpx.get")
def test_falls_back_on_truncated_tree(self, mock_get, mock_fallback):
"""When tree is truncated, fall back to recursive Contents API."""
repo_resp = MagicMock(status_code=200, json=lambda: {"default_branch": "main"})
tree_resp = MagicMock(status_code=200, json=lambda: {"truncated": True, "tree": []})
mock_get.side_effect = [repo_resp, tree_resp]
src = self._source()
files = src._download_directory("owner/repo", "skills/my-skill")
assert files == {"SKILL.md": "# ok"}
mock_fallback.assert_called_once_with("owner/repo", "skills/my-skill")
class TestDownloadDirectoryRecursive:
"""Tests for the Contents API fallback path."""
def _source(self):
auth = MagicMock(spec=GitHubAuth)
auth.get_headers.return_value = {}
return GitHubSource(auth=auth)
@patch.object(GitHubSource, "_fetch_file_content")
@patch("tools.skills_hub.httpx.get")
def test_recursive_downloads_subdirectories(self, mock_get, mock_fetch):
"""Contents API recursion includes subdirectories."""
root_resp = MagicMock(status_code=200, json=lambda: [
{"name": "SKILL.md", "type": "file", "path": "skill/SKILL.md"},
{"name": "scripts", "type": "dir", "path": "skill/scripts"},
])
sub_resp = MagicMock(status_code=200, json=lambda: [
{"name": "run.py", "type": "file", "path": "skill/scripts/run.py"},
])
mock_get.side_effect = [root_resp, sub_resp]
mock_fetch.side_effect = lambda repo, path: f"content-of-{path}"
src = self._source()
files = src._download_directory_recursive("owner/repo", "skill")
assert "SKILL.md" in files
assert "scripts/run.py" in files
# ---------------------------------------------------------------------------
# Install-path safety (lock-file → uninstall rmtree boundary)
# ---------------------------------------------------------------------------
tests/tools/test_skills_sync.py
+5 -6
View File
@@ -765,23 +765,22 @@ class TestOptOutToggleAndRemove:
return bundled
def test_marker_toggle(self, tmp_path):
from tools.skills_sync import (
set_bundled_skills_opt_out, is_bundled_skills_opt_out,
)
from tools.skills_sync import set_bundled_skills_opt_out
home = tmp_path / "home"
home.mkdir()
marker = home / ".no-bundled-skills"
with patch("tools.skills_sync.HERMES_HOME", home):
assert is_bundled_skills_opt_out() is False
assert not marker.exists()
r = set_bundled_skills_opt_out(True)
assert r["ok"] and r["changed"]
assert is_bundled_skills_opt_out() is True
assert marker.exists()
# idempotent
r2 = set_bundled_skills_opt_out(True)
assert r2["ok"] and r2["changed"] is False
# opt back in
r3 = set_bundled_skills_opt_out(False)
assert r3["ok"] and r3["changed"]
assert is_bundled_skills_opt_out() is False
assert not marker.exists()
def test_remove_keeps_user_modified(self, tmp_path):
from tools.skills_sync import (
tests/tools/test_skills_sync_client.py
+1 -3
View File
@@ -327,7 +327,7 @@ class TestDevGate:
import hermes_cli.auth as auth_mod
monkeypatch.setattr(auth_mod, "resolve_nous_runtime_credentials",
lambda **kw: {"api_key": token, "base_url": "https://x"})
assert ssc.dev_gate_open() is False
assert ssc.resolve_identity()["nous_admin"] is False
def test_maybe_push_inert_when_gate_closed(self, monkeypatch):
token = _jwt({"sub": "u"})
@@ -882,7 +882,6 @@ class TestOrgIdentityGate:
lambda **kw: {"api_key": token, "base_url": "https://x"})
with pytest.raises(ssc.SyncInertError):
ssc.resolve_org_identity()
assert ssc.org_sync_available() is False
def test_org_identity_with_role(self, monkeypatch):
token = _jwt({"sub": "u", "org_id": "org-9", "org_role": "MEMBER"})
@@ -892,7 +891,6 @@ class TestOrgIdentityGate:
ident = ssc.resolve_org_identity()
assert ident["org_id"] == "org-9"
assert ident["org_role"] == "MEMBER"
assert ssc.org_sync_available() is True
def test_org_mirror_excluded_from_personal_sync(self, tmp_path, monkeypatch):
# A skill under _org/<id>/ must never be personal-sync eligible.
tests/tools/test_transcription_command_providers.py
-20
View File
@@ -34,8 +34,6 @@ from tools.transcription_tools import (
_get_command_stt_output_format,
_get_command_stt_timeout,
_get_named_stt_provider_config,
_has_any_command_stt_provider,
_iter_command_stt_providers,
_render_command_stt_template,
_resolve_command_stt_provider_config,
_transcribe_command_stt,
@@ -145,24 +143,6 @@ class TestSTTCommandHelpers:
assert DEFAULT_COMMAND_STT_OUTPUT_FORMAT == "txt"
def test_iter_command_providers_yields_only_command_type(self):
cfg = {
"providers": {
"cmd-one": {"type": "command", "command": "x"},
"no-cmd": {"type": "command"}, # no command field
"wrong-type": {"type": "http", "command": "x"},
"cmd-two": {"command": "y"}, # implicit type
},
}
names = {name for name, _ in _iter_command_stt_providers(cfg)}
assert names == {"cmd-one", "cmd-two"}
def test_has_any_command_provider_true_when_one_configured(self):
cfg = {"providers": {"custom": {"command": "x"}}}
assert _has_any_command_stt_provider(cfg) is True
# ---------------------------------------------------------------------------
# Template rendering
# ---------------------------------------------------------------------------
tests/tools/test_tts_long_form_chunking.py
+1 -1
View File
@@ -141,7 +141,7 @@ class TestBuildAudioDeliveryFiles:
out = str(tmp_path / "output.mp3")
profile = AudioDeliveryProfile(platform="default", max_file_bytes=10000)
with patch("tools.tts_tool._concat_audio_files") as mock_concat:
with patch("tools.tts_tool_delivery._concat_audio_files") as mock_concat:
mock_concat.return_value = out
# Copy the first file to output so the size check passes
Path(out).write_bytes(b"\x00" * 300)
tests/tools/test_tts_max_text_length.py
+2 -2
View File
@@ -79,7 +79,7 @@ class TestTextToSpeechToolChunking:
return output_path
monkeypatch.setattr("tools.tts_tool._generate_openai_tts", fake_openai)
monkeypatch.setattr("tools.tts_tool._concat_audio_files", fake_combine)
monkeypatch.setattr("tools.tts_tool_delivery._concat_audio_files", fake_combine)
monkeypatch.setattr("tools.tts_tool._load_tts_config",
lambda: {"provider": "openai"})
@@ -134,7 +134,7 @@ class TestTextToSpeechToolChunking:
return output_path
monkeypatch.setattr("tools.tts_tool._generate_openai_tts", fake_openai)
monkeypatch.setattr("tools.tts_tool._concat_audio_files", fake_combine)
monkeypatch.setattr("tools.tts_tool_delivery._concat_audio_files", fake_combine)
monkeypatch.setattr("tools.tts_tool._load_tts_config",
lambda: {"provider": "openai",
"openai": {"max_text_length": 100}})
tests/tools/test_tts_model_cache_lru.py
+1 -1
View File
@@ -24,7 +24,7 @@ def test_loads_on_miss_and_serves_from_cache_on_hit():
def test_hit_refreshes_recency_so_eviction_is_lru_not_fifo(monkeypatch):
monkeypatch.setattr(tts, "_TTS_MODEL_CACHE_MAX", 2)
monkeypatch.setattr("tools.tts_tool_local._TTS_MODEL_CACHE_MAX", 2)
cache: dict = {}
tts._tts_cache_get_or_load(cache, "a", lambda: "a")
tts._tts_cache_get_or_load(cache, "b", lambda: "b")
tests/tools/test_tts_output_timestamp.py
+1 -1
View File
@@ -17,7 +17,7 @@ class TestDefaultOutputTimestampResolution:
from tools import tts_tool
src = inspect.getsource(tts_tool.text_to_speech_tool)
src = inspect.getsource(tts_tool._resolve_output_base)
assert "%Y%m%d_%H%M%S_%f" in src, (
"default TTS output timestamp lost its microsecond component — "
"concurrent calls in the same second would collide again (#43911)"
tests/tools/test_tts_response_body_cap.py
+1 -1
View File
@@ -30,7 +30,7 @@ class StreamingResponse:
@pytest.fixture(autouse=True)
def small_tts_body_cap(monkeypatch):
monkeypatch.setattr(tts_tool, "TTS_RESPONSE_BODY_LIMIT_BYTES", 8)
monkeypatch.setattr("tools.tts_tool_providers.TTS_RESPONSE_BODY_LIMIT_BYTES", 8)
def test_xai_tts_rejects_oversized_audio_response(tmp_path, monkeypatch):
tests/tools/test_vision_tools.py
-37
View File
@@ -12,7 +12,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from tools.vision_tools import (
_validate_image_url,
_handle_vision_analyze,
_determine_mime_type,
_image_to_base64_data_url,
@@ -41,42 +40,6 @@ VALID_PNG = (
_RESOLVES = [(2, 1, 6, "", ("93.184.216.34", 0))]
# ---------------------------------------------------------------------------
# _validate_image_url — urlparse-based validation
# ---------------------------------------------------------------------------
class TestValidateImageUrl:
"""Tests for URL validation, including urlparse-based netloc check."""
def test_accepts_valid_http_and_https_urls(self):
with patch("tools.url_safety.socket.getaddrinfo", return_value=_RESOLVES):
assert _validate_image_url("https://example.com/image.jpg") is True
assert _validate_image_url("http://cdn.example.org/photo.png") is True
# CDN endpoints that redirect to images should still pass.
assert _validate_image_url("https://cdn.example.com/abcdef123") is True
assert _validate_image_url("https://img.example.com/pic?w=200&h=200") is True
assert _validate_image_url("http://example.com:8080/image.png") is True
assert _validate_image_url("https://example.com/") is True
def test_localhost_url_blocked_by_ssrf(self):
"""localhost URLs are blocked by SSRF protection."""
assert _validate_image_url("http://localhost:8080/image.png") is False
def test_rejects_malformed_and_non_string_inputs(self):
# http:// alone has no network location — urlparse catches this.
assert _validate_image_url("http://") is False
assert _validate_image_url("https://") is False
assert _validate_image_url("http:") is False
assert _validate_image_url("") is False
assert _validate_image_url(" ") is False
assert _validate_image_url(None) is False
assert _validate_image_url(12345) is False
assert _validate_image_url(True) is False
assert _validate_image_url(["https://example.com"]) is False
# ---------------------------------------------------------------------------
# _determine_mime_type
# ---------------------------------------------------------------------------
tests/tools/test_voice_mode.py
+1 -113
View File
@@ -1069,7 +1069,7 @@ class TestStreamLeakOnStartFailure:
# ============================================================================
# listen_for_speech — VAD barge-in monitor
# full_duplex_listen — VAD barge-in listener
# ============================================================================
class _FakeInputStream:
@@ -1092,118 +1092,6 @@ class _FakeInputStream:
return self._np.full((frames, 1), level, dtype=self._np.int16), False
class TestListenForSpeech:
"""listen_for_speech: calibration → sustained-speech trigger → barge-in."""
CALIB_BLOCKS = 14 # 400ms / 30ms
TRIP_BLOCKS = 10 # 300ms / 30ms
def _run(self, mock_sd, levels, should_stop=None, **kwargs):
np = pytest.importorskip("numpy")
stream = _FakeInputStream(np, levels)
mock_sd.InputStream.return_value = stream
from tools.voice_mode import listen_for_speech
stops = iter([False] * 200 + [True] * 10_000)
return listen_for_speech(should_stop or (lambda: next(stops)), **kwargs), stream
def test_sustained_speech_triggers(self, mock_sd):
levels = [0] * self.CALIB_BLOCKS + [5000] * 50
heard, _ = self._run(mock_sd, levels)
assert heard is True
def test_returns_false_when_audio_unavailable(self, monkeypatch):
monkeypatch.setattr("tools.voice_mode._import_audio", MagicMock(side_effect=OSError("no audio")))
from tools.voice_mode import listen_for_speech
assert listen_for_speech(lambda: False) is False
def test_quiet_then_loud_playback_does_not_trip(self, mock_sd):
"""TTS that starts quiet and gets louder must NOT trip barge-in.
This is the core regression: a one-shot calibration freezes the
floor from the quiet opening, then louder TTS exceeds the stale
floor and false-triggers. The rolling window keeps the floor
current so the louder passage is absorbed into the floor.
"""
levels = [100] * self.CALIB_BLOCKS + [200] * 30 + [500] * 30 + [1000] * 30
heard, _ = self._run(mock_sd, levels)
assert heard is False
def test_silence_calibration_does_not_false_trip_on_tts(self, mock_sd):
"""Calibration during an inter-sentence gap must NOT false-trip.
If the grace period ends during a pause between TTS sentences, the
calibration window samples near-silence. Without the min_floor clamp,
min_floor locks near zero, the trigger drops to 400 RMS (SILENCE_RMS_THRESHOLD
* 2), and the next TTS sentence at 800 RMS exceeds it — those blocks are
excluded from the rolling window (rms >= trigger), the floor freezes, and
after sustained_ms the VAD false-triggers and cuts playback mid-sentence.
With the clamp, min_floor stays at SILENCE_RMS_THRESHOLD * 2 = 400, the
trigger is max(400, 400 * 8.0) = 3200, and 800-RMS TTS stays below it and
feeds the rolling floor. No false trip.
"""
# calibration_ms=800 → CALIB_BLOCKS = 800/30 ≈ 26 blocks of silence
# Then TTS resumes at 800 RMS — must NOT trip (below 3200 trigger).
calib = 800 // 30
levels = [0] * calib + [800] * 100
heard, _ = self._run(
mock_sd, levels,
sustained_ms=1000,
calibration_ms=800,
)
assert heard is False
class TestListenForSpeechCapture:
"""capture=True: the barge monitor records the interruption with pre-roll,
so the utterance is complete from its first syllable — nothing is lost
between detection and a recorder restart."""
CALIB_BLOCKS = 14 # 400ms / 30ms
LOUD_BLOCKS = 30 # speech: trips after 10, keeps talking
BLOCK = 480 # 16000 * 0.03
def _run(self, mock_sd, monkeypatch, levels, should_stop=None, **kwargs):
np = pytest.importorskip("numpy")
stream = _FakeInputStream(np, levels)
mock_sd.InputStream.return_value = stream
written = {}
monkeypatch.setattr(
"tools.voice_mode.AudioRecorder._write_wav",
staticmethod(lambda audio: written.update(audio=audio) or "/tmp/barge.wav"),
)
from tools.voice_mode import listen_for_speech
stops = iter([False] * 200 + [True] * 10_000)
path = listen_for_speech(
should_stop or (lambda: next(stops)), capture=True, **kwargs
)
return path, written.get("audio"), stream
def test_captured_utterance_includes_speech_onset(self, mock_sd, monkeypatch):
"""Every loud block — including the ones BEFORE detection tripped —
must land in the WAV. That pre-roll is the whole point."""
triggered = []
levels = [0] * self.CALIB_BLOCKS + [5000] * self.LOUD_BLOCKS + [0] * 500
path, audio, _ = self._run(
mock_sd, monkeypatch, levels,
should_stop=lambda: False,
on_trigger=lambda: triggered.append(True),
)
assert path == "/tmp/barge.wav"
assert triggered == [True]
assert int((audio == 5000).sum()) == self.LOUD_BLOCKS * self.BLOCK
def test_no_trip_returns_none(self, mock_sd, monkeypatch):
triggered = []
path, audio, _ = self._run(
mock_sd, monkeypatch, [0] * 500,
on_trigger=lambda: triggered.append(True),
)
assert path is None
assert audio is None
assert triggered == []
class TestFullDuplexListen:
"""full_duplex_listen: one agent-turn listener spanning generation and
playback — pre-playback calibration, phase-aware trigger, grace window."""
tools/voice_mode_transcript.py
+161
View File
@@ -0,0 +1,161 @@
"""Pure-text helpers for voice mode: Whisper hallucination filter, voice-chat
stop phrases, and the TTS self-echo guard. No audio dependencies."""
import difflib
import re
from typing import Optional
def _voice_config() -> dict:
"""``voice`` section of config.yaml, or ``{}`` when missing, malformed,
or the config system can't be imported (broken config mid-install)."""
try:
from hermes_cli.config import load_config
voice_cfg = load_config().get("voice", {})
return voice_cfg if isinstance(voice_cfg, dict) else {}
except Exception:
return {}
# Whisper commonly hallucinates these phrases on silent/near-silent audio
# (matched with trailing '.'/'!' stripped, so the bare form suffices).
WHISPER_HALLUCINATIONS = {
"thank you", "thanks for watching", "subscribe to my channel", "like and subscribe",
"please subscribe", "thank you for watching", "bye", "you", "the end",
# Non-English hallucinations (common on silence)
"продолжение следует", "sous-titres", "sous-titres réalisés par la communauté d'amara.org",
"sottotitoli creati dalla comunità amara.org", "untertitel von stephanie geiges",
"amara.org", "www.mooji.org", "ご視聴ありがとうございました",
}
# Repetitive hallucinations (e.g. "Thank you. Thank you. Thank you.")
_HALLUCINATION_REPEAT_RE = re.compile(
r'^(?:thank you|thanks|bye|you|ok|okay|the end|\.|\s|,|!)+$',
flags=re.IGNORECASE,
)
def is_whisper_hallucination(transcript: str) -> bool:
"""Check if a transcript is a known Whisper hallucination on silence."""
cleaned = transcript.strip().lower()
if not cleaned:
return True
return (
cleaned.rstrip('.!') in WHISPER_HALLUCINATIONS
or bool(_HALLUCINATION_REPEAT_RE.match(cleaned))
)
DEFAULT_VOICE_STOP_PHRASES = ("stop",)
def _load_voice_stop_phrases() -> tuple:
"""Configured ``voice.stop_phrases`` (default ``("stop",)``); an empty tuple
disables the feature. Malformed config (dict, list of non-strings) falls
back to the default rather than crashing the voice loop."""
try:
raw = _voice_config().get("stop_phrases", DEFAULT_VOICE_STOP_PHRASES)
if isinstance(raw, str):
raw = [raw]
if isinstance(raw, (list, tuple)):
return tuple(
str(p).strip().lower() for p in raw
if isinstance(p, (str, int, float)) and str(p).strip()
)
except Exception:
pass
return DEFAULT_VOICE_STOP_PHRASES
def _configured_stop_phrases() -> tuple:
"""Resolve ``_load_voice_stop_phrases`` through ``tools.voice_mode`` so
``patch("tools.voice_mode._load_voice_stop_phrases")`` still takes effect."""
from tools import voice_mode as _vm
return _vm._load_voice_stop_phrases()
def is_voice_stop_phrase(transcript: str, stop_phrases: Optional[tuple] = None) -> bool:
"""True when *transcript* is EXACTLY a configured stop phrase.
Deliberately strict: the whole utterance — lowercased, surrounding
punctuation stripped — must equal a phrase, so "stop doing that and try
again" still reaches the agent. ``voice.stop_phrases: []`` disables.
"""
if not transcript:
return False
cleaned = transcript.strip().lower().strip(".,!?;: \t\n\"'")
if not cleaned:
return False
if stop_phrases is None:
stop_phrases = _configured_stop_phrases()
return cleaned in stop_phrases
# Similarity ratio (difflib.SequenceMatcher) above which a playback-phase barge
# transcript is treated as a self-capture of Hermes' own TTS: the full-duplex
# listener has no echo cancellation, so speaker bleed can trip the barge
# trigger and get transcribed near-verbatim (a TTS -> STT -> TTS loop).
DEFAULT_TTS_ECHO_SIMILARITY_THRESHOLD = 0.6
# Minimum normalized-transcript length before the sliding-window fallback
# runs. Below this a genuine one-word barge-in ("yes") landing verbatim inside
# a longer reply would score a trivial 1.0 and be misread as self-capture; a
# real self-capture spans pre-roll plus time-to-silence, so it is longer.
MIN_FRAGMENT_LENGTH_FOR_ECHO = 10
def _normalize_for_echo_compare(text: str) -> str:
return re.sub(r"\s+", " ", text).strip().lower()
def is_tts_echo(
transcript: str,
spoken_text: str,
threshold: float = DEFAULT_TTS_ECHO_SIMILARITY_THRESHOLD,
) -> bool:
"""True when *transcript* looks like a self-capture of *spoken_text*.
Character-level similarity (language-agnostic, no word tokenization): a
genuine user interjection is very unlikely to closely match Hermes' own
words, so a high ratio signals speaker-bleed self-capture (fail-closed
guard for the playback-phase listener, which has no echo cancellation).
The playback-phase capture is cut when the trigger fires and only spans
pre-roll plus time-to-silence, so for replies longer than a clause the
transcript is a short FRAGMENT of `spoken_text` and the whole-string
ratio dilutes toward 0. When it misses, a window sized to the transcript
slides across `spoken_text` (character-based, so it works without word
boundaries). Transcripts shorter than `MIN_FRAGMENT_LENGTH_FOR_ECHO` skip
this fallback: a short interjection trivially matches a short window.
"""
if not transcript or not spoken_text:
return False
a = _normalize_for_echo_compare(transcript)
b = _normalize_for_echo_compare(spoken_text)
if not a or not b:
return False
def _similar(x: str, y: str) -> bool:
return difflib.SequenceMatcher(None, x, y).ratio() >= threshold
if _similar(a, b):
return True
if len(a) < MIN_FRAGMENT_LENGTH_FOR_ECHO or len(a) >= len(b):
return False
return any(_similar(a, b[start : start + len(a)]) for start in range(0, len(b) - len(a) + 1))
def voice_stop_hint() -> str:
"""One-line 'Say "stop" to end the voice chat.' hint for voice-mode start.
Uses the first ``voice.stop_phrases`` entry so a custom phrase renders
correctly; returns "" when stop phrases are disabled so surfaces show no
hint. Every surface announcing voice-mode start (CLI, TUI, desktop) uses
this one owner instead of hardcoding the wording.
"""
phrases = _configured_stop_phrases()
if not phrases:
return ""
return f'Say "{phrases[0]}" to end the voice chat.'