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'<svg xmlns="http://www.w3.org/2000/svg"/>')
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/<id>/ 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.'