Merge branch 'main' of github.com:NousResearch/hermes-agent into feat/ink-refactor

tests/tools/test_approval_heartbeat.py

@@ -0,0 +1,200 @@
+"""Tests for the activity-heartbeat behavior of the blocking gateway approval wait.
+
+Regression test for false gateway inactivity timeouts firing while the agent
+is legitimately blocked waiting for a user to respond to a dangerous-command
+approval prompt.  Before the fix, ``entry.event.wait(timeout=...)`` blocked
+silently — no ``_touch_activity()`` calls — and the gateway's inactivity
+watchdog (``agent.gateway_timeout``, default 1800s) would kill the agent
+while the user was still choosing whether to approve.
+
+The fix polls the event in short slices and fires ``touch_activity_if_due``
+between slices, mirroring ``_wait_for_process`` in ``tools/environments/base.py``.
+"""
+
+import os
+import threading
+import time
+from unittest.mock import patch
+
+
+def _clear_approval_state():
+    """Reset all module-level approval state between tests."""
+    from tools import approval as mod
+    mod._gateway_queues.clear()
+    mod._gateway_notify_cbs.clear()
+    mod._session_approved.clear()
+    mod._permanent_approved.clear()
+    mod._pending.clear()
+
+
+class TestApprovalHeartbeat:
+    """The blocking gateway approval wait must fire activity heartbeats.
+
+    Without heartbeats, the gateway's inactivity watchdog kills the agent
+    thread while it's legitimately waiting for a slow user to respond to
+    an approval prompt (observed in real user logs: MRB, April 2026).
+    """
+
+    SESSION_KEY = "heartbeat-test-session"
+
+    def setup_method(self):
+        _clear_approval_state()
+        self._saved_env = {
+            k: os.environ.get(k)
+            for k in ("HERMES_GATEWAY_SESSION", "HERMES_YOLO_MODE",
+                      "HERMES_SESSION_KEY")
+        }
+        os.environ.pop("HERMES_YOLO_MODE", None)
+        os.environ["HERMES_GATEWAY_SESSION"] = "1"
+        # The blocking wait path reads the session key via contextvar OR
+        # os.environ fallback.  Contextvars don't propagate across threads
+        # by default, so env var is the portable way to drive this in tests.
+        os.environ["HERMES_SESSION_KEY"] = self.SESSION_KEY
+
+    def teardown_method(self):
+        for k, v in self._saved_env.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+        _clear_approval_state()
+
+    def test_heartbeat_fires_while_waiting_for_approval(self):
+        """touch_activity_if_due is called repeatedly during the wait."""
+        from tools.approval import (
+            check_all_command_guards,
+            register_gateway_notify,
+            resolve_gateway_approval,
+        )
+
+        register_gateway_notify(self.SESSION_KEY, lambda _payload: None)
+
+        # Use an Event to signal from _fake_touch back to the main thread
+        # so we can resolve as soon as the first heartbeat fires — avoids
+        # flakiness from fixed sleeps racing against thread startup.
+        first_heartbeat = threading.Event()
+        heartbeat_calls: list[str] = []
+
+        def _fake_touch(state, label):
+            # Bypass the 10s throttle so the heartbeat fires every loop
+            # iteration; we're measuring whether the call happens at all.
+            heartbeat_calls.append(label)
+            state["last_touch"] = 0.0
+            first_heartbeat.set()
+
+        result_holder: dict = {}
+
+        def _run_check():
+            try:
+                with patch(
+                    "tools.environments.base.touch_activity_if_due",
+                    side_effect=_fake_touch,
+                ):
+                    result_holder["result"] = check_all_command_guards(
+                        "rm -rf /tmp/nonexistent-heartbeat-target", "local"
+                    )
+            except Exception as exc:  # pragma: no cover
+                result_holder["exc"] = exc
+
+        thread = threading.Thread(target=_run_check, daemon=True)
+        thread.start()
+
+        # Wait for at least one heartbeat to fire — bounded at 10s to catch
+        # a genuinely hung worker thread without making a green run slow.
+        assert first_heartbeat.wait(timeout=10.0), (
+            "no heartbeat fired within 10s — the approval wait is blocking "
+            "without firing activity pings, which is the exact bug this "
+            "test exists to catch"
+        )
+
+        # Resolve the approval so the thread exits cleanly.
+        resolve_gateway_approval(self.SESSION_KEY, "once")
+        thread.join(timeout=5)
+
+        assert not thread.is_alive(), "approval wait did not exit after resolve"
+        assert "exc" not in result_holder, (
+            f"check_all_command_guards raised: {result_holder.get('exc')!r}"
+        )
+
+        # The fix: heartbeats fire while waiting.  Before the fix this list
+        # was empty because event.wait() blocked for the full timeout with
+        # no activity pings.
+        assert heartbeat_calls, "expected at least one heartbeat"
+        assert all(
+            call == "waiting for user approval" for call in heartbeat_calls
+        ), f"unexpected heartbeat labels: {set(heartbeat_calls)}"
+
+        # Sanity: the approval was resolved with "once" → command approved.
+        assert result_holder["result"]["approved"] is True
+
+    def test_wait_returns_immediately_on_user_response(self):
+        """Polling slices don't delay responsiveness — resolve is near-instant."""
+        from tools.approval import (
+            check_all_command_guards,
+            register_gateway_notify,
+            resolve_gateway_approval,
+        )
+
+        register_gateway_notify(self.SESSION_KEY, lambda _payload: None)
+
+        start_time = time.monotonic()
+        result_holder: dict = {}
+
+        def _run_check():
+            result_holder["result"] = check_all_command_guards(
+                "rm -rf /tmp/nonexistent-fast-target", "local"
+            )
+
+        thread = threading.Thread(target=_run_check, daemon=True)
+        thread.start()
+
+        # Resolve almost immediately — the wait loop should return within
+        # its current 1s poll slice.
+        time.sleep(0.1)
+        resolve_gateway_approval(self.SESSION_KEY, "once")
+        thread.join(timeout=5)
+        elapsed = time.monotonic() - start_time
+
+        assert not thread.is_alive()
+        assert result_holder["result"]["approved"] is True
+        # Generous bound to tolerate CI load; the previous single-wait
+        # impl returned in <10ms, the polling impl is bounded by the 1s
+        # slice length.
+        assert elapsed < 3.0, f"resolution took {elapsed:.2f}s, expected <3s"
+
+    def test_heartbeat_import_failure_does_not_break_wait(self):
+        """If tools.environments.base can't be imported, the wait still works."""
+        from tools.approval import (
+            check_all_command_guards,
+            register_gateway_notify,
+            resolve_gateway_approval,
+        )
+
+        register_gateway_notify(self.SESSION_KEY, lambda _payload: None)
+
+        result_holder: dict = {}
+        import builtins
+        real_import = builtins.__import__
+
+        def _fail_environments_base(name, *args, **kwargs):
+            if name == "tools.environments.base":
+                raise ImportError("simulated")
+            return real_import(name, *args, **kwargs)
+
+        def _run_check():
+            with patch.object(builtins, "__import__",
+                              side_effect=_fail_environments_base):
+                result_holder["result"] = check_all_command_guards(
+                    "rm -rf /tmp/nonexistent-import-fail-target", "local"
+                )
+
+        thread = threading.Thread(target=_run_check, daemon=True)
+        thread.start()
+
+        time.sleep(0.2)
+        resolve_gateway_approval(self.SESSION_KEY, "once")
+        thread.join(timeout=5)
+
+        assert not thread.is_alive()
+        # Even when heartbeat import fails, the approval flow completes.
+        assert result_holder["result"]["approved"] is True

tests/tools/test_checkpoint_manager.py

@@ -587,3 +587,112 @@ class TestSecurity:
         
         result = mgr.restore(str(work_dir), target_hash, file_path="subdir/test.txt")
         assert result["success"] is True
+
+
+# =========================================================================
+# GPG / global git config isolation
+# =========================================================================
+# Regression tests for the bug where users with ``commit.gpgsign = true``
+# in their global git config got a pinentry popup (or a failed commit)
+# every time the agent took a background snapshot.
+
+import os as _os
+
+
+class TestGpgAndGlobalConfigIsolation:
+    def test_git_env_isolates_global_and_system_config(self, tmp_path):
+        """_git_env must null out GIT_CONFIG_GLOBAL / GIT_CONFIG_SYSTEM so the
+        shadow repo does not inherit user-level gpgsign, hooks, aliases, etc."""
+        env = _git_env(tmp_path / "shadow", str(tmp_path))
+        assert env["GIT_CONFIG_GLOBAL"] == _os.devnull
+        assert env["GIT_CONFIG_SYSTEM"] == _os.devnull
+        assert env["GIT_CONFIG_NOSYSTEM"] == "1"
+
+    def test_init_sets_commit_gpgsign_false(self, work_dir, checkpoint_base, monkeypatch):
+        monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base)
+        shadow = _shadow_repo_path(str(work_dir))
+        _init_shadow_repo(shadow, str(work_dir))
+        # Inspect the shadow's own config directly — the settings must be
+        # written into the repo, not just inherited via env vars.
+        result = subprocess.run(
+            ["git", "config", "--file", str(shadow / "config"), "--get", "commit.gpgsign"],
+            capture_output=True, text=True,
+        )
+        assert result.stdout.strip() == "false"
+
+    def test_init_sets_tag_gpgsign_false(self, work_dir, checkpoint_base, monkeypatch):
+        monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base)
+        shadow = _shadow_repo_path(str(work_dir))
+        _init_shadow_repo(shadow, str(work_dir))
+        result = subprocess.run(
+            ["git", "config", "--file", str(shadow / "config"), "--get", "tag.gpgSign"],
+            capture_output=True, text=True,
+        )
+        assert result.stdout.strip() == "false"
+
+    def test_checkpoint_works_with_global_gpgsign_and_broken_gpg(
+        self, work_dir, checkpoint_base, monkeypatch, tmp_path
+    ):
+        """The real bug scenario: user has global commit.gpgsign=true but GPG
+        is broken or pinentry is unavailable.  Before the fix, every snapshot
+        either failed or spawned a pinentry window.  After the fix, snapshots
+        succeed without ever invoking GPG."""
+        monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base)
+
+        # Fake HOME with global gpgsign=true and a deliberately broken GPG
+        # binary.  If isolation fails, the commit will try to exec this
+        # nonexistent path and the checkpoint will fail.
+        fake_home = tmp_path / "fake_home"
+        fake_home.mkdir()
+        (fake_home / ".gitconfig").write_text(
+            "[user]\n    email = real@user.com\n    name = Real User\n"
+            "[commit]\n    gpgsign = true\n"
+            "[tag]\n    gpgSign = true\n"
+            "[gpg]\n    program = /nonexistent/fake-gpg-binary\n"
+        )
+        monkeypatch.setenv("HOME", str(fake_home))
+        monkeypatch.delenv("GPG_TTY", raising=False)
+        monkeypatch.delenv("DISPLAY", raising=False)  # block GUI pinentry
+
+        mgr = CheckpointManager(enabled=True)
+        assert mgr.ensure_checkpoint(str(work_dir), reason="with-global-gpgsign") is True
+        assert len(mgr.list_checkpoints(str(work_dir))) == 1
+
+    def test_checkpoint_works_on_prefix_shadow_without_local_gpgsign(
+        self, work_dir, checkpoint_base, monkeypatch, tmp_path
+    ):
+        """Users with shadow repos created before the fix will not have
+        commit.gpgsign=false in their shadow's own config.  The inline
+        ``--no-gpg-sign`` flag on the commit call must cover them."""
+        monkeypatch.setattr("tools.checkpoint_manager.CHECKPOINT_BASE", checkpoint_base)
+
+        # Simulate a pre-fix shadow repo: init without commit.gpgsign=false
+        # in its own config.  _init_shadow_repo now writes it, so we must
+        # manually remove it to mimic the pre-fix state.
+        shadow = _shadow_repo_path(str(work_dir))
+        _init_shadow_repo(shadow, str(work_dir))
+        subprocess.run(
+            ["git", "config", "--file", str(shadow / "config"),
+             "--unset", "commit.gpgsign"],
+            capture_output=True, text=True, check=False,
+        )
+        subprocess.run(
+            ["git", "config", "--file", str(shadow / "config"),
+             "--unset", "tag.gpgSign"],
+            capture_output=True, text=True, check=False,
+        )
+
+        # And simulate hostile global config
+        fake_home = tmp_path / "fake_home"
+        fake_home.mkdir()
+        (fake_home / ".gitconfig").write_text(
+            "[commit]\n    gpgsign = true\n"
+            "[gpg]\n    program = /nonexistent/fake-gpg-binary\n"
+        )
+        monkeypatch.setenv("HOME", str(fake_home))
+        monkeypatch.delenv("GPG_TTY", raising=False)
+        monkeypatch.delenv("DISPLAY", raising=False)
+
+        mgr = CheckpointManager(enabled=True)
+        assert mgr.ensure_checkpoint(str(work_dir), reason="prefix-shadow") is True
+        assert len(mgr.list_checkpoints(str(work_dir))) == 1

tests/tools/test_tts_gemini.py

@@ -0,0 +1,287 @@
+"""Tests for the Google Gemini TTS provider in tools/tts_tool.py."""
+
+import base64
+import struct
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def clean_env(monkeypatch):
+    for key in (
+        "GEMINI_API_KEY",
+        "GOOGLE_API_KEY",
+        "GEMINI_BASE_URL",
+        "HERMES_SESSION_PLATFORM",
+    ):
+        monkeypatch.delenv(key, raising=False)
+
+
+@pytest.fixture
+def fake_pcm_bytes():
+    # 0.1s of silence at 24kHz mono 16-bit = 4800 bytes
+    return b"\x00" * 4800
+
+
+@pytest.fixture
+def mock_gemini_response(fake_pcm_bytes):
+    """A successful Gemini generateContent response."""
+    resp = MagicMock()
+    resp.status_code = 200
+    resp.json.return_value = {
+        "candidates": [
+            {
+                "content": {
+                    "parts": [
+                        {
+                            "inlineData": {
+                                "mimeType": "audio/L16;codec=pcm;rate=24000",
+                                "data": base64.b64encode(fake_pcm_bytes).decode(),
+                            }
+                        }
+                    ]
+                }
+            }
+        ]
+    }
+    return resp
+
+
+class TestWrapPcmAsWav:
+    def test_riff_header_structure(self):
+        from tools.tts_tool import _wrap_pcm_as_wav
+
+        pcm = b"\x01\x02\x03\x04" * 10
+        wav = _wrap_pcm_as_wav(pcm, sample_rate=24000, channels=1, sample_width=2)
+
+        assert wav[:4] == b"RIFF"
+        assert wav[8:12] == b"WAVE"
+        assert wav[12:16] == b"fmt "
+        # Audio format (PCM=1)
+        assert struct.unpack("<H", wav[20:22])[0] == 1
+        # Channels
+        assert struct.unpack("<H", wav[22:24])[0] == 1
+        # Sample rate
+        assert struct.unpack("<I", wav[24:28])[0] == 24000
+        # Bits per sample
+        assert struct.unpack("<H", wav[34:36])[0] == 16
+        assert wav[36:40] == b"data"
+        assert wav[44:] == pcm
+
+    def test_header_size_is_44(self):
+        from tools.tts_tool import _wrap_pcm_as_wav
+
+        pcm = b"\xff" * 100
+        wav = _wrap_pcm_as_wav(pcm)
+        assert len(wav) == 44 + len(pcm)
+
+
+class TestGenerateGeminiTts:
+    def test_missing_api_key_raises_value_error(self, tmp_path):
+        from tools.tts_tool import _generate_gemini_tts
+
+        output_path = str(tmp_path / "test.wav")
+        with pytest.raises(ValueError, match="GEMINI_API_KEY"):
+            _generate_gemini_tts("Hello", output_path, {})
+
+    def test_google_api_key_fallback(self, tmp_path, monkeypatch, mock_gemini_response):
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GOOGLE_API_KEY", "from-google-env")
+        output_path = str(tmp_path / "test.wav")
+
+        with patch("requests.post", return_value=mock_gemini_response) as mock_post:
+            _generate_gemini_tts("Hi", output_path, {})
+
+        # Confirm it used the GOOGLE_API_KEY as the query parameter
+        _, kwargs = mock_post.call_args
+        assert kwargs["params"]["key"] == "from-google-env"
+
+    def test_wav_output_fast_path(self, tmp_path, monkeypatch, mock_gemini_response, fake_pcm_bytes):
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+        output_path = str(tmp_path / "test.wav")
+
+        with patch("requests.post", return_value=mock_gemini_response):
+            result = _generate_gemini_tts("Hi", output_path, {})
+
+        assert result == output_path
+        data = (tmp_path / "test.wav").read_bytes()
+        assert data[:4] == b"RIFF"
+        assert data[8:12] == b"WAVE"
+        # Audio payload should match the PCM we put in
+        assert data[44:] == fake_pcm_bytes
+
+    def test_default_voice_and_model(self, tmp_path, monkeypatch, mock_gemini_response):
+        from tools.tts_tool import (
+            DEFAULT_GEMINI_TTS_MODEL,
+            DEFAULT_GEMINI_TTS_VOICE,
+            _generate_gemini_tts,
+        )
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+
+        with patch("requests.post", return_value=mock_gemini_response) as mock_post:
+            _generate_gemini_tts("Hi", str(tmp_path / "test.wav"), {})
+
+        args, kwargs = mock_post.call_args
+        assert DEFAULT_GEMINI_TTS_MODEL in args[0]
+        payload = kwargs["json"]
+        voice = (
+            payload["generationConfig"]["speechConfig"]["voiceConfig"]
+            ["prebuiltVoiceConfig"]["voiceName"]
+        )
+        assert voice == DEFAULT_GEMINI_TTS_VOICE
+
+    def test_custom_voice(self, tmp_path, monkeypatch, mock_gemini_response):
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+        config = {"gemini": {"voice": "Puck"}}
+
+        with patch("requests.post", return_value=mock_gemini_response) as mock_post:
+            _generate_gemini_tts("Hi", str(tmp_path / "test.wav"), config)
+
+        payload = mock_post.call_args[1]["json"]
+        voice = (
+            payload["generationConfig"]["speechConfig"]["voiceConfig"]
+            ["prebuiltVoiceConfig"]["voiceName"]
+        )
+        assert voice == "Puck"
+
+    def test_custom_model(self, tmp_path, monkeypatch, mock_gemini_response):
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+        config = {"gemini": {"model": "gemini-2.5-pro-preview-tts"}}
+
+        with patch("requests.post", return_value=mock_gemini_response) as mock_post:
+            _generate_gemini_tts("Hi", str(tmp_path / "test.wav"), config)
+
+        endpoint = mock_post.call_args[0][0]
+        assert "gemini-2.5-pro-preview-tts" in endpoint
+
+    def test_response_modality_is_audio(self, tmp_path, monkeypatch, mock_gemini_response):
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+
+        with patch("requests.post", return_value=mock_gemini_response) as mock_post:
+            _generate_gemini_tts("Hi", str(tmp_path / "test.wav"), {})
+
+        payload = mock_post.call_args[1]["json"]
+        assert payload["generationConfig"]["responseModalities"] == ["AUDIO"]
+
+    def test_http_error_raises_runtime_error(self, tmp_path, monkeypatch):
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+        err_resp = MagicMock()
+        err_resp.status_code = 400
+        err_resp.json.return_value = {"error": {"message": "Invalid voice"}}
+
+        with patch("requests.post", return_value=err_resp):
+            with pytest.raises(RuntimeError, match="HTTP 400.*Invalid voice"):
+                _generate_gemini_tts("Hi", str(tmp_path / "test.wav"), {})
+
+    def test_empty_audio_raises(self, tmp_path, monkeypatch):
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.json.return_value = {
+            "candidates": [
+                {"content": {"parts": [{"inlineData": {"data": ""}}]}}
+            ]
+        }
+
+        with patch("requests.post", return_value=resp):
+            with pytest.raises(RuntimeError, match="empty audio"):
+                _generate_gemini_tts("Hi", str(tmp_path / "test.wav"), {})
+
+    def test_malformed_response_raises(self, tmp_path, monkeypatch):
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.json.return_value = {"candidates": []}  # no content
+
+        with patch("requests.post", return_value=resp):
+            with pytest.raises(RuntimeError, match="malformed"):
+                _generate_gemini_tts("Hi", str(tmp_path / "test.wav"), {})
+
+    def test_snake_case_inline_data_accepted(self, tmp_path, monkeypatch, fake_pcm_bytes):
+        """Some Gemini SDK versions return inline_data instead of inlineData."""
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.json.return_value = {
+            "candidates": [
+                {
+                    "content": {
+                        "parts": [
+                            {
+                                "inline_data": {
+                                    "data": base64.b64encode(fake_pcm_bytes).decode()
+                                }
+                            }
+                        ]
+                    }
+                }
+            ]
+        }
+
+        output_path = str(tmp_path / "test.wav")
+        with patch("requests.post", return_value=resp):
+            _generate_gemini_tts("Hi", output_path, {})
+
+        data = (tmp_path / "test.wav").read_bytes()
+        assert data[:4] == b"RIFF"
+
+    def test_custom_base_url_env(self, tmp_path, monkeypatch, mock_gemini_response):
+        from tools.tts_tool import _generate_gemini_tts
+
+        monkeypatch.setenv("GEMINI_API_KEY", "test-key")
+        monkeypatch.setenv("GEMINI_BASE_URL", "https://custom-gemini.example.com/v1beta")
+
+        with patch("requests.post", return_value=mock_gemini_response) as mock_post:
+            _generate_gemini_tts("Hi", str(tmp_path / "test.wav"), {})
+
+        assert mock_post.call_args[0][0].startswith("https://custom-gemini.example.com/v1beta/")
+
+
+class TestGeminiInCheckRequirements:
+    def test_gemini_api_key_satisfies_requirements(self, monkeypatch):
+        from tools.tts_tool import check_tts_requirements
+
+        # Strip everything else
+        for key in (
+            "ELEVENLABS_API_KEY",
+            "OPENAI_API_KEY",
+            "VOICE_TOOLS_OPENAI_KEY",
+            "MINIMAX_API_KEY",
+            "XAI_API_KEY",
+            "MISTRAL_API_KEY",
+            "GOOGLE_API_KEY",
+        ):
+            monkeypatch.delenv(key, raising=False)
+        monkeypatch.setenv("GEMINI_API_KEY", "k")
+
+        # Force edge_tts import to fail so we actually hit the gemini check
+        import builtins
+
+        real_import = builtins.__import__
+
+        def fake_import(name, *args, **kwargs):
+            if name == "edge_tts":
+                raise ImportError("simulated")
+            return real_import(name, *args, **kwargs)
+
+        with patch("builtins.__import__", side_effect=fake_import):
+            assert check_tts_requirements() is True