Merge branch 'main' of github.com:NousResearch/hermes-agent into bb/gui

tests/agent/test_title_generator.py

@@ -136,6 +136,21 @@ class TestAutoTitleSession:
             auto_title_session(db, "sess-1", "hi", "hello")
             db.set_session_title.assert_called_once_with("sess-1", "New Title")
 
+    def test_invokes_title_callback_after_setting_title(self):
+        db = MagicMock()
+        db.get_session_title.return_value = None
+        seen = []
+        with patch("agent.title_generator.generate_title", return_value="Readable Session"):
+            auto_title_session(
+                db,
+                "sess-1",
+                "hello",
+                "hi there",
+                title_callback=seen.append,
+            )
+        db.set_session_title.assert_called_once_with("sess-1", "Readable Session")
+        assert seen == ["Readable Session"]
+
     def test_skips_if_generation_fails(self):
         db = MagicMock()
         db.get_session_title.return_value = None
@@ -182,7 +197,13 @@ class TestMaybeAutoTitle:
             import time
             time.sleep(0.3)
             mock_auto.assert_called_once_with(
-                db, "sess-1", "hello", "hi there", failure_callback=None, main_runtime=None
+                db,
+                "sess-1",
+                "hello",
+                "hi there",
+                failure_callback=None,
+                main_runtime=None,
+                title_callback=None,
             )
 
     def test_forwards_failure_callback_to_worker(self):
@@ -202,7 +223,13 @@ class TestMaybeAutoTitle:
             import time
             time.sleep(0.3)
             mock_auto.assert_called_once_with(
-                db, "sess-1", "hello", "hi there", failure_callback=_cb, main_runtime=None
+                db,
+                "sess-1",
+                "hello",
+                "hi there",
+                failure_callback=_cb,
+                main_runtime=None,
+                title_callback=None,
             )
 
     def test_skips_if_no_response(self):

tests/cron/test_cron_no_agent.py

@@ -0,0 +1,332 @@
+"""Tests for cronjob no_agent mode — script-driven jobs that skip the LLM.
+
+Covers:
+
+* ``create_job(no_agent=True)`` shape, validation, and serialization.
+* ``cronjob(action='create', no_agent=True)`` tool-level validation.
+* ``cronjob(action='update')`` flipping no_agent on/off.
+* ``scheduler.run_job`` short-circuit path: success/silent/failure.
+* Shell script support in ``_run_job_script`` (.sh runs via bash).
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+
+@pytest.fixture
+def hermes_env(tmp_path, monkeypatch):
+    """Isolate HERMES_HOME for each test so jobs/scripts don't leak."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    (home / "scripts").mkdir()
+    (home / "cron").mkdir()
+
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    # Reload modules that cache get_hermes_home() at import time.
+    import importlib
+    import hermes_constants
+    importlib.reload(hermes_constants)
+    import cron.jobs
+    importlib.reload(cron.jobs)
+    import cron.scheduler
+    importlib.reload(cron.scheduler)
+
+    return home
+
+
+# ---------------------------------------------------------------------------
+# create_job / update_job: data-layer semantics
+# ---------------------------------------------------------------------------
+
+
+def test_create_job_no_agent_requires_script(hermes_env):
+    from cron.jobs import create_job
+
+    with pytest.raises(ValueError, match="no_agent=True requires a script"):
+        create_job(prompt=None, schedule="every 5m", no_agent=True)
+
+
+def test_create_job_no_agent_stores_field(hermes_env):
+    from cron.jobs import create_job
+
+    script_path = hermes_env / "scripts" / "watchdog.sh"
+    script_path.write_text("#!/bin/bash\necho hi\n")
+
+    job = create_job(
+        prompt=None,
+        schedule="every 5m",
+        script="watchdog.sh",
+        no_agent=True,
+        deliver="local",
+    )
+    assert job["no_agent"] is True
+    assert job["script"] == "watchdog.sh"
+    # Prompt can be empty/None for no_agent jobs.
+    assert job["prompt"] in (None, "")
+
+
+def test_create_job_default_is_not_no_agent(hermes_env):
+    from cron.jobs import create_job
+
+    job = create_job(prompt="say hi", schedule="every 5m", deliver="local")
+    assert job.get("no_agent") is False
+
+
+def test_update_job_roundtrips_no_agent_flag(hermes_env):
+    from cron.jobs import create_job, update_job, get_job
+
+    script_path = hermes_env / "scripts" / "w.sh"
+    script_path.write_text("echo hi\n")
+    job = create_job(prompt=None, schedule="every 5m", script="w.sh", no_agent=True, deliver="local")
+
+    update_job(job["id"], {"no_agent": False})
+    reloaded = get_job(job["id"])
+    assert reloaded["no_agent"] is False
+
+    update_job(job["id"], {"no_agent": True})
+    reloaded = get_job(job["id"])
+    assert reloaded["no_agent"] is True
+
+
+# ---------------------------------------------------------------------------
+# cronjob tool: API-layer validation
+# ---------------------------------------------------------------------------
+
+
+def test_cronjob_tool_create_no_agent_without_script_errors(hermes_env):
+    from tools.cronjob_tools import cronjob
+
+    result = json.loads(
+        cronjob(action="create", schedule="every 5m", no_agent=True, deliver="local")
+    )
+    assert result.get("success") is False
+    assert "no_agent=True requires a script" in result.get("error", "")
+
+
+def test_cronjob_tool_create_no_agent_with_script_succeeds(hermes_env):
+    from tools.cronjob_tools import cronjob
+
+    script_path = hermes_env / "scripts" / "alert.sh"
+    script_path.write_text("#!/bin/bash\necho alert\n")
+
+    result = json.loads(
+        cronjob(
+            action="create",
+            schedule="every 5m",
+            script="alert.sh",
+            no_agent=True,
+            deliver="local",
+        )
+    )
+    assert result.get("success") is True
+    assert result["job"]["no_agent"] is True
+    assert result["job"]["script"] == "alert.sh"
+
+
+def test_cronjob_tool_update_toggles_no_agent(hermes_env):
+    from tools.cronjob_tools import cronjob
+
+    script_path = hermes_env / "scripts" / "w.sh"
+    script_path.write_text("echo hi\n")
+
+    created = json.loads(
+        cronjob(
+            action="create",
+            schedule="every 5m",
+            script="w.sh",
+            no_agent=True,
+            deliver="local",
+        )
+    )
+    job_id = created["job_id"]
+
+    off = json.loads(cronjob(action="update", job_id=job_id, no_agent=False, prompt="run"))
+    assert off["success"] is True
+    assert off["job"].get("no_agent") in (False, None)
+
+    on = json.loads(cronjob(action="update", job_id=job_id, no_agent=True))
+    assert on["success"] is True
+    assert on["job"]["no_agent"] is True
+
+
+def test_cronjob_tool_update_no_agent_without_script_errors(hermes_env):
+    """Flipping no_agent=True on a job that has no script must fail."""
+    from tools.cronjob_tools import cronjob
+
+    created = json.loads(
+        cronjob(action="create", schedule="every 5m", prompt="do a thing", deliver="local")
+    )
+    job_id = created["job_id"]
+
+    result = json.loads(cronjob(action="update", job_id=job_id, no_agent=True))
+    assert result.get("success") is False
+    assert "without a script" in result.get("error", "")
+
+
+def test_cronjob_tool_create_does_not_require_prompt_when_no_agent(hermes_env):
+    """The 'prompt or skill required' rule is relaxed for no_agent jobs."""
+    from tools.cronjob_tools import cronjob
+
+    script_path = hermes_env / "scripts" / "w.sh"
+    script_path.write_text("echo hi\n")
+
+    result = json.loads(
+        cronjob(
+            action="create",
+            schedule="every 5m",
+            script="w.sh",
+            no_agent=True,
+            deliver="local",
+        )
+    )
+    assert result.get("success") is True
+
+
+# ---------------------------------------------------------------------------
+# scheduler.run_job: short-circuit behavior
+# ---------------------------------------------------------------------------
+
+
+def test_run_job_no_agent_success_returns_script_stdout(hermes_env):
+    """Happy path: script exits 0 with output, delivered verbatim."""
+    from cron.jobs import create_job
+    from cron.scheduler import run_job
+
+    script_path = hermes_env / "scripts" / "alert.sh"
+    script_path.write_text("#!/bin/bash\necho 'RAM 92% on host'\n")
+
+    job = create_job(
+        prompt=None, schedule="every 5m", script="alert.sh", no_agent=True, deliver="local"
+    )
+    success, doc, final_response, error = run_job(job)
+    assert success is True
+    assert error is None
+    assert "RAM 92% on host" in final_response
+    assert "RAM 92% on host" in doc
+
+
+def test_run_job_no_agent_empty_output_is_silent(hermes_env):
+    """Empty stdout → SILENT_MARKER, which suppresses delivery downstream."""
+    from cron.jobs import create_job
+    from cron.scheduler import run_job, SILENT_MARKER
+
+    script_path = hermes_env / "scripts" / "quiet.sh"
+    script_path.write_text("#!/bin/bash\n# nothing to say\n")
+
+    job = create_job(
+        prompt=None, schedule="every 5m", script="quiet.sh", no_agent=True, deliver="local"
+    )
+    success, doc, final_response, error = run_job(job)
+    assert success is True
+    assert error is None
+    assert final_response == SILENT_MARKER
+
+
+def test_run_job_no_agent_wake_gate_is_silent(hermes_env):
+    """wakeAgent=false gate in stdout triggers a silent run."""
+    from cron.jobs import create_job
+    from cron.scheduler import run_job, SILENT_MARKER
+
+    script_path = hermes_env / "scripts" / "gated.sh"
+    script_path.write_text('#!/bin/bash\necho \'{"wakeAgent": false}\'\n')
+
+    job = create_job(
+        prompt=None, schedule="every 5m", script="gated.sh", no_agent=True, deliver="local"
+    )
+    success, doc, final_response, error = run_job(job)
+    assert success is True
+    assert final_response == SILENT_MARKER
+
+
+def test_run_job_no_agent_script_failure_delivers_error(hermes_env):
+    """Non-zero exit → success=False, error alert is the delivered message."""
+    from cron.jobs import create_job
+    from cron.scheduler import run_job
+
+    script_path = hermes_env / "scripts" / "broken.sh"
+    script_path.write_text("#!/bin/bash\necho oops >&2\nexit 3\n")
+
+    job = create_job(
+        prompt=None, schedule="every 5m", script="broken.sh", no_agent=True, deliver="local"
+    )
+    success, doc, final_response, error = run_job(job)
+    assert success is False
+    assert error is not None
+    assert "oops" in final_response or "exited with code 3" in final_response
+    assert "Cron watchdog" in final_response  # alert header
+
+
+def test_run_job_no_agent_never_invokes_aiagent(hermes_env):
+    """no_agent jobs must NOT import/construct the AIAgent."""
+    from cron.jobs import create_job
+
+    script_path = hermes_env / "scripts" / "alert.sh"
+    script_path.write_text("#!/bin/bash\necho alert\n")
+
+    job = create_job(
+        prompt=None, schedule="every 5m", script="alert.sh", no_agent=True, deliver="local"
+    )
+
+    with patch("run_agent.AIAgent") as ai_mock:
+        from cron.scheduler import run_job
+
+        run_job(job)
+
+    ai_mock.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# _run_job_script: shell-script support
+# ---------------------------------------------------------------------------
+
+
+def test_run_job_script_shell_script_runs_via_bash(hermes_env):
+    """.sh files should execute under /bin/bash even without a shebang line."""
+    from cron.scheduler import _run_job_script
+
+    script_path = hermes_env / "scripts" / "shelly.sh"
+    # No shebang — relies on the interpreter-by-extension rule.
+    script_path.write_text('echo "shell: $BASH_VERSION" | head -c 7\n')
+
+    ok, output = _run_job_script("shelly.sh")
+    assert ok is True
+    assert output.startswith("shell:")
+
+
+def test_run_job_script_bash_extension_also_runs_via_bash(hermes_env):
+    from cron.scheduler import _run_job_script
+
+    script_path = hermes_env / "scripts" / "thing.bash"
+    script_path.write_text('printf "via bash\\n"\n')
+
+    ok, output = _run_job_script("thing.bash")
+    assert ok is True
+    assert output == "via bash"
+
+
+def test_run_job_script_python_still_runs_via_python(hermes_env):
+    """Regression: .py files must keep running via sys.executable."""
+    from cron.scheduler import _run_job_script
+
+    script_path = hermes_env / "scripts" / "py.py"
+    script_path.write_text("import sys\nprint(f'python {sys.version_info.major}')\n")
+
+    ok, output = _run_job_script("py.py")
+    assert ok is True
+    assert output.startswith("python ")
+
+
+def test_run_job_script_path_traversal_still_blocked(hermes_env):
+    """Security regression: shell-script support must NOT loosen containment."""
+    from cron.scheduler import _run_job_script
+
+    # Absolute path outside the scripts dir should be rejected.
+    ok, output = _run_job_script("/etc/passwd")
+    assert ok is False
+    assert "Blocked" in output or "outside" in output

tests/cron/test_jobs.py

@@ -1,6 +1,7 @@
 """Tests for cron/jobs.py — schedule parsing, job CRUD, and due-job detection."""
 
 import json
+import threading
 import pytest
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -745,6 +746,100 @@ class TestEnabledToolsets:
         assert fetched["enabled_toolsets"] == ["web", "delegation"]
 
 
+class TestMarkJobRunConcurrency:
+    """Regression tests for concurrent parallel job state writes.
+
+    tick() dispatches multiple jobs to separate threads simultaneously.
+    Without _jobs_file_lock protecting the load→modify→save cycle in
+    mark_job_run(), concurrent writes can clobber each other's updates
+    (last-writer-wins), leaving some jobs with stale last_status / last_run_at.
+    """
+
+    def test_three_concurrent_mark_job_run_no_overwrites(self, tmp_cron_dir):
+        """Run mark_job_run() for 3 jobs in parallel threads; all must land correctly."""
+        # Create 3 distinct recurring jobs
+        job_a = create_job(prompt="Job A", schedule="every 1h")
+        job_b = create_job(prompt="Job B", schedule="every 1h")
+        job_c = create_job(prompt="Job C", schedule="every 1h")
+
+        errors: list = []
+
+        def run_mark(job_id: str, success: bool, error_msg=None):
+            try:
+                mark_job_run(job_id, success=success, error=error_msg)
+            except Exception as exc:  # pragma: no cover
+                errors.append(exc)
+
+        # Fire all three concurrently
+        threads = [
+            threading.Thread(target=run_mark, args=(job_a["id"], True)),
+            threading.Thread(target=run_mark, args=(job_b["id"], False, "timeout")),
+            threading.Thread(target=run_mark, args=(job_c["id"], True)),
+        ]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert not errors, f"Unexpected exceptions in worker threads: {errors}"
+
+        # Verify each job has the correct state — no overwrites
+        a = get_job(job_a["id"])
+        b = get_job(job_b["id"])
+        c = get_job(job_c["id"])
+
+        assert a is not None, "Job A was unexpectedly deleted"
+        assert b is not None, "Job B was unexpectedly deleted"
+        assert c is not None, "Job C was unexpectedly deleted"
+
+        assert a["last_status"] == "ok", f"Job A last_status wrong: {a['last_status']}"
+        assert a["last_run_at"] is not None, "Job A last_run_at not set"
+        assert a["repeat"]["completed"] == 1, f"Job A completed count wrong: {a['repeat']['completed']}"
+
+        assert b["last_status"] == "error", f"Job B last_status wrong: {b['last_status']}"
+        assert b["last_error"] == "timeout", f"Job B last_error wrong: {b['last_error']}"
+        assert b["last_run_at"] is not None, "Job B last_run_at not set"
+        assert b["repeat"]["completed"] == 1, f"Job B completed count wrong: {b['repeat']['completed']}"
+
+        assert c["last_status"] == "ok", f"Job C last_status wrong: {c['last_status']}"
+        assert c["last_run_at"] is not None, "Job C last_run_at not set"
+        assert c["repeat"]["completed"] == 1, f"Job C completed count wrong: {c['repeat']['completed']}"
+
+    def test_repeated_concurrent_runs_accumulate_completed_count(self, tmp_cron_dir):
+        """Stress test: 10 threads each call mark_job_run on a different job once.
+
+        The completed count for every job must be exactly 1 after all threads finish,
+        confirming no thread's write was silently dropped.
+        """
+        n = 10
+        jobs = [create_job(prompt=f"Stress job {i}", schedule="every 1h") for i in range(n)]
+        errors: list = []
+
+        def run_mark(job_id: str):
+            try:
+                mark_job_run(job_id, success=True)
+            except Exception as exc:  # pragma: no cover
+                errors.append(exc)
+
+        threads = [threading.Thread(target=run_mark, args=(j["id"],)) for j in jobs]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert not errors, f"Unexpected exceptions: {errors}"
+
+        for job in jobs:
+            updated = get_job(job["id"])
+            assert updated is not None, f"Job {job['id']} was deleted"
+            assert updated["last_status"] == "ok", (
+                f"Job {job['id']} has wrong last_status: {updated['last_status']}"
+            )
+            assert updated["repeat"]["completed"] == 1, (
+                f"Job {job['id']} completed count is {updated['repeat']['completed']}, expected 1"
+            )
+
+
 class TestSaveJobOutput:
     def test_creates_output_file(self, tmp_cron_dir):
         output_file = save_job_output("test123", "# Results\nEverything ok.")

tests/cron/test_scheduler.py

@@ -1307,6 +1307,103 @@ class TestRunJobConfigLogging:
             f"Expected 'failed to parse prefill messages' warning in logs, got: {[r.message for r in caplog.records]}"
 
 
+class TestRunJobConfigEnvVarExpansion:
+    """Verify that ${VAR} references in config.yaml are expanded when running cron jobs."""
+
+    _RUNTIME = {
+        "api_key": "test-key",
+        "base_url": "https://example.invalid/v1",
+        "provider": "openrouter",
+        "api_mode": "chat_completions",
+    }
+
+    def test_model_env_ref_in_config_yaml_is_expanded(self, tmp_path, monkeypatch):
+        """${VAR} in config.yaml model: is expanded using env after .env is loaded."""
+        (tmp_path / "config.yaml").write_text("model: ${_HERMES_TEST_CRON_MODEL}\n")
+        monkeypatch.setenv("_HERMES_TEST_CRON_MODEL", "gpt-4o-mini-cron-test")
+
+        job = {"id": "env-job", "name": "env test", "prompt": "hi"}
+        fake_db = MagicMock()
+
+        with patch("cron.scheduler._hermes_home", tmp_path), \
+             patch("cron.scheduler._resolve_origin", return_value=None), \
+             patch("dotenv.load_dotenv"), \
+             patch("hermes_state.SessionDB", return_value=fake_db), \
+             patch("hermes_cli.runtime_provider.resolve_runtime_provider",
+                   return_value=self._RUNTIME), \
+             patch("run_agent.AIAgent") as mock_agent_cls:
+            mock_agent = MagicMock()
+            mock_agent.run_conversation.return_value = {"final_response": "ok"}
+            mock_agent_cls.return_value = mock_agent
+            success, _, _, error = run_job(job)
+
+        assert success is True
+        assert error is None
+        kwargs = mock_agent_cls.call_args.kwargs
+        assert kwargs["model"] == "gpt-4o-mini-cron-test", (
+            f"Expected model='gpt-4o-mini-cron-test', got {kwargs['model']!r}. "
+            "config.yaml ${VAR} was not expanded in the cron execution path."
+        )
+
+    def test_fallback_model_env_ref_in_config_yaml_is_expanded(self, tmp_path, monkeypatch):
+        """${VAR} in config.yaml fallback_providers model: is expanded."""
+        (tmp_path / "config.yaml").write_text(
+            "fallback_providers:\n"
+            "  - provider: openrouter\n"
+            "    model: ${_HERMES_TEST_CRON_FALLBACK}\n"
+        )
+        monkeypatch.setenv("_HERMES_TEST_CRON_FALLBACK", "gpt-4o-fallback-test")
+
+        job = {"id": "fb-job", "name": "fallback test", "prompt": "hi"}
+        fake_db = MagicMock()
+
+        with patch("cron.scheduler._hermes_home", tmp_path), \
+             patch("cron.scheduler._resolve_origin", return_value=None), \
+             patch("dotenv.load_dotenv"), \
+             patch("hermes_state.SessionDB", return_value=fake_db), \
+             patch("hermes_cli.runtime_provider.resolve_runtime_provider",
+                   return_value=self._RUNTIME), \
+             patch("run_agent.AIAgent") as mock_agent_cls:
+            mock_agent = MagicMock()
+            mock_agent.run_conversation.return_value = {"final_response": "ok"}
+            mock_agent_cls.return_value = mock_agent
+            run_job(job)
+
+        kwargs = mock_agent_cls.call_args.kwargs
+        fb = kwargs.get("fallback_model") or []
+        fb_list = fb if isinstance(fb, list) else [fb]
+        expanded = [e.get("model") for e in fb_list if isinstance(e, dict)]
+        assert "gpt-4o-fallback-test" in expanded, (
+            f"Expected expanded fallback model in {expanded!r}. "
+            "config.yaml ${VAR} in fallback_providers was not expanded."
+        )
+
+    def test_unexpanded_ref_passthrough_when_var_unset(self, tmp_path, monkeypatch):
+        """When the env var is not set, the literal ${VAR} is kept verbatim (not crashed)."""
+        (tmp_path / "config.yaml").write_text("model: ${_HERMES_TEST_CRON_UNSET_VAR}\n")
+        monkeypatch.delenv("_HERMES_TEST_CRON_UNSET_VAR", raising=False)
+
+        job = {"id": "unset-job", "name": "unset var test", "prompt": "hi"}
+        fake_db = MagicMock()
+
+        with patch("cron.scheduler._hermes_home", tmp_path), \
+             patch("cron.scheduler._resolve_origin", return_value=None), \
+             patch("dotenv.load_dotenv"), \
+             patch("hermes_state.SessionDB", return_value=fake_db), \
+             patch("hermes_cli.runtime_provider.resolve_runtime_provider",
+                   return_value=self._RUNTIME), \
+             patch("run_agent.AIAgent") as mock_agent_cls:
+            mock_agent = MagicMock()
+            mock_agent.run_conversation.return_value = {"final_response": "ok"}
+            mock_agent_cls.return_value = mock_agent
+            success, _, _, error = run_job(job)
+
+        assert success is True
+        kwargs = mock_agent_cls.call_args.kwargs
+        # Unresolved refs are kept verbatim — _expand_env_vars contract
+        assert kwargs["model"] == "${_HERMES_TEST_CRON_UNSET_VAR}"
+
+
 class TestRunJobSkillBacked:
     def test_run_job_preserves_skill_env_passthrough_into_worker_thread(self, tmp_path):
         job = {

tests/gateway/test_discord_document_handling.py

@@ -9,6 +9,7 @@ import os
 import sys
 from datetime import datetime, timezone
 from types import SimpleNamespace
+from typing import Optional
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -111,7 +112,7 @@ def adapter(monkeypatch):
 def make_attachment(
     *,
     filename: str,
-    content_type: str,
+    content_type: Optional[str],
     size: int = 1024,
     url: str = "https://cdn.discordapp.com/attachments/fake/file",
 ) -> SimpleNamespace:

tests/gateway/test_email.py

@@ -425,6 +425,91 @@ class TestDispatchMessage(unittest.TestCase):
         self.assertEqual(event.source.user_name, "John Doe")
         self.assertEqual(event.source.chat_type, "dm")
 
+    def test_non_allowlisted_sender_dropped(self):
+        """Senders not in EMAIL_ALLOWED_USERS should be dropped before dispatch."""
+        import asyncio
+        with patch.dict(os.environ, {
+            "EMAIL_ALLOWED_USERS": "hermes@test.com,admin@test.com",
+        }):
+            adapter = self._make_adapter()
+            adapter._message_handler = MagicMock()
+
+            msg_data = {
+                "uid": b"99",
+                "sender_addr": "outsider@evil.com",
+                "sender_name": "Spammer",
+                "subject": "Buy now!!!",
+                "message_id": "<spam@evil.com>",
+                "in_reply_to": "",
+                "body": "Cheap meds",
+                "attachments": [],
+                "date": "",
+            }
+
+            asyncio.run(adapter._dispatch_message(msg_data))
+            # Handler should NOT be called for non-allowlisted sender
+            adapter._message_handler.assert_not_called()
+            # Thread context should NOT be created
+            self.assertNotIn("outsider@evil.com", adapter._thread_context)
+
+    def test_allowlisted_sender_proceeds(self):
+        """Senders in EMAIL_ALLOWED_USERS should proceed to dispatch normally."""
+        import asyncio
+        with patch.dict(os.environ, {
+            "EMAIL_ALLOWED_USERS": "hermes@test.com,admin@test.com",
+        }):
+            adapter = self._make_adapter()
+            captured_events = []
+
+            async def mock_handler(event):
+                captured_events.append(event)
+                return None
+
+            adapter._message_handler = mock_handler
+
+            msg_data = {
+                "uid": b"100",
+                "sender_addr": "admin@test.com",
+                "sender_name": "Admin",
+                "subject": "Important",
+                "message_id": "<msg@test.com>",
+                "in_reply_to": "",
+                "body": "Hello",
+                "attachments": [],
+                "date": "",
+            }
+
+            asyncio.run(adapter._dispatch_message(msg_data))
+            self.assertEqual(len(captured_events), 1)
+            self.assertEqual(captured_events[0].source.chat_id, "admin@test.com")
+
+    def test_empty_allowlist_allows_all(self):
+        """When EMAIL_ALLOWED_USERS is not set, all senders should proceed."""
+        import asyncio
+        with patch.dict(os.environ, {}, clear=False):
+            # Ensure EMAIL_ALLOWED_USERS is not in the env
+            if "EMAIL_ALLOWED_USERS" in os.environ:
+                del os.environ["EMAIL_ALLOWED_USERS"]
+
+            adapter = self._make_adapter()
+            adapter._message_handler = MagicMock()
+
+            msg_data = {
+                "uid": b"101",
+                "sender_addr": "anyone@test.com",
+                "sender_name": "Anyone",
+                "subject": "Hey",
+                "message_id": "<any@test.com>",
+                "in_reply_to": "",
+                "body": "Hi",
+                "attachments": [],
+                "date": "",
+            }
+
+            asyncio.run(adapter._dispatch_message(msg_data))
+            # Handler should be called when no allowlist is configured
+            adapter._message_handler.assert_called()
+
 
 class TestThreadContext(unittest.TestCase):
     """Test email reply threading logic."""

tests/gateway/test_stale_code_self_check.py

@@ -3,25 +3,34 @@
 A gateway that survives ``hermes update`` keeps pre-update modules cached
 in ``sys.modules``.  Later imports of names added post-update (e.g.
 ``cfg_get`` from PR #17304) raise ImportError against the stale module
-object.  The self-check in ``GatewayRunner._detect_stale_code()`` detects
-this by comparing boot-time sentinel-file mtimes against current ones,
-and ``_trigger_stale_code_restart()`` triggers a graceful restart.
+object.
+
+The self-check compares the git HEAD SHA at boot to the current SHA on
+disk.  ``hermes update`` always moves HEAD forward via ``git pull``;
+agent-driven file edits (Hermes editing ``run_agent.py`` / ``gateway/run.py``
+during a self-dev session) never move HEAD — so the SHA signal is free of
+the false-positive class that the earlier mtime-based check suffered from.
 """
 
 import os
 import time
 from pathlib import Path
-from unittest.mock import MagicMock, patch
 
 import pytest
 
 from gateway.run import (
     GatewayRunner,
     _compute_repo_mtime,
+    _read_git_head_sha,
     _STALE_CODE_SENTINELS,
+    _GIT_SHA_CACHE_TTL_SECS,
 )
 
 
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
 def _make_tmp_repo(tmp_path: Path) -> Path:
     """Create a fake repo with all stale-code sentinel files."""
     for rel in _STALE_CODE_SENTINELS:
@@ -31,109 +40,303 @@ def _make_tmp_repo(tmp_path: Path) -> Path:
     return tmp_path
 
 
-def _make_runner(repo_root: Path, *, boot_mtime: float, boot_wall: float):
+def _make_git_repo(tmp_path: Path, sha: str = "a" * 40, branch: str = "main") -> Path:
+    """Stamp a minimal .git directory so _read_git_head_sha can resolve a SHA.
+
+    We don't run real git — just lay down the files the reader walks
+    (.git/HEAD pointing at refs/heads/<branch>, refs/heads/<branch>
+    containing the SHA).
+    """
+    git_dir = tmp_path / ".git"
+    git_dir.mkdir(parents=True, exist_ok=True)
+    (git_dir / "HEAD").write_text(f"ref: refs/heads/{branch}\n")
+    refs_dir = git_dir / "refs" / "heads"
+    refs_dir.mkdir(parents=True, exist_ok=True)
+    (refs_dir / branch).write_text(f"{sha}\n")
+    return tmp_path
+
+
+def _set_head_sha(repo_root: Path, sha: str, branch: str = "main") -> None:
+    """Rewrite the current branch ref to a new SHA (simulates git pull)."""
+    (repo_root / ".git" / "refs" / "heads" / branch).write_text(f"{sha}\n")
+
+
+def _make_runner(
+    repo_root: Path,
+    *,
+    boot_sha: str | None,
+    boot_wall: float = None,
+    boot_mtime: float = 0.0,
+):
     """Bare GatewayRunner with just the stale-check attributes set."""
+    if boot_wall is None:
+        boot_wall = time.time()
     runner = object.__new__(GatewayRunner)
     runner._repo_root_for_staleness = repo_root
     runner._boot_wall_time = boot_wall
+    runner._boot_git_sha = boot_sha
     runner._boot_repo_mtime = boot_mtime
     runner._stale_code_notified = set()
     runner._stale_code_restart_triggered = False
+    runner._cached_current_sha = boot_sha
+    runner._cached_current_sha_at = boot_wall
     return runner
 
 
-def test_compute_repo_mtime_returns_newest(tmp_path):
-    """_compute_repo_mtime returns the newest mtime across sentinel files."""
-    repo = _make_tmp_repo(tmp_path)
+# ---------------------------------------------------------------------------
+# _read_git_head_sha — raw SHA reader
+# ---------------------------------------------------------------------------
 
-    # Stamp a baseline mtime across all sentinels
-    baseline = time.time() - 100
-    for rel in _STALE_CODE_SENTINELS:
-        os.utime(repo / rel, (baseline, baseline))
-
-    # Touch one file forward
-    newer = time.time()
-    os.utime(repo / "hermes_cli/config.py", (newer, newer))
-
-    result = _compute_repo_mtime(repo)
-    assert abs(result - newer) < 1.0  # within 1s (filesystem mtime resolution)
+def test_read_git_head_sha_branch_ref(tmp_path):
+    """Resolves ref: refs/heads/<branch> → SHA from refs/heads/<branch>."""
+    sha = "b" * 40
+    _make_git_repo(tmp_path, sha=sha, branch="main")
+    assert _read_git_head_sha(tmp_path) == sha
 
 
-def test_compute_repo_mtime_missing_files_returns_zero(tmp_path):
-    """Missing sentinel files return 0.0 (treated as 'can't tell' upstream)."""
-    # tmp_path has none of the sentinels
-    assert _compute_repo_mtime(tmp_path) == 0.0
+def test_read_git_head_sha_detached_head(tmp_path):
+    """Detached HEAD: .git/HEAD contains the SHA directly."""
+    sha = "c" * 40
+    git_dir = tmp_path / ".git"
+    git_dir.mkdir()
+    (git_dir / "HEAD").write_text(f"{sha}\n")
+    assert _read_git_head_sha(tmp_path) == sha
 
 
-def test_compute_repo_mtime_partial_files_still_works(tmp_path):
-    """Partial sentinel presence still returns newest of the readable ones."""
-    (tmp_path / "hermes_cli").mkdir()
-    target = tmp_path / "hermes_cli" / "config.py"
-    target.write_text("# partial\n")
-    target_mtime = time.time() - 50
-    os.utime(target, (target_mtime, target_mtime))
-
-    result = _compute_repo_mtime(tmp_path)
-    assert abs(result - target_mtime) < 1.0
+def test_read_git_head_sha_packed_refs(tmp_path):
+    """Falls back to packed-refs when refs/heads/<branch> is missing."""
+    sha = "d" * 40
+    git_dir = tmp_path / ".git"
+    git_dir.mkdir()
+    (git_dir / "HEAD").write_text("ref: refs/heads/main\n")
+    # No refs/heads/main file — only packed-refs
+    (git_dir / "packed-refs").write_text(
+        f"# pack-refs with: peeled fully-peeled sorted\n"
+        f"{sha} refs/heads/main\n"
+    )
+    assert _read_git_head_sha(tmp_path) == sha
 
 
-def test_detect_stale_code_false_when_no_boot_snapshot(tmp_path):
-    """No boot snapshot → can't tell → not stale (no restart loop)."""
-    repo = _make_tmp_repo(tmp_path)
-    runner = _make_runner(repo, boot_mtime=0.0, boot_wall=0.0)
+def test_read_git_head_sha_worktree_gitdir_file(tmp_path):
+    """Worktree: .git is a file with `gitdir: <path>` pointing to the real git dir.
+
+    Real git worktrees store shared refs (refs/heads/*) in the main
+    checkout's .git/ and write a ``commondir`` pointer into the
+    worktree-gitdir.  The reader must follow commondir to resolve the
+    branch ref — this is the layout Hermes dev sessions actually use.
+    """
+    sha = "e" * 40
+    # Main repo layout
+    main_repo = tmp_path / "main-repo"
+    main_git = main_repo / ".git"
+    (main_git / "refs" / "heads").mkdir(parents=True)
+    (main_git / "HEAD").write_text("ref: refs/heads/main\n")
+    (main_git / "refs" / "heads" / "main").write_text("0" * 40 + "\n")
+
+    # Worktree lives in main-repo/.git/worktrees/<name>/
+    worktree_git_dir = main_git / "worktrees" / "feature"
+    worktree_git_dir.mkdir(parents=True)
+    (worktree_git_dir / "HEAD").write_text("ref: refs/heads/feature\n")
+    # commondir points back at the main .git (relative path, "../..")
+    (worktree_git_dir / "commondir").write_text("../..\n")
+    # Feature branch ref lives in the shared refs/heads
+    (main_git / "refs" / "heads" / "feature").write_text(f"{sha}\n")
+
+    # Worktree checkout with .git file pointing at worktree_git_dir
+    worktree = tmp_path / "wt"
+    worktree.mkdir()
+    (worktree / ".git").write_text(f"gitdir: {worktree_git_dir}\n")
+
+    assert _read_git_head_sha(worktree) == sha
+
+
+def test_read_git_head_sha_worktree_packed_refs_in_common(tmp_path):
+    """Worktree + packed-refs in common dir: fallback still resolves."""
+    sha = "f" * 40
+    main_repo = tmp_path / "main-repo"
+    main_git = main_repo / ".git"
+    main_git.mkdir(parents=True)
+    (main_git / "HEAD").write_text("ref: refs/heads/main\n")
+    # packed-refs in the common (main) .git
+    (main_git / "packed-refs").write_text(
+        f"# pack-refs with: peeled fully-peeled sorted\n"
+        f"{sha} refs/heads/feature\n"
+    )
+
+    worktree_git_dir = main_git / "worktrees" / "feature"
+    worktree_git_dir.mkdir(parents=True)
+    (worktree_git_dir / "HEAD").write_text("ref: refs/heads/feature\n")
+    (worktree_git_dir / "commondir").write_text("../..\n")
+
+    worktree = tmp_path / "wt"
+    worktree.mkdir()
+    (worktree / ".git").write_text(f"gitdir: {worktree_git_dir}\n")
+
+    assert _read_git_head_sha(worktree) == sha
+
+
+def test_read_git_head_sha_no_git_returns_none(tmp_path):
+    """No .git dir → None (non-git install, safely disables the check)."""
+    assert _read_git_head_sha(tmp_path) is None
+
+
+def test_read_git_head_sha_malformed_head_returns_none(tmp_path):
+    """Empty HEAD file → None (don't loop on corrupt repos)."""
+    git_dir = tmp_path / ".git"
+    git_dir.mkdir()
+    (git_dir / "HEAD").write_text("")
+    assert _read_git_head_sha(tmp_path) is None
+
+
+# ---------------------------------------------------------------------------
+# _detect_stale_code — the main regression guard
+# ---------------------------------------------------------------------------
+
+def test_detect_stale_code_false_when_sha_unchanged(tmp_path):
+    """Boot SHA == current SHA → not stale (no restart)."""
+    sha = "a" * 40
+    _make_git_repo(tmp_path, sha=sha)
+    runner = _make_runner(tmp_path, boot_sha=sha)
+    # Force fresh read by expiring the cache
+    runner._cached_current_sha_at = 0.0
     assert runner._detect_stale_code() is False
 
 
-def test_detect_stale_code_false_when_files_unchanged(tmp_path):
-    """Source files at boot mtime → not stale."""
-    repo = _make_tmp_repo(tmp_path)
-    # Freeze all sentinels to the same mtime
-    baseline = time.time() - 100
-    for rel in _STALE_CODE_SENTINELS:
-        os.utime(repo / rel, (baseline, baseline))
-
-    runner = _make_runner(repo, boot_mtime=baseline, boot_wall=baseline)
-    assert runner._detect_stale_code() is False
-
-
-def test_detect_stale_code_true_after_update(tmp_path):
-    """Sentinel files newer than boot snapshot → stale."""
-    repo = _make_tmp_repo(tmp_path)
-    baseline = time.time() - 100
-    for rel in _STALE_CODE_SENTINELS:
-        os.utime(repo / rel, (baseline, baseline))
-
-    runner = _make_runner(repo, boot_mtime=baseline, boot_wall=baseline)
-
-    # Simulate hermes update touching config.py
-    new_mtime = time.time()
-    os.utime(repo / "hermes_cli/config.py", (new_mtime, new_mtime))
-
+def test_detect_stale_code_true_after_git_pull(tmp_path):
+    """Boot SHA != current SHA → stale (hermes update happened)."""
+    boot_sha = "a" * 40
+    _make_git_repo(tmp_path, sha=boot_sha)
+    runner = _make_runner(tmp_path, boot_sha=boot_sha)
+    # Simulate git pull moving HEAD forward
+    _set_head_sha(tmp_path, "b" * 40)
+    runner._cached_current_sha_at = 0.0  # expire cache
     assert runner._detect_stale_code() is True
 
 
-def test_detect_stale_code_ignores_subsecond_drift(tmp_path):
-    """2-second slack prevents false positives on coarse-mtime filesystems."""
-    repo = _make_tmp_repo(tmp_path)
-    baseline = time.time() - 100
+def test_detect_stale_code_ignores_agent_file_edits(tmp_path):
+    """THE CORE REGRESSION: agent edits to source files do NOT trigger restart.
+
+    This is the motivating incident for the SHA-based check.  Under the
+    previous mtime-based scheme, any ``patch`` / ``write_file`` call
+    against run_agent.py / gateway/run.py / hermes_cli/config.py would
+    flip the stale-check to True and force a gateway restart on the
+    next message — even though no update actually happened.  SHA
+    comparison decouples the two: git HEAD only moves on ``git pull``,
+    never on file writes.
+    """
+    sha = "a" * 40
+    _make_git_repo(tmp_path, sha=sha)
+    _make_tmp_repo(tmp_path)  # lay down sentinel files too
+    runner = _make_runner(tmp_path, boot_sha=sha)
+
+    # Simulate the agent editing run_agent.py and gateway/run.py with
+    # mtimes far into the future — exactly the scenario that used to
+    # false-positive the old mtime check.
+    future = time.time() + 10_000
     for rel in _STALE_CODE_SENTINELS:
-        os.utime(repo / rel, (baseline, baseline))
+        p = tmp_path / rel
+        if p.is_file():
+            p.write_text("# agent just edited this\n")
+            os.utime(p, (future, future))
 
-    runner = _make_runner(repo, boot_mtime=baseline, boot_wall=baseline)
-
-    # Touch config.py 1s newer — within the 2s slack → not stale
-    os.utime(repo / "hermes_cli/config.py", (baseline + 1.0, baseline + 1.0))
+    # HEAD SHA has NOT moved — check must stay False.
+    runner._cached_current_sha_at = 0.0  # expire cache
     assert runner._detect_stale_code() is False
 
-    # Touch 5s newer → stale
-    os.utime(repo / "hermes_cli/config.py", (baseline + 5.0, baseline + 5.0))
-    assert runner._detect_stale_code() is True
 
+def test_detect_stale_code_false_for_non_git_install(tmp_path):
+    """Non-git install (no .git dir) → check disabled, never fires."""
+    # No .git dir at all; runner's boot_sha is None
+    runner = _make_runner(tmp_path, boot_sha=None)
+    # Even if we pretended the current SHA differed, the check should
+    # short-circuit on boot_sha=None and return False.
+    assert runner._detect_stale_code() is False
+
+
+def test_detect_stale_code_false_when_no_boot_wall_time(tmp_path):
+    """No boot snapshot at all → can't tell → not stale (no restart loop)."""
+    runner = _make_runner(tmp_path, boot_sha="a" * 40, boot_wall=0.0)
+    assert runner._detect_stale_code() is False
+
+
+def test_detect_stale_code_handles_disappearing_git_dir(tmp_path):
+    """.git vanishes mid-run → current_sha = None → not stale (don't loop)."""
+    sha = "a" * 40
+    _make_git_repo(tmp_path, sha=sha)
+    runner = _make_runner(tmp_path, boot_sha=sha)
+    # Nuke the git dir after boot
+    import shutil
+    shutil.rmtree(tmp_path / ".git")
+    runner._cached_current_sha_at = 0.0  # expire cache
+    assert runner._detect_stale_code() is False
+
+
+# ---------------------------------------------------------------------------
+# SHA cache
+# ---------------------------------------------------------------------------
+
+def test_current_sha_cache_collapses_bursts(tmp_path, monkeypatch):
+    """Consecutive calls inside the TTL window reuse the cached SHA."""
+    sha = "a" * 40
+    _make_git_repo(tmp_path, sha=sha)
+    runner = _make_runner(tmp_path, boot_sha=sha)
+
+    read_calls = {"n": 0}
+    real_reader = _read_git_head_sha
+
+    def counting_reader(repo_root):
+        read_calls["n"] += 1
+        return real_reader(repo_root)
+
+    from gateway import run as run_mod
+    monkeypatch.setattr(run_mod, "_read_git_head_sha", counting_reader)
+
+    # Force cache expiry so the first call definitely reads
+    runner._cached_current_sha_at = 0.0
+    runner._current_git_sha_cached()
+    first_count = read_calls["n"]
+
+    # Immediate second/third calls should hit cache (no new read)
+    runner._current_git_sha_cached()
+    runner._current_git_sha_cached()
+    assert read_calls["n"] == first_count
+
+
+def test_current_sha_cache_expires_after_ttl(tmp_path, monkeypatch):
+    """After _GIT_SHA_CACHE_TTL_SECS elapses, a fresh read happens."""
+    sha = "a" * 40
+    _make_git_repo(tmp_path, sha=sha)
+    runner = _make_runner(tmp_path, boot_sha=sha)
+
+    read_calls = {"n": 0}
+    real_reader = _read_git_head_sha
+
+    def counting_reader(repo_root):
+        read_calls["n"] += 1
+        return real_reader(repo_root)
+
+    from gateway import run as run_mod
+    monkeypatch.setattr(run_mod, "_read_git_head_sha", counting_reader)
+
+    runner._cached_current_sha_at = 0.0
+    runner._current_git_sha_cached()
+    first = read_calls["n"]
+
+    # Age the cache past the TTL
+    runner._cached_current_sha_at = time.time() - (_GIT_SHA_CACHE_TTL_SECS + 1.0)
+    runner._current_git_sha_cached()
+    assert read_calls["n"] == first + 1
+
+
+# ---------------------------------------------------------------------------
+# _trigger_stale_code_restart — idempotency preserved
+# ---------------------------------------------------------------------------
 
 def test_trigger_stale_code_restart_is_idempotent(tmp_path):
     """Calling _trigger_stale_code_restart twice only requests restart once."""
-    repo = _make_tmp_repo(tmp_path)
-    runner = _make_runner(repo, boot_mtime=1.0, boot_wall=1.0)
+    sha = "a" * 40
+    _make_git_repo(tmp_path, sha=sha)
+    runner = _make_runner(tmp_path, boot_sha=sha)
 
     calls = []
 
@@ -153,8 +356,9 @@ def test_trigger_stale_code_restart_is_idempotent(tmp_path):
 
 def test_trigger_stale_code_restart_survives_request_failure(tmp_path):
     """If request_restart raises, we swallow and mark as triggered anyway."""
-    repo = _make_tmp_repo(tmp_path)
-    runner = _make_runner(repo, boot_mtime=1.0, boot_wall=1.0)
+    sha = "a" * 40
+    _make_git_repo(tmp_path, sha=sha)
+    runner = _make_runner(tmp_path, boot_sha=sha)
 
     def boom(*, detached=False, via_service=False):
         raise RuntimeError("no event loop")
@@ -168,56 +372,41 @@ def test_trigger_stale_code_restart_survives_request_failure(tmp_path):
     assert runner._stale_code_restart_triggered is True
 
 
-def test_detect_stale_code_handles_disappearing_repo_root(tmp_path):
-    """If the repo root vanishes after boot, return False (don't loop)."""
-    repo = _make_tmp_repo(tmp_path)
-    baseline = time.time() - 100
-    for rel in _STALE_CODE_SENTINELS:
-        os.utime(repo / rel, (baseline, baseline))
-
-    runner = _make_runner(repo, boot_mtime=baseline, boot_wall=baseline)
-
-    # Remove all sentinel files — _compute_repo_mtime returns 0.0
-    for rel in _STALE_CODE_SENTINELS:
-        (repo / rel).unlink(missing_ok=True)
-
-    assert runner._detect_stale_code() is False
-
+# ---------------------------------------------------------------------------
+# Class-level defaults — tests that build bare runners via object.__new__
+# ---------------------------------------------------------------------------
 
 def test_class_level_defaults_prevent_uninitialized_access():
     """Partial construction via object.__new__ must not crash _detect_stale_code."""
     runner = object.__new__(GatewayRunner)
     # Don't set any instance attrs — class-level defaults should kick in
     runner._repo_root_for_staleness = Path(".")
-    # _boot_wall_time / _boot_repo_mtime fall through to class defaults (0.0)
+    # _boot_wall_time / _boot_git_sha fall through to class defaults
+    # (0.0 and None respectively)
     assert runner._detect_stale_code() is False
     # _stale_code_restart_triggered falls through to class default (False)
     assert runner._stale_code_restart_triggered is False
 
 
-def test_init_captures_boot_snapshot(monkeypatch, tmp_path):
-    """GatewayRunner.__init__ captures a usable stale-code baseline."""
-    # Stub out the heavy parts of __init__ we don't need.  We only want
-    # to prove the stale-code snapshot is captured before anything else.
-    from gateway import run as run_mod
+# ---------------------------------------------------------------------------
+# Legacy mtime reader kept for compatibility — light sanity check only
+# ---------------------------------------------------------------------------
 
-    calls = {}
+def test_compute_repo_mtime_still_returns_newest(tmp_path):
+    """_compute_repo_mtime remains available for any legacy callers."""
+    repo = _make_tmp_repo(tmp_path)
 
-    def fake_compute(repo_root):
-        calls["repo_root"] = repo_root
-        return 1234567890.0
+    baseline = time.time() - 100
+    for rel in _STALE_CODE_SENTINELS:
+        os.utime(repo / rel, (baseline, baseline))
 
-    monkeypatch.setattr(run_mod, "_compute_repo_mtime", fake_compute)
+    newer = time.time()
+    os.utime(repo / "hermes_cli/config.py", (newer, newer))
 
-    # Build a runner without running the full __init__ — then manually
-    # exercise the stale-check init block that __init__ contains.
-    runner = object.__new__(GatewayRunner)
-    runner._boot_wall_time = time.time()
-    runner._repo_root_for_staleness = Path(run_mod.__file__).resolve().parent.parent
-    runner._boot_repo_mtime = run_mod._compute_repo_mtime(runner._repo_root_for_staleness)
-    runner._stale_code_notified = set()
-    runner._stale_code_restart_triggered = False
+    result = _compute_repo_mtime(repo)
+    assert abs(result - newer) < 1.0
 
-    assert runner._boot_repo_mtime == 1234567890.0
-    assert calls["repo_root"] == runner._repo_root_for_staleness
-    assert runner._boot_wall_time > 0
+
+def test_compute_repo_mtime_missing_files_returns_zero(tmp_path):
+    """Legacy sanity: missing sentinels → 0.0."""
+    assert _compute_repo_mtime(tmp_path) == 0.0

tests/hermes_cli/test_cmd_update.py

@@ -173,3 +173,78 @@ class TestCmdUpdateBranchFallback:
             mock_input.assert_not_called()
             captured = capsys.readouterr()
             assert "Non-interactive session" in captured.out
+
+
+class TestCmdUpdateProfileSkillSync:
+    """cmd_update syncs bundled skills to all profiles, including the active one.
+
+    Regression guard for #16176: previously the active profile was excluded
+    from the seed_profile_skills loop, leaving it on stale skill content.
+    """
+
+    @patch("shutil.which", return_value=None)
+    @patch("subprocess.run")
+    def test_active_profile_included_in_skill_sync(
+        self, mock_run, _mock_which, mock_args, capsys
+    ):
+        from pathlib import Path
+
+        mock_run.side_effect = _make_run_side_effect(
+            branch="main", verify_ok=True, commit_count="1"
+        )
+
+        default_p = SimpleNamespace(name="default", path=Path("/fake/.hermes"))
+        active_p = SimpleNamespace(name="bit", path=Path("/fake/.hermes/profiles/bit"))
+        other_p = SimpleNamespace(name="work", path=Path("/fake/.hermes/profiles/work"))
+        all_profiles = [default_p, active_p, other_p]
+
+        synced_paths = []
+
+        def fake_seed(path, quiet=False):
+            synced_paths.append(path)
+            return {"copied": [], "updated": [], "user_modified": []}
+
+        empty_sync = {"copied": [], "updated": [], "user_modified": [], "cleaned": []}
+
+        with (
+            patch("hermes_cli.profiles.list_profiles", return_value=all_profiles),
+            patch("hermes_cli.profiles.seed_profile_skills", side_effect=fake_seed),
+            patch("tools.skills_sync.sync_skills", return_value=empty_sync),
+        ):
+            cmd_update(mock_args)
+
+        assert active_p.path in synced_paths, (
+            f"Active profile 'bit' must be included in skill sync; got: {synced_paths}"
+        )
+        assert set(synced_paths) == {p.path for p in all_profiles}, (
+            f"All profiles must be synced; got: {synced_paths}"
+        )
+
+    @patch("shutil.which", return_value=None)
+    @patch("subprocess.run")
+    def test_single_profile_default_is_synced(
+        self, mock_run, _mock_which, mock_args, capsys
+    ):
+        from pathlib import Path
+
+        mock_run.side_effect = _make_run_side_effect(
+            branch="main", verify_ok=True, commit_count="1"
+        )
+
+        default_p = SimpleNamespace(name="default", path=Path("/fake/.hermes"))
+        synced_paths = []
+
+        def fake_seed(path, quiet=False):
+            synced_paths.append(path)
+            return {"copied": [], "updated": [], "user_modified": []}
+
+        empty_sync = {"copied": [], "updated": [], "user_modified": [], "cleaned": []}
+
+        with (
+            patch("hermes_cli.profiles.list_profiles", return_value=[default_p]),
+            patch("hermes_cli.profiles.seed_profile_skills", side_effect=fake_seed),
+            patch("tools.skills_sync.sync_skills", return_value=empty_sync),
+        ):
+            cmd_update(mock_args)
+
+        assert default_p.path in synced_paths

tests/hermes_cli/test_commands.py

@@ -109,6 +109,12 @@ class TestResolveCommand:
         assert resolve_command("reload_mcp").name == "reload-mcp"
         assert resolve_command("tasks").name == "agents"
 
+    def test_topic_is_gateway_command(self):
+        topic = resolve_command("topic")
+        assert topic is not None
+        assert topic.name == "topic"
+        assert "topic" in GATEWAY_KNOWN_COMMANDS
+
     def test_leading_slash_stripped(self):
         assert resolve_command("/help").name == "help"
         assert resolve_command("/bg").name == "background"

tests/hermes_cli/test_doctor.py

@@ -663,3 +663,79 @@ def test_run_doctor_opencode_go_skips_invalid_models_probe(monkeypatch, tmp_path
     )
     assert not any(url == "https://opencode.ai/zen/go/v1/models" for url, _, _ in calls)
     assert not any("opencode" in url.lower() and "models" in url.lower() for url, _, _ in calls)
+
+
+class TestGitHubTokenCheck:
+    """Tests for GitHub token / gh auth detection in doctor."""
+
+    def test_no_token_and_not_gh_authenticated_shows_warn(self, monkeypatch, tmp_path):
+        home = tmp_path / ".hermes"
+        home.mkdir(parents=True, exist_ok=True)
+        monkeypatch.setenv("HERMES_HOME", str(home))
+        monkeypatch.setenv("PATH", "/nonexistent")  # gh not found
+
+        from hermes_cli.doctor import run_doctor, _DHH
+        import io, contextlib
+
+        buf = io.StringIO()
+        with contextlib.redirect_stdout(buf):
+            run_doctor(Namespace(fix=False))
+        out = buf.getvalue()
+
+        assert "No GITHUB_TOKEN" in out
+        assert "60 req/hr" in out
+
+    def test_token_env_present_shows_ok(self, monkeypatch, tmp_path):
+        home = tmp_path / ".hermes"
+        home.mkdir(parents=True, exist_ok=True)
+        monkeypatch.setenv("HERMES_HOME", str(home))
+        monkeypatch.setenv("GITHUB_TOKEN", "ghp_test123")
+        monkeypatch.setenv("PATH", "/nonexistent")  # gh not found
+
+        from hermes_cli.doctor import run_doctor
+        import io, contextlib
+
+        buf = io.StringIO()
+        with contextlib.redirect_stdout(buf):
+            run_doctor(Namespace(fix=False))
+        out = buf.getvalue()
+
+        assert "GitHub token configured" in out
+
+    def test_gh_authenticated_without_env_token_shows_ok(self, monkeypatch, tmp_path):
+        home = tmp_path / ".hermes"
+        home.mkdir(parents=True, exist_ok=True)
+        monkeypatch.setenv("HERMES_HOME", str(home))
+        # No GITHUB_TOKEN or GH_TOKEN
+        monkeypatch.delenv("GITHUB_TOKEN", raising=False)
+        monkeypatch.delenv("GH_TOKEN", raising=False)
+
+        # Mock gh to return success
+        import shutil
+        real_which = shutil.which
+        def mock_which(cmd):
+            return "/usr/local/bin/gh" if cmd == "gh" else real_which(cmd)
+        monkeypatch.setattr(shutil, "which", mock_which)
+
+        call_log = []
+        def mock_run(cmd, **kwargs):
+            call_log.append(cmd)
+            if cmd[:2] == ["gh", "auth"]:
+                result = types.SimpleNamespace(returncode=0, stdout="", stderr="")
+            else:
+                result = types.SimpleNamespace(returncode=1, stdout="", stderr="")
+            return result
+
+        import subprocess
+        monkeypatch.setattr(subprocess, "run", mock_run)
+
+        from hermes_cli.doctor import run_doctor
+        import io, contextlib
+
+        buf = io.StringIO()
+        with contextlib.redirect_stdout(buf):
+            run_doctor(Namespace(fix=False))
+        out = buf.getvalue()
+
+        assert "gh auth" in str(call_log) or any(c[0] == "gh" for c in call_log), f"gh not called: {call_log}"
+        assert "GitHub authenticated via gh CLI" in out or "token configured" in out

tests/hermes_cli/test_gateway_service.py

@@ -182,6 +182,43 @@ class TestGeneratedSystemdUnits:
 
         assert "/home/test/.nvm/versions/node/v24.14.0/bin" in unit
 
+    def test_user_unit_includes_wsl_windows_interop_paths(self, monkeypatch):
+        monkeypatch.setattr(gateway_cli, "is_wsl", lambda: True)
+        monkeypatch.setenv(
+            "PATH",
+            "/usr/local/bin:/mnt/c/WINDOWS/system32:/mnt/c/WINDOWS/System32/WindowsPowerShell/v1.0/",
+        )
+        monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: None)
+
+        unit = gateway_cli.generate_systemd_unit(system=False)
+
+        assert "/mnt/c/WINDOWS/system32" in unit
+        assert "/mnt/c/WINDOWS/System32/WindowsPowerShell/v1.0/" in unit
+
+    def test_user_unit_omits_windows_interop_paths_outside_wsl(self, monkeypatch):
+        monkeypatch.setattr(gateway_cli, "is_wsl", lambda: False)
+        monkeypatch.setenv("PATH", "/usr/local/bin:/mnt/c/WINDOWS/system32")
+        monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: None)
+
+        unit = gateway_cli.generate_systemd_unit(system=False)
+
+        assert "/mnt/c/WINDOWS/system32" not in unit
+
+    def test_system_unit_includes_wsl_windows_interop_paths(self, monkeypatch):
+        monkeypatch.setattr(gateway_cli, "is_wsl", lambda: True)
+        monkeypatch.setattr(
+            gateway_cli,
+            "_system_service_identity",
+            lambda run_as_user=None: ("alice", "alice", "/home/alice"),
+        )
+        monkeypatch.setattr(gateway_cli, "_hermes_home_for_target_user", lambda home: "/home/alice/.hermes")
+        monkeypatch.setenv("PATH", "/usr/local/bin:/mnt/c/WINDOWS/system32")
+        monkeypatch.setattr(gateway_cli.shutil, "which", lambda cmd: None)
+
+        unit = gateway_cli.generate_systemd_unit(system=True, run_as_user="alice")
+
+        assert "/mnt/c/WINDOWS/system32" in unit
+
     def test_system_unit_avoids_recursive_execstop_and_uses_extended_stop_timeout(self):
         unit = gateway_cli.generate_systemd_unit(system=True)
 

tests/hermes_cli/test_ollama_cloud_provider.py

@@ -401,6 +401,103 @@ class TestOllamaCloudProvidersNew:
         assert pdef.transport == "openai_chat"
 
 
+# ── Cloud Suffix Stripping ──
+
+class TestOllamaCloudSuffixStripping:
+    """models.dev appends :cloud / -cloud suffixes that the live API omits.
+
+    fetch_ollama_cloud_models() must normalise these before the dedup merge so
+    users never see broken IDs like 'kimi-k2.6:cloud' in the model picker.
+    """
+
+    def test_strips_colon_cloud_suffix(self, tmp_path, monkeypatch):
+        """:cloud suffix from models.dev is stripped before merge."""
+        from hermes_cli.models import fetch_ollama_cloud_models
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
+
+        mock_mdev = {
+            "ollama-cloud": {
+                "models": {"kimi-k2.6:cloud": {"tool_call": True}}
+            }
+        }
+        with patch("agent.models_dev.fetch_models_dev", return_value=mock_mdev):
+            result = fetch_ollama_cloud_models(force_refresh=True)
+
+        assert "kimi-k2.6" in result
+        assert "kimi-k2.6:cloud" not in result
+
+    def test_strips_dash_cloud_suffix(self, tmp_path, monkeypatch):
+        """-cloud suffix from models.dev is stripped before merge."""
+        from hermes_cli.models import fetch_ollama_cloud_models
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
+
+        mock_mdev = {
+            "ollama-cloud": {
+                "models": {"qwen3-coder:480b-cloud": {"tool_call": True}}
+            }
+        }
+        with patch("agent.models_dev.fetch_models_dev", return_value=mock_mdev):
+            result = fetch_ollama_cloud_models(force_refresh=True)
+
+        assert "qwen3-coder:480b" in result
+        assert "qwen3-coder:480b-cloud" not in result
+
+    def test_no_duplicate_when_live_clean_and_mdev_suffixed(self, tmp_path, monkeypatch):
+        """Live API returns clean ID; mdev has :cloud variant — result has exactly one entry."""
+        from hermes_cli.models import fetch_ollama_cloud_models
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        monkeypatch.setenv("OLLAMA_API_KEY", "test-key")
+
+        mock_mdev = {
+            "ollama-cloud": {
+                "models": {
+                    "kimi-k2.6:cloud": {"tool_call": True},
+                    "glm-5.1:cloud": {"tool_call": True},
+                }
+            }
+        }
+        with patch("hermes_cli.models.fetch_api_models", return_value=["kimi-k2.6", "glm-5.1"]), \
+             patch("agent.models_dev.fetch_models_dev", return_value=mock_mdev):
+            result = fetch_ollama_cloud_models(force_refresh=True)
+
+        assert result.count("kimi-k2.6") == 1
+        assert result.count("glm-5.1") == 1
+        assert "kimi-k2.6:cloud" not in result
+        assert "glm-5.1:cloud" not in result
+
+    def test_unsuffixed_model_id_unchanged(self, tmp_path, monkeypatch):
+        """Model IDs without :cloud / -cloud suffix are passed through unchanged."""
+        from hermes_cli.models import fetch_ollama_cloud_models
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
+
+        mock_mdev = {
+            "ollama-cloud": {
+                "models": {"nemotron-3-nano:30b": {"tool_call": True}}
+            }
+        }
+        with patch("agent.models_dev.fetch_models_dev", return_value=mock_mdev):
+            result = fetch_ollama_cloud_models(force_refresh=True)
+
+        assert "nemotron-3-nano:30b" in result
+
+    def test_strip_suffix_helper(self):
+        """Unit test for the _strip_ollama_cloud_suffix helper."""
+        from hermes_cli.models import _strip_ollama_cloud_suffix
+
+        assert _strip_ollama_cloud_suffix("kimi-k2.6:cloud") == "kimi-k2.6"
+        assert _strip_ollama_cloud_suffix("glm-5.1:cloud") == "glm-5.1"
+        assert _strip_ollama_cloud_suffix("qwen3-coder:480b-cloud") == "qwen3-coder:480b"
+        assert _strip_ollama_cloud_suffix("nemotron-3-nano:30b") == "nemotron-3-nano:30b"
+        assert _strip_ollama_cloud_suffix("") == ""
+
+
 # ── Auxiliary Model ──
 
 class TestOllamaCloudAuxiliary:

tests/plugins/test_kanban_dashboard_plugin.py

@@ -914,3 +914,161 @@ def test_create_task_probe_error_does_not_break_create(client, monkeypatch):
     )
     assert r.status_code == 200
     assert r.json()["task"]["title"] == "resilient"
+
+
+
+# ---------------------------------------------------------------------------
+# Home-channel subscription endpoints (#19534 follow-up: GUI opt-in)
+# ---------------------------------------------------------------------------
+#
+# Dashboard surface for per-task, per-platform notification toggles. The
+# backend endpoints read the live GatewayConfig, so tests set env vars
+# (BOT_TOKEN + HOME_CHANNEL) to simulate a user who has run /sethome on
+# telegram and discord.
+
+
+@pytest.fixture
+def with_home_channels(monkeypatch):
+    """Simulate a user with home channels set on telegram and discord."""
+    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "abc:fake")
+    monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "1234567")
+    monkeypatch.setenv("TELEGRAM_HOME_CHANNEL_THREAD_ID", "42")
+    monkeypatch.setenv("TELEGRAM_HOME_CHANNEL_NAME", "Main TG")
+    monkeypatch.setenv("DISCORD_BOT_TOKEN", "disc_fake")
+    monkeypatch.setenv("DISCORD_HOME_CHANNEL", "9999999")
+    monkeypatch.setenv("DISCORD_HOME_CHANNEL_NAME", "Main Discord")
+    # Slack has a token but NO home — should be excluded from the list.
+    monkeypatch.setenv("SLACK_BOT_TOKEN", "slack_fake")
+
+
+def test_home_channels_lists_only_platforms_with_home(client, with_home_channels):
+    """GET /home-channels returns entries only for platforms where the
+    user has set a home; untoggled-subscribed bool is false by default."""
+    r = client.get("/api/plugins/kanban/home-channels")
+    assert r.status_code == 200
+    platforms = {h["platform"] for h in r.json()["home_channels"]}
+    assert platforms == {"telegram", "discord"}, (
+        f"slack has a token but no home — must not appear. got {platforms}"
+    )
+    for h in r.json()["home_channels"]:
+        assert h["subscribed"] is False
+
+
+def test_home_channels_no_task_id_all_unsubscribed(client, with_home_channels):
+    """Without task_id, every entry's subscribed=false (UI "no task" state)."""
+    r = client.get("/api/plugins/kanban/home-channels")
+    assert r.status_code == 200
+    assert all(not h["subscribed"] for h in r.json()["home_channels"])
+
+
+def test_home_subscribe_creates_notify_sub_row(client, with_home_channels):
+    """POST .../home-subscribe/telegram writes a kanban_notify_subs row
+    keyed to the telegram home's (chat_id, thread_id)."""
+    from hermes_cli import kanban_db as kb
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+
+    r = client.post(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/telegram")
+    assert r.status_code == 200
+    assert r.json()["ok"] is True
+
+    conn = kb.connect()
+    try:
+        subs = kb.list_notify_subs(conn, t["id"])
+    finally:
+        conn.close()
+    assert len(subs) == 1
+    assert subs[0]["platform"] == "telegram"
+    assert subs[0]["chat_id"] == "1234567"
+    assert subs[0]["thread_id"] == "42"
+
+
+def test_home_subscribe_flips_subscribed_flag_in_subsequent_get(client, with_home_channels):
+    """After subscribe, the GET endpoint reports subscribed=true for that
+    platform and false for the others."""
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    client.post(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/telegram")
+
+    r = client.get(f"/api/plugins/kanban/home-channels?task_id={t['id']}")
+    flags = {h["platform"]: h["subscribed"] for h in r.json()["home_channels"]}
+    assert flags == {"telegram": True, "discord": False}
+
+
+def test_home_subscribe_is_idempotent(client, with_home_channels):
+    """Re-subscribing keeps a single row at the DB layer."""
+    from hermes_cli import kanban_db as kb
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    client.post(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/telegram")
+    client.post(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/telegram")
+    client.post(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/telegram")
+    conn = kb.connect()
+    try:
+        assert len(kb.list_notify_subs(conn, t["id"])) == 1
+    finally:
+        conn.close()
+
+
+def test_home_subscribe_unknown_platform_returns_404(client, with_home_channels):
+    """Platforms without a home configured (slack in the fixture) return 404."""
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    r = client.post(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/slack")
+    assert r.status_code == 404
+    assert "slack" in r.json()["detail"]
+
+
+def test_home_subscribe_unknown_task_returns_404(client, with_home_channels):
+    r = client.post("/api/plugins/kanban/tasks/t_nonexistent/home-subscribe/telegram")
+    assert r.status_code == 404
+
+
+def test_home_unsubscribe_removes_notify_sub_row(client, with_home_channels):
+    """DELETE .../home-subscribe/telegram removes the matching row."""
+    from hermes_cli import kanban_db as kb
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+    client.post(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/telegram")
+    r = client.delete(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/telegram")
+    assert r.status_code == 200
+
+    conn = kb.connect()
+    try:
+        assert kb.list_notify_subs(conn, t["id"]) == []
+    finally:
+        conn.close()
+
+
+def test_home_subscribe_multiple_platforms_independent(client, with_home_channels):
+    """Subscribing on telegram does not affect discord and vice versa."""
+    from hermes_cli import kanban_db as kb
+    t = client.post("/api/plugins/kanban/tasks", json={"title": "x"}).json()["task"]
+
+    client.post(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/telegram")
+    client.post(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/discord")
+
+    conn = kb.connect()
+    try:
+        subs = {s["platform"]: s for s in kb.list_notify_subs(conn, t["id"])}
+    finally:
+        conn.close()
+    assert set(subs) == {"telegram", "discord"}
+
+    # Unsubscribe telegram only.
+    client.delete(f"/api/plugins/kanban/tasks/{t['id']}/home-subscribe/telegram")
+    conn = kb.connect()
+    try:
+        subs = {s["platform"]: s for s in kb.list_notify_subs(conn, t["id"])}
+    finally:
+        conn.close()
+    assert set(subs) == {"discord"}
+
+
+def test_home_channels_empty_when_no_homes_configured(client, monkeypatch):
+    """Zero platforms with a home -> empty list (UI hides the section)."""
+    # No BOT_TOKEN env vars set → load_gateway_config().platforms is empty.
+    for var in [
+        "TELEGRAM_BOT_TOKEN", "TELEGRAM_HOME_CHANNEL",
+        "DISCORD_BOT_TOKEN", "DISCORD_HOME_CHANNEL",
+        "SLACK_BOT_TOKEN",
+    ]:
+        monkeypatch.delenv(var, raising=False)
+    r = client.get("/api/plugins/kanban/home-channels")
+    assert r.status_code == 200
+    assert r.json()["home_channels"] == []

tests/run_agent/test_iteration_budget_race.py

@@ -0,0 +1,109 @@
+"""Tests for IterationBudget thread safety.
+
+The `used` property must acquire the lock before reading `_used` to prevent
+data races with concurrent `consume()` / `refund()` calls.
+"""
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+
+def test_iteration_budget_used_is_thread_safe():
+    """Iterating `used` while other threads consume/refund must not crash.
+
+    Before the fix, `used` returned `_used` directly without holding the lock,
+    so a concurrent `consume()` could observe a partially-updated value or
+    cause the C-level `list.append` to raise a ValueError ("list size changed").
+    """
+    from run_agent import IterationBudget
+
+    budget = IterationBudget(max_total=1000)
+    num_threads = 10
+    operations_per_thread = 200
+
+    errors = []
+
+    def worker(consume: bool):
+        try:
+            for _ in range(operations_per_thread):
+                if consume:
+                    budget.consume()
+                else:
+                    budget.refund()
+                # Also read `used` to exercise the property
+                _ = budget.used
+        except Exception as exc:
+            errors.append(exc)
+
+    with ThreadPoolExecutor(max_workers=num_threads * 2) as executor:
+        # Half the threads consume, half refund
+        futures = []
+        for i in range(num_threads):
+            consume = i < num_threads // 2
+            futures.append(executor.submit(worker, consume))
+            futures.append(executor.submit(worker, consume))
+
+        for f in futures:
+            f.result()
+
+    assert not errors, f"Thread safety violation: {errors}"
+    # Final value should be within expected bounds
+    assert 0 <= budget.used <= budget.max_total
+
+
+def test_iteration_budget_consume_returns_false_when_exhausted():
+    """consume() must return False once the budget is exhausted."""
+    from run_agent import IterationBudget
+
+    budget = IterationBudget(max_total=3)
+    assert budget.consume() is True
+    assert budget.consume() is True
+    assert budget.consume() is True
+    assert budget.consume() is False
+
+
+def test_iteration_budget_refund_restores_consume():
+    """refund() after consume() must allow one more consume()."""
+    from run_agent import IterationBudget
+
+    budget = IterationBudget(max_total=2)
+    assert budget.consume() is True
+    assert budget.consume() is True
+    assert budget.consume() is False  # exhausted
+    budget.refund()
+    assert budget.consume() is True
+
+
+def test_iteration_budget_used_reflects_consume_and_refund():
+    """used property must accurately reflect consume() and refund() calls."""
+    from run_agent import IterationBudget
+
+    budget = IterationBudget(max_total=10)
+
+    assert budget.used == 0
+    budget.consume()
+    assert budget.used == 1
+    budget.consume()
+    assert budget.used == 2
+    budget.refund()
+    assert budget.used == 1
+    budget.refund()
+    assert budget.used == 0
+
+
+def test_iteration_budget_remaining():
+    """remaining property must equal max_total - used."""
+    from run_agent import IterationBudget
+
+    budget = IterationBudget(max_total=5)
+
+    assert budget.remaining == 5
+    budget.consume()
+    assert budget.remaining == 4
+    budget.consume()
+    budget.consume()
+    assert budget.remaining == 2
+    budget.refund()
+    assert budget.remaining == 3

tests/skills/test_google_workspace_credential_files.py

@@ -0,0 +1,102 @@
+"""Regression test: google-workspace SKILL.md must declare required_credential_files.
+
+PR #9931 accidentally removed the required_credential_files header, which broke
+credential file mounting in Docker/Modal remote backends (#16452). This test
+prevents the regression from silently reappearing.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+SKILL_MD = (
+    Path(__file__).resolve().parents[2]
+    / "skills/productivity/google-workspace/SKILL.md"
+)
+
+_EXPECTED_PATHS = {"google_token.json", "google_client_secret.json"}
+
+
+def _parse_frontmatter(content: str) -> dict:
+    from agent.skill_utils import parse_frontmatter
+
+    fm, _ = parse_frontmatter(content)
+    return fm
+
+
+class TestGoogleWorkspaceCredentialFiles:
+    def test_required_credential_files_present_in_skill_md(self):
+        content = SKILL_MD.read_text(encoding="utf-8")
+        fm = _parse_frontmatter(content)
+        entries = fm.get("required_credential_files")
+        assert entries, "required_credential_files missing from google-workspace SKILL.md"
+        assert isinstance(entries, list), "required_credential_files must be a list"
+        paths = {
+            (e["path"] if isinstance(e, dict) else e)
+            for e in entries
+        }
+        assert _EXPECTED_PATHS <= paths, (
+            f"Missing entries in required_credential_files: {_EXPECTED_PATHS - paths}"
+        )
+
+    def test_entries_are_registered_when_files_exist(self, tmp_path):
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        (hermes_home / "google_token.json").write_text("{}")
+        (hermes_home / "google_client_secret.json").write_text("{}")
+
+        from tools.credential_files import (
+            clear_credential_files,
+            get_credential_file_mounts,
+            register_credential_files,
+        )
+
+        clear_credential_files()
+        try:
+            content = SKILL_MD.read_text(encoding="utf-8")
+            fm = _parse_frontmatter(content)
+            entries = fm.get("required_credential_files", [])
+
+            with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
+                missing = register_credential_files(entries)
+
+            assert missing == [], f"Unexpected missing files: {missing}"
+            mounts = get_credential_file_mounts()
+            container_paths = {m["container_path"] for m in mounts}
+            assert "/root/.hermes/google_token.json" in container_paths
+            assert "/root/.hermes/google_client_secret.json" in container_paths
+        finally:
+            clear_credential_files()
+
+    def test_missing_token_is_reported(self, tmp_path):
+        """google_token.json absent (first-time setup) — reported as missing, client secret still mounts."""
+        hermes_home = tmp_path / ".hermes"
+        hermes_home.mkdir()
+        (hermes_home / "google_client_secret.json").write_text("{}")
+
+        from tools.credential_files import (
+            clear_credential_files,
+            get_credential_file_mounts,
+            register_credential_files,
+        )
+
+        clear_credential_files()
+        try:
+            content = SKILL_MD.read_text(encoding="utf-8")
+            fm = _parse_frontmatter(content)
+            entries = fm.get("required_credential_files", [])
+
+            with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
+                missing = register_credential_files(entries)
+
+            assert "google_token.json" in missing
+            mounts = get_credential_file_mounts()
+            container_paths = {m["container_path"] for m in mounts}
+            assert "/root/.hermes/google_client_secret.json" in container_paths
+            assert "/root/.hermes/google_token.json" not in container_paths
+        finally:
+            clear_credential_files()

tests/test_hermes_state.py

@@ -35,6 +35,7 @@ class TestSessionLifecycle:
         assert session["model"] == "test-model"
         assert session["ended_at"] is None
 
+
     def test_get_nonexistent_session(self, db):
         assert db.get_session("nonexistent") is None
 
@@ -1421,6 +1422,242 @@ class TestSchemaInit:
         columns = {row[1] for row in cursor.fetchall()}
         assert "title" in columns
 
+    def test_topic_mode_schema_is_not_auto_migrated_on_open(self, tmp_path):
+        """Opening an old DB should not add topic-mode columns until /topic opts in.
+
+        The gateway must remain rollback-safe: simply upgrading Hermes and starting
+        the old bot should not eagerly mutate the state DB for this feature.
+        """
+        old_db = tmp_path / "old.db"
+        import sqlite3
+
+        conn = sqlite3.connect(old_db)
+        conn.executescript(
+            """
+            CREATE TABLE schema_version (version INTEGER NOT NULL);
+            INSERT INTO schema_version VALUES (11);
+            CREATE TABLE sessions (
+                id TEXT PRIMARY KEY,
+                source TEXT NOT NULL,
+                user_id TEXT,
+                model TEXT,
+                model_config TEXT,
+                system_prompt TEXT,
+                parent_session_id TEXT,
+                started_at REAL NOT NULL,
+                ended_at REAL,
+                end_reason TEXT,
+                message_count INTEGER DEFAULT 0,
+                tool_call_count INTEGER DEFAULT 0,
+                input_tokens INTEGER DEFAULT 0,
+                output_tokens INTEGER DEFAULT 0,
+                cache_read_tokens INTEGER DEFAULT 0,
+                cache_write_tokens INTEGER DEFAULT 0,
+                reasoning_tokens INTEGER DEFAULT 0,
+                billing_provider TEXT,
+                billing_base_url TEXT,
+                billing_mode TEXT,
+                estimated_cost_usd REAL,
+                actual_cost_usd REAL,
+                cost_status TEXT,
+                cost_source TEXT,
+                pricing_version TEXT,
+                title TEXT,
+                api_call_count INTEGER DEFAULT 0,
+                FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
+            );
+            CREATE TABLE messages (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                session_id TEXT NOT NULL REFERENCES sessions(id),
+                role TEXT NOT NULL,
+                content TEXT,
+                tool_call_id TEXT,
+                tool_calls TEXT,
+                tool_name TEXT,
+                timestamp REAL NOT NULL,
+                token_count INTEGER,
+                finish_reason TEXT,
+                reasoning TEXT,
+                reasoning_content TEXT,
+                reasoning_details TEXT,
+                codex_reasoning_items TEXT,
+                codex_message_items TEXT
+            );
+            """
+        )
+        conn.close()
+
+        db = SessionDB(db_path=old_db)
+        cursor = db._conn.execute("PRAGMA table_info(sessions)")
+        columns = {row[1] for row in cursor.fetchall()}
+        assert {"chat_id", "chat_type", "thread_id", "session_key"}.isdisjoint(columns)
+        db.close()
+
+    def test_apply_telegram_topic_migration_creates_topic_tables_explicitly(self, tmp_path):
+        """The /topic opt-in path owns the DB migration for Telegram topic mode."""
+        old_db = tmp_path / "old.db"
+        import sqlite3
+
+        conn = sqlite3.connect(old_db)
+        conn.executescript(
+            """
+            CREATE TABLE schema_version (version INTEGER NOT NULL);
+            INSERT INTO schema_version VALUES (11);
+            CREATE TABLE sessions (
+                id TEXT PRIMARY KEY,
+                source TEXT NOT NULL,
+                user_id TEXT,
+                model TEXT,
+                model_config TEXT,
+                system_prompt TEXT,
+                parent_session_id TEXT,
+                started_at REAL NOT NULL,
+                ended_at REAL,
+                end_reason TEXT,
+                message_count INTEGER DEFAULT 0,
+                tool_call_count INTEGER DEFAULT 0,
+                input_tokens INTEGER DEFAULT 0,
+                output_tokens INTEGER DEFAULT 0,
+                cache_read_tokens INTEGER DEFAULT 0,
+                cache_write_tokens INTEGER DEFAULT 0,
+                reasoning_tokens INTEGER DEFAULT 0,
+                billing_provider TEXT,
+                billing_base_url TEXT,
+                billing_mode TEXT,
+                estimated_cost_usd REAL,
+                actual_cost_usd REAL,
+                cost_status TEXT,
+                cost_source TEXT,
+                pricing_version TEXT,
+                title TEXT,
+                api_call_count INTEGER DEFAULT 0,
+                FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
+            );
+            CREATE TABLE messages (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                session_id TEXT NOT NULL REFERENCES sessions(id),
+                role TEXT NOT NULL,
+                content TEXT,
+                tool_call_id TEXT,
+                tool_calls TEXT,
+                tool_name TEXT,
+                timestamp REAL NOT NULL,
+                token_count INTEGER,
+                finish_reason TEXT,
+                reasoning TEXT,
+                reasoning_content TEXT,
+                reasoning_details TEXT,
+                codex_reasoning_items TEXT,
+                codex_message_items TEXT
+            );
+            """
+        )
+        conn.close()
+
+        db = SessionDB(db_path=old_db)
+        db.apply_telegram_topic_migration()
+
+        tables = {
+            row[0]
+            for row in db._conn.execute(
+                "SELECT name FROM sqlite_master WHERE type = 'table'"
+            ).fetchall()
+        }
+        assert "telegram_dm_topic_mode" in tables
+        assert "telegram_dm_topic_bindings" in tables
+        assert db.get_meta("telegram_dm_topic_schema_version") == "2"
+        db.close()
+
+    def test_telegram_topic_binding_roundtrip_requires_explicit_schema(self, tmp_path):
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session(
+            session_id="topic-session",
+            source="telegram",
+            user_id="208214988",
+        )
+
+        assert db.get_telegram_topic_binding(chat_id="208214988", thread_id="17585") is None
+
+        db.bind_telegram_topic(
+            chat_id="208214988",
+            thread_id="17585",
+            user_id="208214988",
+            session_key="telegram:dm:208214988:thread:17585",
+            session_id="topic-session",
+        )
+
+        binding = db.get_telegram_topic_binding(chat_id="208214988", thread_id="17585")
+        assert binding is not None
+        assert binding["chat_id"] == "208214988"
+        assert binding["thread_id"] == "17585"
+        assert binding["user_id"] == "208214988"
+        assert binding["session_key"] == "telegram:dm:208214988:thread:17585"
+        assert binding["session_id"] == "topic-session"
+        assert db.get_meta("telegram_dm_topic_schema_version") == "2"
+        db.close()
+
+    def test_telegram_topic_binding_refuses_to_relink_session_to_another_topic(self, tmp_path):
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session(
+            session_id="topic-session",
+            source="telegram",
+            user_id="208214988",
+        )
+        db.bind_telegram_topic(
+            chat_id="208214988",
+            thread_id="17585",
+            user_id="208214988",
+            session_key="key-17585",
+            session_id="topic-session",
+        )
+
+        with pytest.raises(ValueError, match="already linked"):
+            db.bind_telegram_topic(
+                chat_id="208214988",
+                thread_id="99999",
+                user_id="208214988",
+                session_key="key-99999",
+                session_id="topic-session",
+            )
+        db.close()
+
+    def test_list_unlinked_telegram_sessions_for_user_excludes_bound_and_other_users(self, tmp_path):
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session(
+            session_id="old-unlinked",
+            source="telegram",
+            user_id="208214988",
+        )
+        db.set_session_title("old-unlinked", "Old research")
+        db.append_message("old-unlinked", "user", "first prompt")
+        db.create_session(
+            session_id="already-linked",
+            source="telegram",
+            user_id="208214988",
+        )
+        db.bind_telegram_topic(
+            chat_id="208214988",
+            thread_id="17585",
+            user_id="208214988",
+            session_key="key-17585",
+            session_id="already-linked",
+        )
+        db.create_session(
+            session_id="other-user",
+            source="telegram",
+            user_id="someone-else",
+        )
+
+        sessions = db.list_unlinked_telegram_sessions_for_user(
+            chat_id="208214988",
+            user_id="208214988",
+        )
+
+        assert [s["id"] for s in sessions] == ["old-unlinked"]
+        assert sessions[0]["title"] == "Old research"
+        assert sessions[0]["preview"] == "first prompt"
+        db.close()
+
     def test_migration_from_v2(self, tmp_path):
         """Simulate a v2 database and verify migration adds title column."""
         import sqlite3

tests/tools/test_file_operations.py

@@ -2,6 +2,7 @@
 
 import os
 import pytest
+import subprocess
 from pathlib import Path
 from unittest.mock import MagicMock
 
@@ -388,6 +389,66 @@ class TestSearchPathValidation:
         assert "search failed" in result.error.lower() or "Search error" in result.error
 
 
+class TestSearchFilesFallbackHiddenPaths:
+    def _make_env(self):
+        env = MagicMock()
+        env.cwd = "/"
+
+        def execute(command, **kwargs):
+            completed = subprocess.run(
+                command,
+                shell=True,
+                text=True,
+                capture_output=True,
+            )
+            return {
+                "output": completed.stdout,
+                "returncode": completed.returncode,
+            }
+
+        env.execute = execute
+        return env
+
+    def test_hidden_root_with_hidden_ancestor_includes_files(self, tmp_path, monkeypatch):
+        """Fallback find should include visible files when path is inside hidden root."""
+        root = tmp_path / ".hermes" / "logs"
+        root.mkdir(parents=True)
+        visible_file = root / "agent.log"
+        hidden_dir_file = root / ".hidden" / "secret.log"
+        nested_hidden_file = root / "nested" / ".secret.log"
+        visible_nested_file = root / "nested" / "visible.log"
+
+        for p in [visible_file, nested_hidden_file, visible_nested_file, hidden_dir_file]:
+            p.parent.mkdir(parents=True, exist_ok=True)
+            p.write_text("x")
+
+        ops = ShellFileOperations(self._make_env())
+        monkeypatch.setattr(ops, "_has_command", lambda command: command == "find")
+        result = ops._search_files("*.log", str(root), limit=50, offset=0)
+
+        assert result.error is None
+        assert set(result.files) == {str(visible_file), str(visible_nested_file)}
+
+    def test_normal_root_still_excludes_hidden_descendants(self, tmp_path, monkeypatch):
+        """Fallback find should still exclude hidden descendant paths for normal roots."""
+        root = tmp_path / "repo"
+        root.mkdir()
+        visible_file = root / "agent.log"
+        visible_nested_file = root / "nested" / "visible.log"
+        hidden_dir_file = root / ".hidden" / "secret.log"
+
+        for p in [visible_file, visible_nested_file, hidden_dir_file]:
+            p.parent.mkdir(parents=True, exist_ok=True)
+            p.write_text("x")
+
+        ops = ShellFileOperations(self._make_env())
+        monkeypatch.setattr(ops, "_has_command", lambda command: command == "find")
+        result = ops._search_files("*.log", str(root), limit=50, offset=0)
+
+        assert result.error is None
+        assert set(result.files) == {str(visible_file), str(visible_nested_file)}
+
+
 class TestShellFileOpsWriteDenied:
     def test_write_file_denied_path(self, file_ops):
         result = file_ops.write_file("~/.ssh/authorized_keys", "evil key")

tests/tools/test_file_operations_edge_cases.py

@@ -8,7 +8,7 @@ Covers:
 import pytest
 from unittest.mock import MagicMock, patch
 
-from tools.file_operations import ShellFileOperations
+from tools.file_operations import ShellFileOperations, _parse_search_context_line
 
 
 # =========================================================================
@@ -204,3 +204,67 @@ class TestPaginationBounds:
         rg_commands = [cmd for cmd in commands if cmd.startswith("rg --files")]
         assert rg_commands
         assert "| head -n 1" in rg_commands[0]
+
+
+# =========================================================================
+# Search context parsing
+# =========================================================================
+
+
+class TestSearchContextParsing:
+    def test_parse_search_context_line_prefers_rightmost_numeric_separator(self):
+        parsed = _parse_search_context_line("dir/file-12-name.py-8-context here")
+
+        assert parsed == ("dir/file-12-name.py", 8, "context here")
+
+    def test_search_with_rg_context_handles_filename_with_dash_digits(self):
+        env = MagicMock()
+        env.cwd = "/tmp"
+        ops = ShellFileOperations(env)
+
+        with patch.object(ops, "_exec") as mock_exec:
+            mock_exec.return_value = MagicMock(
+                exit_code=0,
+                stdout="dir/file-12-name.py-8-context here\n",
+            )
+            result = ops._search_with_rg(
+                "needle",
+                path=".",
+                file_glob=None,
+                limit=10,
+                offset=0,
+                output_mode="content",
+                context=1,
+            )
+
+        assert result.error is None
+        assert result.total_count == 1
+        assert result.matches[0].path == "dir/file-12-name.py"
+        assert result.matches[0].line_number == 8
+        assert result.matches[0].content == "context here"
+
+    def test_search_with_grep_context_handles_filename_with_dash_digits(self):
+        env = MagicMock()
+        env.cwd = "/tmp"
+        ops = ShellFileOperations(env)
+
+        with patch.object(ops, "_exec") as mock_exec:
+            mock_exec.return_value = MagicMock(
+                exit_code=0,
+                stdout="dir/file-12-name.py-8-context here\n",
+            )
+            result = ops._search_with_grep(
+                "needle",
+                path=".",
+                file_glob=None,
+                limit=10,
+                offset=0,
+                output_mode="content",
+                context=1,
+            )
+
+        assert result.error is None
+        assert result.total_count == 1
+        assert result.matches[0].path == "dir/file-12-name.py"
+        assert result.matches[0].line_number == 8
+        assert result.matches[0].content == "context here"

tests/tools/test_tts_speed.py

@@ -110,7 +110,7 @@ class TestOpenaiTtsSpeed:
 
 
 # ---------------------------------------------------------------------------
-# MiniMax TTS speed (global fallback wired)
+# MiniMax TTS (new API: raw audio, no speed/voice_setting)
 # ---------------------------------------------------------------------------
 
 class TestMinimaxTtsSpeed:
@@ -118,28 +118,29 @@ class TestMinimaxTtsSpeed:
         monkeypatch.setenv("MINIMAX_API_KEY", "test-key")
         mock_response = MagicMock()
         mock_response.status_code = 200
-        mock_response.json.return_value = {
-            "data": {"audio": "deadbeef"},
-            "base_resp": {"status_code": 0, "status_msg": "success"},
-            "extra_info": {"audio_size": 8},
-        }
+        mock_response.headers = {"Content-Type": "audio/mpeg"}
+        mock_response.content = b"\x00\x01\x02\x03"
 
         # requests is imported locally inside _generate_minimax_tts
         with patch("requests.post", return_value=mock_response) as mock_post:
             from tools.tts_tool import _generate_minimax_tts
-            _generate_minimax_tts("Hello", str(tmp_path / "out.mp3"), tts_config)
-        return mock_post
+            output = _generate_minimax_tts("Hello", str(tmp_path / "out.mp3"), tts_config)
+        return mock_post, output
 
-    def test_global_speed_fallback(self, tmp_path, monkeypatch):
-        """Global tts.speed used when minimax.speed not set."""
-        mock_post = self._run({"speed": 1.5}, tmp_path, monkeypatch)
+    def test_simple_payload(self, tmp_path, monkeypatch):
+        """New API uses flat payload with model, text, voice_id."""
+        mock_post, _ = self._run({}, tmp_path, monkeypatch)
         payload = mock_post.call_args[1]["json"]
-        assert payload["voice_setting"]["speed"] == 1.5
+        assert "model" in payload
+        assert "text" in payload
+        assert "voice_id" in payload
+        assert "voice_setting" not in payload
+        assert "audio_setting" not in payload
+        assert "stream" not in payload
 
-    def test_provider_speed_overrides_global(self, tmp_path, monkeypatch):
-        """tts.minimax.speed takes precedence over tts.speed."""
-        mock_post = self._run(
-            {"speed": 1.5, "minimax": {"speed": 2.0}}, tmp_path, monkeypatch
-        )
-        payload = mock_post.call_args[1]["json"]
-        assert payload["voice_setting"]["speed"] == 2.0
+    def test_writes_raw_audio(self, tmp_path, monkeypatch):
+        """New API returns raw bytes written directly to file."""
+        _, output = self._run({}, tmp_path, monkeypatch)
+        assert output == str(tmp_path / "out.mp3")
+        with open(output, "rb") as f:
+            assert f.read() == b"\x00\x01\x02\x03"

tests/tui_gateway/test_entry_sys_path.py

@@ -0,0 +1,101 @@
+"""Tests for tui_gateway/entry.py sys.path hardening (issue #15989).
+
+When the TUI backend is spawned by Node.js, the Python interpreter may have
+'' or '.' at the front of sys.path, allowing a local utils/ directory in CWD
+to shadow the installed utils module.  entry.py must sanitize sys.path before
+any non-stdlib import is resolved.
+"""
+
+import importlib
+import os
+import sys
+from unittest.mock import patch
+
+
+def _reload_entry_with_env(env_overrides: dict) -> None:
+    """Re-execute entry.py's module-level path setup under a controlled env."""
+    # We only want to exercise the sys.path fixup block, not the signal/import
+    # machinery that follows.  We do this by running the fixup code verbatim in
+    # a fresh copy of sys.path rather than importing the real module (which
+    # would trigger tui_gateway.server imports requiring heavy mocks).
+    original_path = sys.path[:]
+    original_env = {k: os.environ.get(k) for k in env_overrides}
+    try:
+        with patch.dict(os.environ, env_overrides, clear=False):
+            _src_root = os.environ.get("HERMES_PYTHON_SRC_ROOT", "")
+            if _src_root and _src_root not in sys.path:
+                sys.path.insert(0, _src_root)
+            sys.path = [p for p in sys.path if p not in ("", ".")]
+        return sys.path[:]
+    finally:
+        sys.path = original_path
+        for k, v in original_env.items():
+            if v is None:
+                os.environ.pop(k, None)
+            else:
+                os.environ[k] = v
+
+
+def test_empty_string_and_dot_removed_from_sys_path():
+    original = sys.path[:]
+    try:
+        sys.path.insert(0, "")
+        sys.path.insert(0, ".")
+        assert "" in sys.path
+        assert "." in sys.path
+
+        # Run the entry.py fixup logic directly
+        sys.path = [p for p in sys.path if p not in ("", ".")]
+
+        assert "" not in sys.path
+        assert "." not in sys.path
+    finally:
+        sys.path = original
+
+
+def test_hermes_src_root_inserted_at_front():
+    original = sys.path[:]
+    try:
+        fake_root = "/fake/hermes/src"
+        with patch.dict(os.environ, {"HERMES_PYTHON_SRC_ROOT": fake_root}):
+            _src_root = os.environ.get("HERMES_PYTHON_SRC_ROOT", "")
+            if _src_root and _src_root not in sys.path:
+                sys.path.insert(0, _src_root)
+            sys.path = [p for p in sys.path if p not in ("", ".")]
+
+        assert sys.path[0] == fake_root
+    finally:
+        sys.path = original
+
+
+def test_src_root_not_duplicated_if_already_present():
+    original = sys.path[:]
+    try:
+        fake_root = "/already/present"
+        sys.path.insert(0, fake_root)
+        count_before = sys.path.count(fake_root)
+
+        with patch.dict(os.environ, {"HERMES_PYTHON_SRC_ROOT": fake_root}):
+            _src_root = os.environ.get("HERMES_PYTHON_SRC_ROOT", "")
+            if _src_root and _src_root not in sys.path:
+                sys.path.insert(0, _src_root)
+            sys.path = [p for p in sys.path if p not in ("", ".")]
+
+        assert sys.path.count(fake_root) == count_before
+    finally:
+        sys.path = original
+
+
+def test_no_src_root_env_does_not_crash():
+    original = sys.path[:]
+    try:
+        env = {k: v for k, v in os.environ.items() if k != "HERMES_PYTHON_SRC_ROOT"}
+        with patch.dict(os.environ, {}, clear=True):
+            os.environ.update(env)
+            _src_root = os.environ.get("HERMES_PYTHON_SRC_ROOT", "")
+            if _src_root and _src_root not in sys.path:
+                sys.path.insert(0, _src_root)
+            sys.path = [p for p in sys.path if p not in ("", ".")]
+        # No exception raised
+    finally:
+        sys.path = original