From 8813eba345d99268a567934d735610ea71a4e7b4 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:55:41 -0700 Subject: [PATCH] refactor(tools): repoint tests to moved symbols; add voice_mode_transcript module --- tests/tools/test_delegate_control_actions.py | 4 +- tests/tools/test_image_source.py | 4 +- tests/tools/test_process_registry.py | 5 +- tests/tools/test_session_search.py | 19 --- tests/tools/test_skills_hub.py | 82 --------- tests/tools/test_skills_sync.py | 11 +- tests/tools/test_skills_sync_client.py | 4 +- .../test_transcription_command_providers.py | 20 --- tests/tools/test_tts_long_form_chunking.py | 2 +- tests/tools/test_tts_max_text_length.py | 4 +- tests/tools/test_tts_model_cache_lru.py | 2 +- tests/tools/test_tts_output_timestamp.py | 2 +- tests/tools/test_tts_response_body_cap.py | 2 +- tests/tools/test_vision_tools.py | 37 ---- tests/tools/test_voice_mode.py | 114 +------------ tools/voice_mode_transcript.py | 161 ++++++++++++++++++ 16 files changed, 180 insertions(+), 293 deletions(-) create mode 100644 tools/voice_mode_transcript.py diff --git a/tests/tools/test_delegate_control_actions.py b/tests/tools/test_delegate_control_actions.py index 286922f350..318529decc 100644 --- a/tests/tools/test_delegate_control_actions.py +++ b/tests/tools/test_delegate_control_actions.py @@ -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()) diff --git a/tests/tools/test_image_source.py b/tests/tools/test_image_source.py index f7e5f3d351..8a2f568e43 100644 --- a/tests/tools/test_image_source.py +++ b/tests/tools/test_image_source.py @@ -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'') diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index ae97defcfe..0d97a92ee0 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -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 diff --git a/tests/tools/test_session_search.py b/tests/tools/test_session_search.py index 635bcb91e4..de9f1300e2 100644 --- a/tests/tools/test_session_search.py +++ b/tests/tools/test_session_search.py @@ -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. diff --git a/tests/tools/test_skills_hub.py b/tests/tools/test_skills_hub.py index 610ea2b349..0f57c3b097 100644 --- a/tests/tools/test_skills_hub.py +++ b/tests/tools/test_skills_hub.py @@ -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) # --------------------------------------------------------------------------- diff --git a/tests/tools/test_skills_sync.py b/tests/tools/test_skills_sync.py index bb55469c4f..5a79faf5a4 100644 --- a/tests/tools/test_skills_sync.py +++ b/tests/tools/test_skills_sync.py @@ -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 ( diff --git a/tests/tools/test_skills_sync_client.py b/tests/tools/test_skills_sync_client.py index c967eef187..fd9b71fb01 100644 --- a/tests/tools/test_skills_sync_client.py +++ b/tests/tools/test_skills_sync_client.py @@ -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// must never be personal-sync eligible. diff --git a/tests/tools/test_transcription_command_providers.py b/tests/tools/test_transcription_command_providers.py index 5803a0f81b..2d1964c92d 100644 --- a/tests/tools/test_transcription_command_providers.py +++ b/tests/tools/test_transcription_command_providers.py @@ -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 # --------------------------------------------------------------------------- diff --git a/tests/tools/test_tts_long_form_chunking.py b/tests/tools/test_tts_long_form_chunking.py index 1231c4554a..9797c883ab 100644 --- a/tests/tools/test_tts_long_form_chunking.py +++ b/tests/tools/test_tts_long_form_chunking.py @@ -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) diff --git a/tests/tools/test_tts_max_text_length.py b/tests/tools/test_tts_max_text_length.py index d10093a7f1..f86d613315 100644 --- a/tests/tools/test_tts_max_text_length.py +++ b/tests/tools/test_tts_max_text_length.py @@ -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}}) diff --git a/tests/tools/test_tts_model_cache_lru.py b/tests/tools/test_tts_model_cache_lru.py index f5e413f254..76ffe6b653 100644 --- a/tests/tools/test_tts_model_cache_lru.py +++ b/tests/tools/test_tts_model_cache_lru.py @@ -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") diff --git a/tests/tools/test_tts_output_timestamp.py b/tests/tools/test_tts_output_timestamp.py index 99869ff73b..33e391b125 100644 --- a/tests/tools/test_tts_output_timestamp.py +++ b/tests/tools/test_tts_output_timestamp.py @@ -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)" diff --git a/tests/tools/test_tts_response_body_cap.py b/tests/tools/test_tts_response_body_cap.py index 2126ec10ae..5df78a140a 100644 --- a/tests/tools/test_tts_response_body_cap.py +++ b/tests/tools/test_tts_response_body_cap.py @@ -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): diff --git a/tests/tools/test_vision_tools.py b/tests/tools/test_vision_tools.py index d926f96cc6..883d0934bd 100644 --- a/tests/tools/test_vision_tools.py +++ b/tests/tools/test_vision_tools.py @@ -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 # --------------------------------------------------------------------------- diff --git a/tests/tools/test_voice_mode.py b/tests/tools/test_voice_mode.py index be9f24f6de..94fe7478d3 100644 --- a/tests/tools/test_voice_mode.py +++ b/tests/tools/test_voice_mode.py @@ -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.""" diff --git a/tools/voice_mode_transcript.py b/tools/voice_mode_transcript.py new file mode 100644 index 0000000000..b0e22b8dc3 --- /dev/null +++ b/tools/voice_mode_transcript.py @@ -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.'