Merge origin/main into bb/gui

Adopt main's web/ dashboard layout (apps/dashboard removed; web/ restored),
keep bb/gui's desktop CLI/update workspace handling, and preserve main's
mTLS/URL validation MCP changes. Dashboard backend is aligned to main with
only the intended STT provider quarantine/ElevenLabs override reapplied.

tests/tools/test_accretion_caps.py

@@ -18,7 +18,6 @@ churn accumulated ~20B per session_id until the process exited.
 These tests pin the new caps + prune hooks.
 """
 
-import pytest
 
 
 class TestReadTrackerCaps:

tests/tools/test_approval.py

@@ -17,7 +17,6 @@ from tools.approval import (
     is_approved,
     load_permanent,
     prompt_dangerous_approval,
-    submit_pending,
 )
 
 

tests/tools/test_approval_heartbeat.py

@@ -12,9 +12,6 @@ between slices, mirroring ``_wait_for_process`` in ``tools/environments/base.py`
 """
 
 import os
-import threading
-import time
-from unittest.mock import patch
 
 
 def _clear_approval_state():

tests/tools/test_approval_plugin_hooks.py

@@ -13,9 +13,6 @@ import pytest
 import tools.approval as approval_module
 from tools.approval import (
     check_all_command_guards,
-    register_gateway_notify,
-    unregister_gateway_notify,
-    resolve_gateway_approval,
     set_current_session_key,
     clear_session,
 )

tests/tools/test_base_environment.py

@@ -4,10 +4,9 @@ Tests _wrap_command(), _extract_cwd_from_output(), _embed_stdin_heredoc(),
 init_session() failure handling, and the CWD marker contract.
 """
 
-import uuid
 from unittest.mock import MagicMock
 
-from tools.environments.base import BaseEnvironment, _cwd_marker
+from tools.environments.base import BaseEnvironment
 
 
 class _TestableEnv(BaseEnvironment):

tests/tools/test_browser_camofox.py

@@ -1,10 +1,8 @@
 """Tests for the Camofox browser backend."""
 
 import json
-import os
 from unittest.mock import MagicMock, patch
 
-import pytest
 
 from tools.browser_camofox import (
     camofox_back,
@@ -20,6 +18,7 @@ from tools.browser_camofox import (
     camofox_vision,
     check_camofox_available,
     is_camofox_mode,
+    _rewrite_loopback_url_for_camofox,
 )
 
 
@@ -59,6 +58,10 @@ class TestCamofoxMode:
 # ---------------------------------------------------------------------------
 
 
+def _config_with_camofox(**camofox_config):
+    return {"browser": {"camofox": camofox_config}}
+
+
 def _mock_response(status=200, json_data=None):
     resp = MagicMock()
     resp.status_code = status
@@ -73,6 +76,60 @@ def _mock_response(status=200, json_data=None):
 # ---------------------------------------------------------------------------
 
 
+class TestCamofoxLoopbackRewrite:
+    @patch("tools.browser_camofox.load_config")
+    def test_rewrites_localhost_when_enabled(self, mock_config, monkeypatch):
+        monkeypatch.delenv("CAMOFOX_REWRITE_LOOPBACK_URLS", raising=False)
+        monkeypatch.delenv("CAMOFOX_LOOPBACK_HOST_ALIAS", raising=False)
+        mock_config.return_value = _config_with_camofox(rewrite_loopback_urls=True)
+
+        rewritten, metadata = _rewrite_loopback_url_for_camofox("http://127.0.0.1:8766/#settings")
+
+        assert rewritten == "http://host.docker.internal:8766/#settings"
+        assert metadata == {
+            "from": "127.0.0.1",
+            "to": "host.docker.internal",
+            "original_url": "http://127.0.0.1:8766/#settings",
+            "rewritten_url": "http://host.docker.internal:8766/#settings",
+        }
+
+    @patch("tools.browser_camofox.load_config")
+    def test_rewrite_is_opt_in(self, mock_config, monkeypatch):
+        monkeypatch.delenv("CAMOFOX_REWRITE_LOOPBACK_URLS", raising=False)
+        mock_config.return_value = _config_with_camofox(rewrite_loopback_urls=False)
+
+        rewritten, metadata = _rewrite_loopback_url_for_camofox("http://localhost:3000/app?x=1")
+
+        assert rewritten == "http://localhost:3000/app?x=1"
+        assert metadata is None
+
+    @patch("tools.browser_camofox.load_config")
+    def test_preserves_public_urls_when_enabled(self, mock_config, monkeypatch):
+        monkeypatch.delenv("CAMOFOX_REWRITE_LOOPBACK_URLS", raising=False)
+        mock_config.return_value = _config_with_camofox(rewrite_loopback_urls=True)
+
+        rewritten, metadata = _rewrite_loopback_url_for_camofox("https://example.com:8443/path?q=1#top")
+
+        assert rewritten == "https://example.com:8443/path?q=1#top"
+        assert metadata is None
+
+    @patch("tools.browser_camofox.load_config")
+    def test_env_alias_takes_precedence(self, mock_config, monkeypatch):
+        monkeypatch.setenv("CAMOFOX_REWRITE_LOOPBACK_URLS", "true")
+        monkeypatch.setenv("CAMOFOX_LOOPBACK_HOST_ALIAS", "192.168.1.10")
+        mock_config.return_value = _config_with_camofox(
+            rewrite_loopback_urls=False,
+            loopback_host_alias="host.docker.internal",
+        )
+
+        rewritten, metadata = _rewrite_loopback_url_for_camofox("http://[::1]:8080/path")
+
+        assert rewritten == "http://192.168.1.10:8080/path"
+        assert metadata is not None
+        assert metadata["from"] == "::1"
+        assert metadata["to"] == "192.168.1.10"
+
+
 class TestCamofoxNavigate:
     @patch("tools.browser_camofox.requests.post")
     def test_creates_tab_on_first_navigate(self, mock_post, monkeypatch):
@@ -83,6 +140,24 @@ class TestCamofoxNavigate:
         assert result["success"] is True
         assert result["url"] == "https://example.com"
 
+    @patch("tools.browser_camofox.load_config")
+    @patch("tools.browser_camofox.requests.post")
+    def test_navigate_uses_rewritten_loopback_url(self, mock_post, mock_config, monkeypatch):
+        monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")
+        monkeypatch.delenv("CAMOFOX_REWRITE_LOOPBACK_URLS", raising=False)
+        monkeypatch.delenv("CAMOFOX_LOOPBACK_HOST_ALIAS", raising=False)
+        mock_config.return_value = _config_with_camofox(rewrite_loopback_urls=True)
+        mock_post.return_value = _mock_response(json_data={"tabId": "tab_rewrite"})
+
+        result = json.loads(camofox_navigate("http://127.0.0.1:8766/#settings", task_id="t_rewrite"))
+
+        assert result["success"] is True
+        assert result["url"] == "http://host.docker.internal:8766/#settings"
+        assert result["requested_url"] == "http://127.0.0.1:8766/#settings"
+        assert result["url_rewrite"]["to"] == "host.docker.internal"
+        assert "Rewrote loopback URL" in result["warning"]
+        assert mock_post.call_args.kwargs["json"]["url"] == "http://host.docker.internal:8766/#settings"
+
     @patch("tools.browser_camofox.requests.post")
     def test_navigates_existing_tab(self, mock_post, monkeypatch):
         monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377")

tests/tools/test_browser_camofox_state.py

@@ -2,7 +2,6 @@
 
 from unittest.mock import patch
 
-import pytest
 
 
 def _load_module():

tests/tools/test_browser_chromium_check.py

@@ -7,7 +7,6 @@ for the full command timeout before surfacing a useless error.
 """
 
 import os
-from pathlib import Path
 
 import pytest
 

tests/tools/test_browser_cloud_fallback.py

@@ -4,7 +4,7 @@ Covers the fallback logic in _get_session_info() when a cloud provider
 is configured but fails at runtime (issue #10883).
 """
 import logging
-from unittest.mock import Mock, patch
+from unittest.mock import Mock
 
 import pytest
 

tests/tools/test_browser_console.py

@@ -3,7 +3,6 @@
 import json
 import os
 import sys
-from pathlib import Path
 from unittest.mock import patch, MagicMock
 
 import pytest
@@ -251,6 +250,83 @@ class TestBrowserVisionConfig:
         assert mock_llm.call_args.kwargs["temperature"] == 0.1
         assert mock_llm.call_args.kwargs["timeout"] == 120.0
 
+    def test_browser_vision_native_fast_path_returns_multimodal(self, tmp_path):
+        """supports_vision override → screenshot attached natively, no aux call."""
+        from agent.auxiliary_client import clear_runtime_main, set_runtime_main
+        from tools.browser_tool import browser_vision
+
+        shots_dir, screenshot = self._setup_screenshot(tmp_path)
+        annotations = [{"id": 1, "label": "Search box"}]
+        set_runtime_main("brand-new-provider", "llava-v1.6")
+        try:
+            with (
+                patch("hermes_constants.get_hermes_dir", return_value=shots_dir),
+                patch("tools.browser_tool._cleanup_old_screenshots"),
+                patch(
+                    "tools.browser_tool._run_browser_command",
+                    return_value={
+                        "success": True,
+                        "data": {"path": str(screenshot), "annotations": annotations},
+                    },
+                ),
+                patch(
+                    "hermes_cli.config.load_config",
+                    return_value={"model": {"supports_vision": True}},
+                ),
+                patch("tools.browser_tool._get_vision_model") as mock_get_vision_model,
+                patch("tools.browser_tool.call_llm") as mock_llm,
+            ):
+                result = browser_vision("what is on the page?", annotate=True, task_id="test")
+        finally:
+            clear_runtime_main()
+
+        assert isinstance(result, dict)
+        assert result["_multimodal"] is True
+        assert result["meta"]["screenshot_path"] == str(screenshot)
+        assert result["meta"]["annotations"] == annotations
+        assert any(p.get("type") == "image_url" for p in result["content"])
+        assert f"Screenshot path: {screenshot}" in result["text_summary"]
+        mock_get_vision_model.assert_not_called()
+        mock_llm.assert_not_called()
+
+    def test_browser_vision_text_mode_blocks_native_fast_path(self, tmp_path):
+        """Explicit text routing → aux LLM used even with supports_vision."""
+        from agent.auxiliary_client import clear_runtime_main, set_runtime_main
+        from tools.browser_tool import browser_vision
+
+        shots_dir, screenshot = self._setup_screenshot(tmp_path)
+        mock_response = MagicMock()
+        mock_choice = MagicMock()
+        mock_choice.message.content = "Text-mode screenshot analysis"
+        mock_response.choices = [mock_choice]
+
+        set_runtime_main("brand-new-provider", "llava-v1.6")
+        try:
+            with (
+                patch("hermes_constants.get_hermes_dir", return_value=shots_dir),
+                patch("tools.browser_tool._cleanup_old_screenshots"),
+                patch(
+                    "tools.browser_tool._run_browser_command",
+                    return_value={"success": True, "data": {"path": str(screenshot)}},
+                ),
+                patch(
+                    "hermes_cli.config.load_config",
+                    return_value={
+                        "agent": {"image_input_mode": "text"},
+                        "model": {"supports_vision": True},
+                    },
+                ),
+                patch("tools.browser_tool._get_vision_model", return_value="test-model"),
+                patch("tools.browser_tool.call_llm", return_value=mock_response) as mock_llm,
+            ):
+                result = json.loads(browser_vision("what is on the page?", task_id="test"))
+        finally:
+            clear_runtime_main()
+
+        assert result["success"] is True
+        assert result["analysis"] == "Text-mode screenshot analysis"
+        mock_llm.assert_called_once()
+
 
 # ── auto-recording config ────────────────────────────────────────────
 

tests/tools/test_browser_content_none_guard.py

@@ -9,9 +9,8 @@ These tests verify both sites are guarded.
 """
 
 import types
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
-import pytest
 
 
 # ── helpers ────────────────────────────────────────────────────────────────

tests/tools/test_browser_eval_supervisor_path.py

@@ -8,7 +8,7 @@ real browser, no real WebSocket.  Real-CDP coverage lives in
 from __future__ import annotations
 
 import json
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
 
 import pytest
 

tests/tools/test_browser_hardening.py

@@ -1,7 +1,6 @@
 """Tests for browser_tool.py hardening: caching, security, thread safety, truncation."""
 
 import inspect
-import os
 from unittest.mock import MagicMock, patch
 
 import pytest

tests/tools/test_browser_homebrew_paths.py

@@ -2,7 +2,6 @@
 
 import json
 import os
-import subprocess
 from pathlib import Path
 from unittest.mock import patch, MagicMock, mock_open
 

tests/tools/test_browser_orphan_reaper.py

@@ -2,10 +2,7 @@
 daemons whose Python parent exited without cleaning up."""
 
 import os
-import signal
-import textwrap
-from pathlib import Path
-from unittest.mock import patch, MagicMock
+from unittest.mock import patch
 
 import pytest
 

tests/tools/test_browser_supervisor.py

@@ -17,7 +17,6 @@ from __future__ import annotations
 import asyncio
 import base64
 import json
-import os
 import shutil
 import subprocess
 import tempfile
@@ -48,7 +47,6 @@ def chrome_cdp(request):
     Always launches with ``--site-per-process`` so cross-origin iframes
     become real OOPIFs (needed by the iframe interaction tests).
     """
-    import socket
 
     # xdist worker_id is "master" in single-process mode or "gw0".."gwN" otherwise.
     # Under subprocess-per-file isolation there's no xdist, so we fall back
@@ -89,18 +87,45 @@ def chrome_cdp(request):
         except Exception:
             time.sleep(0.25)
     if ws_url is None:
-        proc.terminate()
-        proc.wait(timeout=5)
+        try:
+            proc.terminate()
+            proc.wait(timeout=5)
+        except (subprocess.TimeoutExpired, AssertionError, Exception):
+            try:
+                proc.kill()
+            except Exception:
+                pass
+            try:
+                proc.wait(timeout=2)
+            except (AssertionError, Exception):
+                pass
         shutil.rmtree(profile, ignore_errors=True)
         pytest.skip("Chrome didn't expose CDP in time")
 
     yield ws_url, port
 
-    proc.terminate()
+    # Tear down Chrome. The stdlib `subprocess._wait()` POSIX implementation
+    # has a known race (https://bugs.python.org/issue38630): when SIGCHLD
+    # arrives concurrently with `proc.wait()`, `_try_wait(WNOHANG)` can
+    # return a foreign pid and the `assert pid == self.pid or pid == 0`
+    # fires. We saw this in CI on slice 1 after this fixture's teardown
+    # (PR #33661 follow-up). Swallow the stdlib race + force-kill if wait
+    # hangs, then always reap so we don't leak a zombie.
+    try:
+        proc.terminate()
+    except Exception:
+        pass
     try:
         proc.wait(timeout=3)
-    except Exception:
-        proc.kill()
+    except (subprocess.TimeoutExpired, AssertionError, Exception):
+        try:
+            proc.kill()
+        except Exception:
+            pass
+        try:
+            proc.wait(timeout=2)
+        except (AssertionError, Exception):
+            pass
     shutil.rmtree(profile, ignore_errors=True)
 
 

tests/tools/test_checkpoint_manager.py

@@ -23,8 +23,6 @@ from tools.checkpoint_manager import (
     _project_meta_path,
     _touch_project,
     format_checkpoint_list,
-    DEFAULT_EXCLUDES,
-    CHECKPOINT_BASE,
     prune_checkpoints,
     maybe_auto_prune_checkpoints,
     store_status,

tests/tools/test_clarify_gateway.py

@@ -12,7 +12,6 @@ import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
 
-import pytest
 
 
 def _clear_clarify_state():

tests/tools/test_clarify_tool.py

@@ -3,7 +3,6 @@
 import json
 from typing import List, Optional
 
-import pytest
 
 from tools.clarify_tool import (
     clarify_tool,

tests/tools/test_clipboard.py

@@ -13,7 +13,7 @@ import queue
 import subprocess
 import sys
 from pathlib import Path
-from unittest.mock import patch, MagicMock, PropertyMock, mock_open
+from unittest.mock import patch, MagicMock, mock_open
 
 import pytest
 

tests/tools/test_code_execution.py

@@ -31,7 +31,6 @@ def _force_local_terminal(monkeypatch):
     """
     monkeypatch.setenv("TERMINAL_ENV", "local")
 import sys
-import time
 import threading
 import unittest
 from unittest.mock import patch, MagicMock

tests/tools/test_code_execution_windows_env.py

@@ -21,16 +21,13 @@ bytes.  The child then fails to import with a SyntaxError:
 """
 
 import os
-import socket
 import subprocess
 import sys
 import textwrap
-import unittest.mock as mock
 
 import pytest
 
 from tools.code_execution_tool import (
-    _SAFE_ENV_PREFIXES,
     _SECRET_SUBSTRINGS,
     _WINDOWS_ESSENTIAL_ENV_VARS,
     _scrub_child_env,
@@ -256,20 +253,24 @@ class TestWindowsSocketSmokeTest:
 # ---------------------------------------------------------------------------
 
 def _legacy_posix_scrubber(source_env, is_passthrough):
-    """Verbatim copy of the pre-Windows-fix inline scrubbing logic.
+    """Independent oracle for TestPosixEquivalence — a from-scratch reimpl of
+    _scrub_child_env's POSIX behavior, used to prove the production helper does
+    what we think it does.
 
-    This is the oracle used by TestPosixEquivalence to prove the refactor
-    did not change POSIX behavior.  DO NOT edit this to "match" a future
-    production change — if _scrub_child_env's POSIX behavior legitimately
-    needs to evolve, delete this function and adjust the equivalence test
-    on purpose, so the churn is visible in review.
+    Deliberately updated for #27303 (the broad ``HERMES_`` prefix was dropped
+    in favor of an explicit operational allowlist, and DSN/WEBHOOK were added
+    to the secret substrings).  The original docstring said: if POSIX behavior
+    legitimately needs to evolve, adjust this oracle on purpose so the churn is
+    visible in review — that is what this change is.
     """
     _SAFE_ENV_PREFIXES = ("PATH", "HOME", "USER", "LANG", "LC_", "TERM",
                           "TMPDIR", "TMP", "TEMP", "SHELL", "LOGNAME",
-                          "XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA",
-                          "HERMES_")
+                          "XDG_", "PYTHONPATH", "VIRTUAL_ENV", "CONDA")
     _SECRET_SUBSTRINGS = ("KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL",
-                          "PASSWD", "AUTH")
+                          "PASSWD", "AUTH", "DSN", "WEBHOOK")
+    _HERMES_CHILD_ALLOWED = frozenset({
+        "HERMES_HOME", "HERMES_PROFILE", "HERMES_CONFIG", "HERMES_ENV",
+    })
     out = {}
     for k, v in source_env.items():
         if is_passthrough(k):
@@ -279,6 +280,9 @@ def _legacy_posix_scrubber(source_env, is_passthrough):
             continue
         if any(k.startswith(p) for p in _SAFE_ENV_PREFIXES):
             out[k] = v
+            continue
+        if k in _HERMES_CHILD_ALLOWED:
+            out[k] = v
     return out
 
 
@@ -311,13 +315,20 @@ class TestPosixEquivalence:
         "PYTHONPATH": "/opt/lib",
         "VIRTUAL_ENV": "/home/alice/.venv",
         "CONDA_PREFIX": "/opt/conda",
-        "HERMES_HOME": "/home/alice/.hermes",
-        "HERMES_INTERACTIVE": "1",
+        # HERMES_* handling (#27303): only the operational allowlist passes;
+        # every other HERMES_* is dropped (the broad prefix was removed).
+        "HERMES_HOME": "/home/alice/.hermes",        # allowlisted → kept
+        "HERMES_PROFILE": "default",                 # allowlisted → kept
+        "HERMES_INTERACTIVE": "1",                   # not allowlisted → dropped
+        "HERMES_BASE_URL": "https://api.internal",   # not allowlisted → dropped
+        "HERMES_KANBAN_DB": "postgres://u:p@h/db",   # not allowlisted → dropped
         # Secret-substring blocks
         "OPENAI_API_KEY": "sk-xxx",
         "GITHUB_TOKEN": "ghp_xxx",
         "AWS_SECRET_ACCESS_KEY": "yyy",
         "MY_PASSWORD": "hunter2",
+        "SENTRY_DSN": "https://abc@sentry.io/1",     # DSN substring → blocked
+        "SLACK_WEBHOOK": "https://hooks.slack/x",    # WEBHOOK substring → blocked
         # Uncategorized — must be dropped
         "RANDOM_UNKNOWN": "drop-me",
         "DISPLAY": ":0",

tests/tools/test_computer_use.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import json
 import os
 import sys
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List
 from unittest.mock import MagicMock, patch
 
 import pytest

tests/tools/test_computer_use_capture_routing.py

@@ -24,8 +24,6 @@ from __future__ import annotations
 import base64
 import json
 import os
-from pathlib import Path
-from typing import Any
 from unittest.mock import MagicMock, patch
 
 import pytest

tests/tools/test_config_null_guard.py

@@ -6,7 +6,6 @@ return ``None`` instead of the default — calling ``.lower()`` on that raises
 """
 
 from unittest.mock import patch
-import pytest
 
 
 # ── TTS tool ──────────────────────────────────────────────────────────────

tests/tools/test_credential_files.py

@@ -1,6 +1,5 @@
 """Tests for credential file passthrough and skills directory mounting."""
 
-import json
 import os
 from pathlib import Path
 from unittest.mock import patch

tests/tools/test_cron_approval_mode.py

@@ -1,6 +1,5 @@
 """Tests for approvals.cron_mode — configurable approval behavior for cron jobs."""
 
-import os
 import pytest
 
 import tools.approval as approval_module

tests/tools/test_cronjob_tools.py

@@ -2,7 +2,6 @@
 
 import json
 import pytest
-from pathlib import Path
 
 from tools.cronjob_tools import (
     _scan_cron_prompt,

tests/tools/test_cross_profile_guard.py

@@ -13,7 +13,6 @@ This file tests that the tool surfaces:
 from __future__ import annotations
 
 import json
-import os
 from pathlib import Path
 
 import pytest

tests/tools/test_daytona_environment.py

@@ -2,7 +2,7 @@
 
 import threading
 from types import SimpleNamespace
-from unittest.mock import MagicMock, patch, PropertyMock
+from unittest.mock import MagicMock
 
 import pytest
 

tests/tools/test_delegate.py

@@ -11,7 +11,6 @@ Run with:  python -m pytest tests/test_delegate.py -v
 
 import json
 import os
-import sys
 import threading
 import time
 import unittest

tests/tools/test_delegate_composite_toolsets.py

@@ -1,7 +1,6 @@
 """Tests for composite toolset expansion in delegate_task intersection."""
 
 import unittest
-from unittest.mock import patch
 
 from tools.delegate_tool import _expand_parent_toolsets
 

tests/tools/test_delegate_subagent_timeout_diagnostic.py

@@ -16,12 +16,10 @@ These tests pin:
 """
 from __future__ import annotations
 
-import os
 import threading
 import time
 from pathlib import Path
-from typing import Optional
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
 
 import pytest
 

tests/tools/test_delegate_toolset_scope.py

@@ -6,7 +6,6 @@ parent's enabled_toolsets, it can escalate privileges by requesting
 arbitrary toolsets.
 """
 
-from unittest.mock import MagicMock, patch
 from types import SimpleNamespace
 
 from tools.delegate_tool import _strip_blocked_tools

tests/tools/test_discord_tool.py

@@ -1,7 +1,6 @@
 """Tests for the Discord server introspection and management tool."""
 
 import json
-import os
 import urllib.error
 from io import BytesIO
 from unittest.mock import MagicMock, patch
@@ -14,7 +13,6 @@ from tools.discord_tool import (
     _ADMIN_ACTIONS,
     _CORE_ACTIONS,
     _available_actions,
-    _build_schema,
     _channel_type_name,
     _detect_capabilities,
     _discord_request,

tests/tools/test_docker_orphan_reaper_integration.py

@@ -0,0 +1,139 @@
+"""Integration tests for the docker orphan-reaper wiring in terminal_tool.
+
+The reaper itself is unit-tested in tests/tools/test_docker_environment.py
+under the "Orphan reaper" section. These tests cover the terminal_tool-side
+gates: once-per-process behavior, the disable flag, and the
+``lifetime_seconds`` doubling that determines the reaper's age threshold.
+
+Issue #20561 — without these gates, parallel subagents would each fire the
+reaper on container creation, and the ``terminal.docker_orphan_reaper: false``
+opt-out would silently do nothing.
+"""
+
+import os
+from unittest.mock import patch
+
+import tools.terminal_tool as terminal_tool
+
+
+def _reset_reaper_gate():
+    """Clear the once-per-process flag between tests."""
+    terminal_tool._docker_orphan_reaper_ran = False
+
+
+def test_maybe_reap_runs_once_per_process(monkeypatch):
+    """The reaper sweep must run at most once per Python interpreter.
+    Parallel subagents that each call _create_environment(env_type='docker')
+    would otherwise fire N concurrent docker ps + inspect storms against the
+    daemon and waste 5–10s of startup."""
+    _reset_reaper_gate()
+    call_count = {"reap": 0}
+
+    def _fake_reap(**kwargs):
+        call_count["reap"] += 1
+        return 0
+
+    with patch("tools.environments.docker.reap_orphan_containers", _fake_reap):
+        config = {"docker_orphan_reaper": True}
+        terminal_tool._maybe_reap_docker_orphans(config)
+        terminal_tool._maybe_reap_docker_orphans(config)
+        terminal_tool._maybe_reap_docker_orphans(config)
+
+    assert call_count["reap"] == 1, (
+        f"reaper must run exactly once per process; got {call_count['reap']} calls"
+    )
+
+
+def test_maybe_reap_respects_disable_flag(monkeypatch):
+    """``terminal.docker_orphan_reaper: false`` (via container_config) must
+    skip the sweep entirely — no docker ps, no inspect, no rm. The escape
+    hatch for operators running multiple Hermes processes in the same
+    profile."""
+    _reset_reaper_gate()
+    call_count = {"reap": 0}
+
+    def _fake_reap(**kwargs):
+        call_count["reap"] += 1
+        return 0
+
+    with patch("tools.environments.docker.reap_orphan_containers", _fake_reap):
+        terminal_tool._maybe_reap_docker_orphans({"docker_orphan_reaper": False})
+
+    assert call_count["reap"] == 0, "disabled reaper must not run any docker calls"
+    # The once-per-process gate must NOT be tripped when the reaper is
+    # disabled — that would prevent a subsequent toggle to true from working.
+    assert terminal_tool._docker_orphan_reaper_ran is False
+
+
+def test_maybe_reap_doubles_lifetime_for_max_age(monkeypatch):
+    """The reaper's age threshold is ``2 × lifetime_seconds`` (with a 60s
+    floor). Generous default — gives sibling Hermes processes ample grace
+    to be replaced without their just-exited containers being yanked."""
+    _reset_reaper_gate()
+    captured_args = {}
+
+    def _fake_reap(**kwargs):
+        captured_args.update(kwargs)
+        return 0
+
+    monkeypatch.setenv("TERMINAL_LIFETIME_SECONDS", "300")
+    with patch("tools.environments.docker.reap_orphan_containers", _fake_reap):
+        terminal_tool._maybe_reap_docker_orphans({"docker_orphan_reaper": True})
+
+    assert captured_args.get("max_age_seconds") == 600, (
+        f"expected 2 × 300 = 600, got {captured_args.get('max_age_seconds')}"
+    )
+
+
+def test_maybe_reap_floors_at_60_seconds(monkeypatch):
+    """A user pinning TERMINAL_LIFETIME_SECONDS=0 (or any value <30) would
+    otherwise get an effective age threshold of zero, which would race the
+    user's own just-started container creation. Floor at 60s × 2 = 120s."""
+    _reset_reaper_gate()
+    captured_args = {}
+
+    def _fake_reap(**kwargs):
+        captured_args.update(kwargs)
+        return 0
+
+    monkeypatch.setenv("TERMINAL_LIFETIME_SECONDS", "0")
+    with patch("tools.environments.docker.reap_orphan_containers", _fake_reap):
+        terminal_tool._maybe_reap_docker_orphans({"docker_orphan_reaper": True})
+
+    assert captured_args.get("max_age_seconds") == 120, (
+        f"expected floored 60 × 2 = 120, got {captured_args.get('max_age_seconds')}"
+    )
+
+
+def test_maybe_reap_passes_current_profile_as_filter(monkeypatch):
+    """The reaper must be scoped to the current Hermes profile — a research
+    profile must NEVER reap default's containers. Verifies the
+    profile-filter wiring."""
+    _reset_reaper_gate()
+    captured_args = {}
+
+    def _fake_reap(**kwargs):
+        captured_args.update(kwargs)
+        return 0
+
+    with patch("tools.environments.docker.reap_orphan_containers", _fake_reap), \
+         patch("tools.environments.docker._get_active_profile_name", return_value="research-bot"):
+        terminal_tool._maybe_reap_docker_orphans({"docker_orphan_reaper": True})
+
+    assert captured_args.get("profile_filter") == "research-bot", (
+        f"expected profile_filter='research-bot', got {captured_args.get('profile_filter')!r}"
+    )
+
+
+def test_maybe_reap_swallows_exceptions(monkeypatch):
+    """A reaper crash (docker daemon down, parse error in helper) must NOT
+    block env creation. The reaper is best-effort plumbing, not a critical
+    path; failures get logged at debug level and execution continues."""
+    _reset_reaper_gate()
+
+    def _exploding_reap(**kwargs):
+        raise RuntimeError("docker daemon ate the cat")
+
+    with patch("tools.environments.docker.reap_orphan_containers", _exploding_reap):
+        # Must not raise
+        terminal_tool._maybe_reap_docker_orphans({"docker_orphan_reaper": True})

tests/tools/test_env_passthrough.py

@@ -198,7 +198,6 @@ class TestTerminalIntegration:
     def test_make_run_env_blocklist_override_rejected(self):
         """_make_run_env must NOT expose a blocklisted var to subprocess env
         even after a skill attempts to register it via passthrough."""
-        import os
         from tools.environments.local import (
             _make_run_env,
             _HERMES_PROVIDER_ENV_BLOCKLIST,

tests/tools/test_env_probe.py

@@ -0,0 +1,157 @@
+"""Tests for tools/env_probe.py — local Python toolchain probe."""
+
+import sys
+
+import pytest
+
+from tools import env_probe
+
+
+@pytest.fixture(autouse=True)
+def reset_probe_cache():
+    """Each test starts with a clean cache."""
+    env_probe._reset_cache_for_tests()
+    yield
+    env_probe._reset_cache_for_tests()
+
+
+class TestSilentWhenHealthy:
+    """The probe must emit nothing when the environment is clean — otherwise
+    every prompt for every user pays an unnecessary token tax."""
+
+    def test_clean_env_returns_empty(self, monkeypatch):
+        """python3 + pip module + no PEP 668 → silent."""
+        monkeypatch.setattr(env_probe, "_python_version_of",
+                            lambda b: "3.13.3" if b == "python3" else None)
+        monkeypatch.setattr(env_probe, "_has_pip_module", lambda b: True)
+        monkeypatch.setattr(env_probe, "_detect_pep668", lambda b: False)
+        monkeypatch.setattr(env_probe, "_pip_python_version", lambda: "3.13")
+        monkeypatch.setattr(env_probe.shutil, "which", lambda name: None)
+        assert env_probe.get_environment_probe_line() == ""
+
+    def test_pep668_with_uv_returns_empty(self, monkeypatch):
+        """PEP 668 alone shouldn't trigger output if uv is installed —
+        agent has a viable install path."""
+        monkeypatch.setattr(env_probe, "_python_version_of",
+                            lambda b: "3.12.4" if b == "python3" else None)
+        monkeypatch.setattr(env_probe, "_has_pip_module", lambda b: True)
+        monkeypatch.setattr(env_probe, "_detect_pep668", lambda b: True)
+        monkeypatch.setattr(env_probe, "_pip_python_version", lambda: "3.12")
+        monkeypatch.setattr(env_probe.shutil, "which",
+                            lambda name: "/usr/local/bin/uv" if name == "uv" else None)
+        assert env_probe.get_environment_probe_line() == ""
+
+
+class TestEmitsOnRealProblems:
+    """The probe must produce a usable line for the real failure modes
+    that drove this feature."""
+
+    def test_allen_scenario_python_version_mismatch(self, monkeypatch):
+        """python3 is 3.11 (no pip module), pip on PATH is 3.12, PEP 668 on,
+        no uv — the exact scenario from the Sarasota real-estate task."""
+        monkeypatch.setattr(env_probe, "_python_version_of",
+                            lambda b: {"python3": "3.11.15", "python": None}.get(b))
+        monkeypatch.setattr(env_probe, "_has_pip_module", lambda b: False)
+        monkeypatch.setattr(env_probe, "_detect_pep668", lambda b: True)
+        monkeypatch.setattr(env_probe, "_pip_python_version", lambda: "3.12")
+        monkeypatch.setattr(env_probe.shutil, "which",
+                            lambda name: None if name == "uv" else "/usr/bin/" + name)
+
+        line = env_probe.get_environment_probe_line()
+        assert line  # not silent
+        # Single line — must not blow up the system prompt.
+        assert "\n" not in line
+        # Names the real toolchain state
+        assert "3.11.15" in line
+        assert "no pip module" in line
+        assert "mismatch" in line
+        assert "PEP 668" in line
+        # Points at the right escape hatch
+        assert "venv" in line or "uv" in line
+
+    def test_missing_python3_is_named(self, monkeypatch):
+        """If python3 isn't installed at all, say so."""
+        monkeypatch.setattr(env_probe, "_python_version_of", lambda b: None)
+        monkeypatch.setattr(env_probe, "_has_pip_module", lambda b: False)
+        monkeypatch.setattr(env_probe, "_detect_pep668", lambda b: False)
+        monkeypatch.setattr(env_probe, "_pip_python_version", lambda: None)
+        monkeypatch.setattr(env_probe.shutil, "which", lambda name: None)
+
+        line = env_probe.get_environment_probe_line()
+        assert "python3=missing" in line
+
+    def test_python_missing_but_python3_present(self, monkeypatch):
+        """Common on Debian: only python3 exists, agent shouldn't type
+        `python`."""
+        monkeypatch.setattr(env_probe, "_python_version_of",
+                            lambda b: "3.12.4" if b == "python3" else None)
+        monkeypatch.setattr(env_probe, "_has_pip_module", lambda b: True)
+        monkeypatch.setattr(env_probe, "_detect_pep668", lambda b: True)
+        monkeypatch.setattr(env_probe, "_pip_python_version", lambda: "3.12")
+        monkeypatch.setattr(env_probe.shutil, "which",
+                            lambda name: None if name == "uv" else "/usr/bin/" + name)
+
+        line = env_probe.get_environment_probe_line()
+        # `python=missing` only matters in the non-silent path; PEP 668 (without
+        # uv) is what brings us off-silent here, so check both signals.
+        assert "PEP 668" in line
+        assert "python=missing" in line
+
+
+class TestSkipsRemoteBackends:
+    """Remote backends have their own probe; this one must stay out."""
+
+    def test_docker_returns_empty(self, monkeypatch):
+        monkeypatch.setenv("TERMINAL_ENV", "docker")
+        # Even with a broken local env, docker must emit nothing.
+        monkeypatch.setattr(env_probe, "_python_version_of", lambda b: None)
+        monkeypatch.setattr(env_probe, "_has_pip_module", lambda b: False)
+        assert env_probe.get_environment_probe_line() == ""
+
+    def test_modal_returns_empty(self, monkeypatch):
+        monkeypatch.setenv("TERMINAL_ENV", "modal")
+        assert env_probe.get_environment_probe_line() == ""
+
+    def test_ssh_returns_empty(self, monkeypatch):
+        monkeypatch.setenv("TERMINAL_ENV", "ssh")
+        assert env_probe.get_environment_probe_line() == ""
+
+
+class TestCaching:
+    """The probe runs once per process — the result is deterministic for
+    the lifetime of the agent."""
+
+    def test_result_cached(self, monkeypatch):
+        calls = []
+
+        def counting_version(b):
+            calls.append(b)
+            return "3.12.4" if b == "python3" else None
+
+        monkeypatch.setattr(env_probe, "_python_version_of", counting_version)
+        monkeypatch.setattr(env_probe, "_has_pip_module", lambda b: True)
+        monkeypatch.setattr(env_probe, "_detect_pep668", lambda b: False)
+        monkeypatch.setattr(env_probe, "_pip_python_version", lambda: "3.12")
+        monkeypatch.setattr(env_probe.shutil, "which", lambda name: None)
+
+        env_probe.get_environment_probe_line()
+        env_probe.get_environment_probe_line()
+        env_probe.get_environment_probe_line()
+
+        # Only the first call probes — caller-counting confirms it.
+        # Two calls (python3 + python) on first invocation, zero after.
+        assert len(calls) == 2
+
+
+class TestRobustness:
+    """The probe must NEVER crash the prompt build."""
+
+    def test_subprocess_failure_returns_empty(self, monkeypatch):
+        """If every subprocess fails, just stay silent."""
+        def boom(*a, **kw):
+            raise OSError("simulated")
+        monkeypatch.setattr(env_probe.subprocess, "run", boom)
+        # Should not raise, should just return ""
+        result = env_probe.get_environment_probe_line()
+        # Whatever the result is, it must be a string
+        assert isinstance(result, str)

tests/tools/test_execute_code_approval_cluster.py

@@ -0,0 +1,349 @@
+"""Regression tests for the execute_code approval-bypass cluster.
+
+Covers the canonical fix for issues #4146, #27303, #30882, #33057:
+
+  1. tools.thread_context.propagate_context_to_thread — propagates the agent
+     turn's ContextVars AND thread-local approval/sudo callbacks into worker
+     threads, and clears the callbacks on teardown.
+  2. Both execute_code RPC threads are wrapped with that helper (source guard).
+  3. tools.approval.check_execute_code_guard — the entry-point guard decision
+     matrix (isolated backends, yolo/off, cron-deny, headless-local,
+     gateway approve/deny/timeout/missing-notify, smart mode).
+  4. tools.code_execution_tool._scrub_child_env — broad HERMES_ prefix dropped,
+     operational allowlist kept, DSN/WEBHOOK blocked, passthrough precedence.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures
+import contextvars
+import threading
+
+import pytest
+
+from tools import approval as A
+from tools.thread_context import propagate_context_to_thread
+
+
+# ---------------------------------------------------------------------------
+# 1. Context + callback propagation helper
+# ---------------------------------------------------------------------------
+
+def test_helper_propagates_contextvar_and_approval_callback():
+    from tools import terminal_tool as TT
+
+    probe: contextvars.ContextVar[str] = contextvars.ContextVar(
+        "cluster_probe", default="unset"
+    )
+    probe.set("parent-value")
+    sentinel = object()
+    TT.set_approval_callback(sentinel)
+    try:
+        seen: dict = {}
+
+        def worker():
+            seen["probe"] = probe.get()
+            seen["cb"] = TT._get_approval_callback()
+
+        t = threading.Thread(target=propagate_context_to_thread(worker))
+        t.start()
+        t.join(timeout=5)
+
+        assert seen["probe"] == "parent-value"  # ContextVar propagated
+        assert seen["cb"] is sentinel            # thread-local callback propagated
+    finally:
+        TT.set_approval_callback(None)
+
+
+def test_helper_clears_callbacks_on_teardown():
+    """A recycled worker thread must not retain the propagated callback after
+    the wrapped target finishes (mirrors the GHSA-qg5c-hvr5-hjgr teardown)."""
+    from tools import terminal_tool as TT
+
+    sentinel = object()
+    TT.set_approval_callback(sentinel)
+    try:
+        seen: dict = {}
+
+        def first():
+            seen["during"] = TT._get_approval_callback()
+
+        def second():  # NOT wrapped — runs on the same recycled worker thread
+            seen["after"] = TT._get_approval_callback()
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
+            ex.submit(propagate_context_to_thread(first)).result(timeout=5)
+            ex.submit(second).result(timeout=5)
+
+        assert seen["during"] is sentinel  # installed for the wrapped target
+        assert seen["after"] is None       # cleared on teardown
+    finally:
+        TT.set_approval_callback(None)
+
+
+def test_both_rpc_threads_use_propagation_helper():
+    """Source guard: both execute_code RPC threads must wrap their target with
+    propagate_context_to_thread, or the gateway approval bypass (#33057)
+    silently returns."""
+    import inspect
+    import tools.code_execution_tool as cet
+
+    src = inspect.getsource(cet)
+    assert "propagate_context_to_thread(_rpc_server_loop)" in src, (
+        "local UDS RPC server thread is not wrapped with "
+        "propagate_context_to_thread — gateway approval routing will be lost."
+    )
+    assert "propagate_context_to_thread(_rpc_poll_loop)" in src, (
+        "remote file-RPC poll thread is not wrapped with "
+        "propagate_context_to_thread — gateway approval routing will be lost."
+    )
+
+
+# ---------------------------------------------------------------------------
+# 3. check_execute_code_guard decision matrix
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def gw_session(monkeypatch):
+    """A clean gateway session: HERMES_GATEWAY_SESSION set, a bound session
+    key, and isolated gateway queues/callbacks. Yields the session_key."""
+    monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
+    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+    monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
+    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+    # Force manual mode regardless of host config.
+    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
+
+    session_key = "cluster-test-session"
+    token = A.set_current_session_key(session_key)
+    with A._lock:
+        A._gateway_queues.pop(session_key, None)
+        A._gateway_notify_cbs.pop(session_key, None)
+    try:
+        yield session_key
+    finally:
+        A.reset_current_session_key(token)
+        with A._lock:
+            A._gateway_queues.pop(session_key, None)
+            A._gateway_notify_cbs.pop(session_key, None)
+
+
+def _register_resolver(session_key: str, result):
+    """Register a gateway notify callback that immediately resolves the most
+    recent queued approval entry with *result* (simulating a user response)."""
+    def cb(_approval_data):
+        with A._lock:
+            entries = A._gateway_queues.get(session_key, [])
+            if entries:
+                entry = entries[-1]
+                entry.result = result
+                entry.event.set()
+    with A._lock:
+        A._gateway_notify_cbs[session_key] = cb
+
+
+def test_guard_isolated_backend_approved():
+    # Container backends already sandbox the child — no-op approve.
+    assert A.check_execute_code_guard("import os", "docker")["approved"] is True
+
+
+def test_guard_headless_local_approved(monkeypatch):
+    # Documented #30882 limitation: no approval surface → preserve auto-run.
+    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+    monkeypatch.delenv("HERMES_CRON_SESSION", raising=False)
+    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
+    assert A.check_execute_code_guard("import os", "local")["approved"] is True
+
+
+def test_guard_cron_deny_blocks(monkeypatch):
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
+    monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny")
+    res = A.check_execute_code_guard("import os", "local")
+    assert res["approved"] is False
+    assert res["outcome"] == "blocked"
+
+
+def test_guard_gateway_user_approves_is_one_shot(gw_session):
+    _register_resolver(gw_session, "once")
+    res = A.check_execute_code_guard("import os; print(1)", "local")
+    assert res["approved"] is True
+    assert res.get("user_approved") is True
+    # One-shot: approval must NOT persist to future scripts.
+    assert A.is_approved(gw_session, "execute_code") is False
+
+
+def test_guard_gateway_user_denies_blocks(gw_session):
+    _register_resolver(gw_session, "deny")
+    res = A.check_execute_code_guard("import os", "local")
+    assert res["approved"] is False
+    assert res["outcome"] == "denied"
+    assert res["user_consent"] is False
+
+
+def test_guard_gateway_timeout_blocks(gw_session, monkeypatch):
+    # Register a callback that never resolves; force an immediate timeout.
+    with A._lock:
+        A._gateway_notify_cbs[gw_session] = lambda _d: None
+    monkeypatch.setattr(A, "_get_approval_config", lambda: {"gateway_timeout": 0})
+    res = A.check_execute_code_guard("import os", "local")
+    assert res["approved"] is False
+    assert res["outcome"] == "timeout"
+
+
+def test_guard_gateway_missing_notify_is_pending(gw_session):
+    # No notify callback registered → backward-compat pending approval.
+    res = A.check_execute_code_guard("import os", "local")
+    assert res["approved"] is False
+    assert res["status"] == "pending_approval"
+
+
+def test_guard_smart_mode(gw_session, monkeypatch):
+    monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
+
+    monkeypatch.setattr(A, "_smart_approve", lambda c, d: "approve")
+    res = A.check_execute_code_guard("import os", "local")
+    assert res["approved"] is True and res.get("smart_approved") is True
+
+    monkeypatch.setattr(A, "_smart_approve", lambda c, d: "deny")
+    res = A.check_execute_code_guard("import os", "local")
+    assert res["approved"] is False and res.get("smart_denied") is True
+
+    # escalate → falls through to manual gateway approval
+    monkeypatch.setattr(A, "_smart_approve", lambda c, d: "escalate")
+    _register_resolver(gw_session, "once")
+    res = A.check_execute_code_guard("import os", "local")
+    assert res["approved"] is True
+
+
+def test_guard_session_yolo_bypasses(gw_session):
+    A.enable_session_yolo(gw_session)
+    try:
+        # Even with a denier registered, yolo short-circuits before the prompt.
+        _register_resolver(gw_session, "deny")
+        assert A.check_execute_code_guard("import os", "local")["approved"] is True
+    finally:
+        A.disable_session_yolo(gw_session)
+
+
+# ---------------------------------------------------------------------------
+# 4. Env scrubbing (#27303)
+# ---------------------------------------------------------------------------
+
+def test_env_scrub_hermes_allowlist_and_secret_blocks():
+    from tools.code_execution_tool import _scrub_child_env
+
+    env = {
+        # operational allowlist → kept
+        "HERMES_HOME": "/h", "HERMES_PROFILE": "p",
+        "HERMES_CONFIG": "/c.yaml", "HERMES_ENV": "/e",
+        # other HERMES_* → dropped (broad prefix removed)
+        "HERMES_BASE_URL": "https://x", "HERMES_INTERACTIVE": "1",
+        "HERMES_KANBAN_DB": "postgres://u:p@h/db",
+        # secret substrings (incl. new DSN/WEBHOOK) → dropped
+        "SENTRY_DSN": "https://a@s.io/1", "SLACK_WEBHOOK": "https://h/x",
+        "OPENAI_API_KEY": "sk", "GITHUB_TOKEN": "ghp",
+        # safe prefix → kept; uncategorized → dropped
+        "PATH": "/usr/bin", "RANDOM_X": "y",
+    }
+    out = _scrub_child_env(env, is_passthrough=lambda _: False, is_windows=False)
+
+    for kept in ("HERMES_HOME", "HERMES_PROFILE", "HERMES_CONFIG", "HERMES_ENV", "PATH"):
+        assert kept in out, f"{kept} should be kept"
+    for dropped in (
+        "HERMES_BASE_URL", "HERMES_INTERACTIVE", "HERMES_KANBAN_DB",
+        "SENTRY_DSN", "SLACK_WEBHOOK", "OPENAI_API_KEY", "GITHUB_TOKEN",
+        "RANDOM_X",
+    ):
+        assert dropped not in out, f"{dropped} should be dropped"
+
+
+def test_env_scrub_passthrough_overrides_secret_block():
+    """A skill/config-declared passthrough var is an explicit user opt-in and
+    passes even if it matches a secret substring (precedence is intentional)."""
+    from tools.code_execution_tool import _scrub_child_env
+
+    env = {"MY_SERVICE_DSN": "value"}
+    out = _scrub_child_env(env, is_passthrough=lambda k: k == "MY_SERVICE_DSN",
+                           is_windows=False)
+    assert out.get("MY_SERVICE_DSN") == "value"
+
+
+# ---------------------------------------------------------------------------
+# 5. File-tool sensitive-path refusal (security B1)
+# ---------------------------------------------------------------------------
+
+def test_execute_code_entry_blocks_before_spawn_when_guard_denies(monkeypatch, tmp_path):
+    """Behavioral wiring test: execute_code() consults the entry guard and, on
+    denial, returns the block message WITHOUT spawning the child — proven by a
+    marker file the script would create that never appears."""
+    import json
+
+    import tools.code_execution_tool as cet
+    from tools import terminal_tool as TT
+
+    marker = tmp_path / "child-ran.marker"
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+    monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
+    monkeypatch.setattr(A, "_get_cron_approval_mode", lambda: "deny")
+    monkeypatch.setattr(TT, "_get_env_config", lambda: {"env_type": "local"})
+
+    result = json.loads(
+        cet.execute_code(f"open({str(marker)!r}, 'w').close()", task_id="cluster-t")
+    )
+    assert result["status"] == "error"
+    assert "BLOCKED" in result["error"]
+    assert not marker.exists()  # guard denied before the child was spawned
+
+
+# ---------------------------------------------------------------------------
+# 6. Env-scrub diagnosability mitigation (#27303 follow-up)
+# ---------------------------------------------------------------------------
+
+def test_env_scrub_logs_dropped_hermes_vars(caplog):
+    """Dropping a non-allowlisted, non-secret HERMES_* var must be diagnosable:
+    the scrub emits a one-shot debug log naming the dropped vars and pointing at
+    the env_passthrough opt-in, so the silent behavior change (#27303) doesn't
+    leave users guessing why a sandbox script sees an unset HERMES_* var."""
+    import logging
+
+    from tools.code_execution_tool import _scrub_child_env
+
+    env = {
+        "HERMES_HOME": "/h",          # allowlisted → kept, not logged
+        "HERMES_BASE_URL": "https://x",   # dropped → logged
+        "HERMES_KANBAN_DB": "postgres://u:p@h/db",  # dropped → logged
+        "HERMES_API_KEY": "sk",       # secret → dropped silently (not logged)
+        "PATH": "/usr/bin",           # safe prefix → kept
+    }
+    with caplog.at_level(logging.DEBUG, logger="tools.code_execution_tool"):
+        out = _scrub_child_env(env, is_passthrough=lambda _: False, is_windows=False)
+
+    assert "HERMES_HOME" in out and "PATH" in out
+    assert "HERMES_BASE_URL" not in out and "HERMES_KANBAN_DB" not in out
+
+    msgs = "\n".join(r.getMessage() for r in caplog.records)
+    assert "HERMES_BASE_URL" in msgs and "HERMES_KANBAN_DB" in msgs
+    assert "env_passthrough" in msgs
+    # Secret vars are dropped but must NOT be named in the diagnostic log.
+    assert "HERMES_API_KEY" not in msgs
+
+
+def test_env_scrub_no_log_when_nothing_dropped(caplog):
+    """No diagnostic noise when there are no dropped HERMES_* vars."""
+    import logging
+
+    from tools.code_execution_tool import _scrub_child_env
+
+    with caplog.at_level(logging.DEBUG, logger="tools.code_execution_tool"):
+        _scrub_child_env(
+            {"HERMES_HOME": "/h", "PATH": "/usr/bin"},
+            is_passthrough=lambda _: False,
+            is_windows=False,
+        )
+    assert "dropped" not in "\n".join(r.getMessage() for r in caplog.records)

tests/tools/test_file_operations.py

@@ -8,8 +8,6 @@ from unittest.mock import MagicMock
 
 from tools.file_operations import (
     _is_write_denied,
-    WRITE_DENIED_PATHS,
-    WRITE_DENIED_PREFIXES,
     ReadResult,
     WriteResult,
     PatchResult,
@@ -17,8 +15,6 @@ from tools.file_operations import (
     SearchMatch,
     LintResult,
     ShellFileOperations,
-    BINARY_EXTENSIONS,
-    IMAGE_EXTENSIONS,
     MAX_LINE_LENGTH,
     normalize_read_pagination,
     normalize_search_pagination,

tests/tools/test_file_ops_cwd_tracking.py

@@ -17,10 +17,7 @@ Fix: _exec() now prefers the LIVE ``env.cwd`` over the init-time
 
 from __future__ import annotations
 
-import os
-import tempfile
 
-import pytest
 
 from tools.file_operations import ShellFileOperations
 

tests/tools/test_file_read_guards.py

@@ -21,7 +21,6 @@ from tools.file_tools import (
     _is_blocked_device,
     _invalidate_dedup_for_path,
     _READ_DEDUP_STATUS_MESSAGE,
-    _get_max_read_chars,
     _DEFAULT_MAX_READ_CHARS,
     _read_tracker,
     notify_other_tool_call,

tests/tools/test_file_sync_back.py

@@ -5,9 +5,8 @@ import logging
 import os
 import signal
 import tarfile
-import time
 from pathlib import Path
-from unittest.mock import MagicMock, call, patch
+from unittest.mock import MagicMock, patch
 
 import pytest
 

tests/tools/test_file_tools.py

@@ -9,10 +9,7 @@ import logging
 from unittest.mock import MagicMock, patch
 
 from tools.file_tools import (
-    READ_FILE_SCHEMA,
-    WRITE_FILE_SCHEMA,
     PATCH_SCHEMA,
-    SEARCH_FILES_SCHEMA,
 )
 
 

tests/tools/test_file_tools_live.py

@@ -13,12 +13,10 @@ import pytest
 
 
 
-import json
 import os
 import sys
 from pathlib import Path
 
-import pytest
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
 
@@ -388,7 +386,6 @@ class TestExpandPath:
         # The path should be returned as-is (no expansion).
         assert result == malicious
         # Verify the injected command did NOT execute
-        import os
         assert not os.path.exists("/tmp/_hermes_injection_test")
 
     def test_tilde_username_with_subpath(self, ops):

tests/tools/test_fuzzy_match.py

@@ -429,3 +429,118 @@ class TestFormatNoMatchHint:
         )
         assert result == ""
 
+
+class TestEscapeNormalizedNewString:
+    """Regression tests for unescaping common sequences in new_string when
+    the matched region of the file contains real control characters.
+
+    Issue #33733: LLMs overwhelmingly represent tabs as the two-character
+    sequence ``\\t`` (backslash + t) in JSON tool-call arguments. When the
+    file already contains real tab bytes (0x09), writing new_string
+    verbatim leaves literal ``\\t`` characters and corrupts the file.
+
+    The fix unescapes ``\\t`` -> tab and ``\\r`` -> CR in new_string when
+    the matched file region actually contains those control characters,
+    regardless of which match strategy fired. ``\\n`` is excluded because
+    newlines serialize correctly through JSON.
+    """
+
+    def test_tab_in_new_string_unescaped_under_escape_normalized(self):
+        """File has real tab, model sends literal \\t in BOTH old and new.
+
+        Match strategy is ``escape_normalized``.
+        """
+        content = "def hello():\n\tprint(\"before\")\n"
+        old_string = "def hello():\n\\tprint(\"before\")\n"
+        new_string = "def hello():\n\\tprint(\"after\")\n"
+        new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None, f"Unexpected error: {err}"
+        assert count == 1
+        assert strategy == "escape_normalized"
+        assert "\tprint(\"after\")" in new
+        assert "\\t" not in new
+
+    def test_tab_in_new_string_unescaped_under_exact(self):
+        """File has real tab, old_string has real tab too (matches via
+        ``exact``), but new_string still arrives with literal ``\\t``.
+
+        This is the issue's headline reproduction — the previous fix that
+        gated on ``strategy_name == "escape_normalized"`` missed this case.
+        """
+        content = "def hello():\n\tprint(\"before\")\n"
+        old_string = "\tprint(\"before\")"           # real tab
+        new_string = "\\tprint(\"after\")"           # literal backslash + t
+        new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None, f"Unexpected error: {err}"
+        assert count == 1
+        assert strategy == "exact"
+        assert "\tprint(\"after\")" in new
+        assert "\\t" not in new
+
+    def test_carriage_return_in_new_string_unescaped(self):
+        """File has real CR, model sends literal \\r in new_string."""
+        content = "line1\r\nline2\r\n"
+        old_string = "line1\\r\\nline2\\r\\n"
+        new_string = "replaced\\r\\n"
+        new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None, f"Unexpected error: {err}"
+        assert count == 1
+        assert strategy == "escape_normalized"
+        assert "replaced\r" in new
+
+    def test_newline_in_new_string_NOT_unescaped(self):
+        """``\\n`` is intentionally left alone — newlines serialize correctly
+        through JSON, and unescaping would corrupt source-code escape
+        sequences far more often than help.
+        """
+        content = "line1\nline2\n"
+        old_string = "line1\nline2"
+        new_string = "alpha\\nbeta"                 # literal backslash + n
+        new, count, _, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None, f"Unexpected error: {err}"
+        assert count == 1
+        # The literal two-character sequence ``\n`` must survive verbatim.
+        assert "alpha\\nbeta" in new
+        # And there should be no real newline added where ``\\n`` sat.
+        assert "alpha\nbeta" not in new
+
+    def test_mixed_tab_and_newline_only_tab_unescaped(self):
+        """When new_string contains both \\t and \\n, only \\t is converted."""
+        content = "def foo():\n\tpass\n"
+        old_string = "def foo():\n\tpass\n"
+        new_string = "def bar():\\n\\treturn 1\\n"
+        new, count, _, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None, f"Unexpected error: {err}"
+        assert count == 1
+        # \t -> real tab
+        assert "\treturn 1" in new
+        assert "\\t" not in new
+        # \n preserved as literal backslash-n
+        assert "\\n" in new
+
+    def test_exact_match_preserves_literal_backslash_t_in_string_literal(self):
+        """If the matched region of the file does NOT contain a real tab,
+        new_string's literal ``\\t`` is preserved — the file genuinely uses
+        a backslash-t sequence (e.g. a Python source line ``sep = "\\t"``).
+        """
+        content = 'sep = "\\t"\n'                   # source contains backslash + t
+        old_string = 'sep = "\\t"\n'
+        new_string = 'sep = "\\tab"\n'              # still backslash + t literal
+        new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None, f"Unexpected error: {err}"
+        assert count == 1
+        assert strategy == "exact"
+        # File still has the literal two-char ``\t`` — no tab byte injected.
+        assert 'sep = "\\tab"' in new
+        assert "\t" not in new
+
+    def test_no_escape_sequences_passthrough(self):
+        """When new_string has no \\t or \\r, the helper is a no-op."""
+        content = "def foo():\n    return 1\n"
+        old_string = "def foo():\n    return 1\n"
+        new_string = "def foo():\n    return 2\n"
+        new, count, _, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None
+        assert count == 1
+        assert "return 2" in new
+

tests/tools/test_hardline_blocklist.py

@@ -6,12 +6,10 @@ gateway /yolo, approvals.mode=off, or cron approve mode.
 
 Inspired by Mercury Agent's permission-hardened blocklist.
 """
-import os
 
 import pytest
 
 from tools.approval import (
-    DANGEROUS_PATTERNS,
     HARDLINE_PATTERNS,
     check_all_command_guards,
     check_dangerous_command,

tests/tools/test_heartbeat_stale_thresholds.py

@@ -1,6 +1,5 @@
 """Tests for delegate heartbeat stale threshold configuration."""
 
-import pytest
 
 
 class TestHeartbeatStaleThresholds:

tests/tools/test_hidden_dir_filter.py

@@ -7,8 +7,7 @@ This caused quarantined skills (.hub/quarantine/) to appear as installed.
 Now uses Path.parts which is platform-independent.
 """
 
-import os
-from pathlib import Path, PurePosixPath, PureWindowsPath
+from pathlib import Path
 
 
 def _old_filter_matches(path_str: str) -> bool:

tests/tools/test_interrupt.py

@@ -65,7 +65,7 @@ class TestPreToolCheck:
 
     def test_all_tools_skipped_when_interrupted(self):
         """Mock an interrupted agent and verify no tools execute."""
-        from unittest.mock import MagicMock, patch
+        from unittest.mock import MagicMock
 
         # Build a fake assistant_message with 3 tool calls
         tc1 = MagicMock()

tests/tools/test_kanban_tools.py

@@ -1338,6 +1338,7 @@ def test_worker_complete_rejects_stale_run_id(worker_env, monkeypatch):
     try:
         run1 = kb.latest_run(conn, worker_env)
         kb._set_worker_pid(conn, worker_env, 98765)
+        monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0")
         monkeypatch.setattr(_kb, "_pid_alive", lambda pid: False)
         assert kb.detect_crashed_workers(conn) == [worker_env]
 

tests/tools/test_lazy_deps.py

@@ -12,7 +12,6 @@ call is mocked — we never actually shell out during unit tests.
 
 from __future__ import annotations
 
-from typing import Iterator
 
 import pytest
 

tests/tools/test_line_ending_preservation.py

@@ -9,8 +9,6 @@ See issue #507 (Roo Code deep-dive, item 2c).
 """
 
 import json
-import os
-import tempfile
 
 import pytest
 

tests/tools/test_llm_content_none_guard.py

@@ -12,7 +12,6 @@ reasoning fields when content is empty.
 
 import asyncio
 import types
-from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 

tests/tools/test_local_background_child_hang.py

@@ -10,7 +10,6 @@ of the backgrounded service (indefinitely for a uvicorn server).
 The fix switches ``_drain()`` to select()-based non-blocking reads and
 stops draining shortly after bash exits even if the pipe hasn't EOF'd.
 """
-import json
 import subprocess
 import time
 

tests/tools/test_local_env_blocklist.py

@@ -93,6 +93,59 @@ class TestProviderEnvBlocklist:
         for var in registry_vars:
             assert var not in result_env, f"{var} leaked into subprocess env"
 
+    def test_bedrock_bearer_token_is_stripped(self):
+        """The Bedrock-specific bearer token is a Hermes inference secret
+        (analogous to OPENAI_API_KEY) and must not leak into subprocesses.
+
+        Regression for #32314: AWS_BEARER_TOKEN_BEDROCK leaked into terminal /
+        execute_code children because the ``bedrock`` ProviderConfig declares
+        ``api_key_env_vars=()`` (auth_type="aws_sdk") and the blocklist builder
+        only consulted that field. The reporter caught it when ``opencode
+        models`` run inside a Hermes terminal enumerated the entire Bedrock
+        catalog off the leaked bearer token.
+        """
+        result_env = _run_with_env(extra_os_env={
+            "AWS_BEARER_TOKEN_BEDROCK": "bedrock-bearer-secret",
+        })
+
+        assert "AWS_BEARER_TOKEN_BEDROCK" not in result_env, (
+            "AWS_BEARER_TOKEN_BEDROCK leaked into subprocess env (see #32314)"
+        )
+
+    def test_general_aws_credential_chain_is_preserved(self):
+        """The GENERAL AWS credential chain must STILL pass through to
+        subprocesses — this is the no-regression guard for #32314.
+
+        Per SECURITY.md §3.2 the local terminal is the user's trusted operator
+        shell. A user running ``aws``/``terraform``/``cdk``/``boto3`` in the
+        agent terminal must keep the same AWS access their own shell has.
+        Stripping these would (a) break every user who does AWS work in the
+        agent terminal — not just Bedrock users, since the registry is iterated
+        unconditionally — and (b) be unrecoverable, because env_passthrough.py
+        refuses to re-allow anything in _HERMES_PROVIDER_ENV_BLOCKLIST
+        (GHSA-rhgp-j443-p4rf). Only the Bedrock inference bearer token is
+        Hermes-managed; the rest belongs to the user.
+        """
+        general_chain = {
+            "AWS_ACCESS_KEY_ID": "AKIAIOSFODNN7EXAMPLE",
+            "AWS_SECRET_ACCESS_KEY": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+            "AWS_SESSION_TOKEN": "session-token",
+            "AWS_PROFILE": "production",
+            "AWS_DEFAULT_REGION": "us-east-1",
+            "AWS_REGION": "us-east-1",
+            "AWS_SHARED_CREDENTIALS_FILE": "/home/user/.aws/credentials",
+            "AWS_CONFIG_FILE": "/home/user/.aws/config",
+            "AWS_WEB_IDENTITY_TOKEN_FILE": "/var/run/secrets/token",
+            "AWS_ROLE_ARN": "arn:aws:iam::123456789012:role/example",
+        }
+        result_env = _run_with_env(extra_os_env=general_chain)
+
+        for var, value in general_chain.items():
+            assert result_env.get(var) == value, (
+                f"{var} was stripped from subprocess env — this is a "
+                f"capability regression (see #32314 discussion)"
+            )
+
     def test_non_registry_provider_vars_are_stripped(self):
         """Extra provider vars not in PROVIDER_REGISTRY must also be blocked."""
         extra_provider_vars = {
@@ -213,6 +266,36 @@ class TestBlocklistCoverage:
                     f"(provider={pconfig.id}) missing from blocklist"
                 )
 
+    def test_bedrock_bearer_token_is_in_blocklist(self):
+        """auth_type='aws_sdk' providers contribute their Hermes-managed
+        inference token (the Bedrock bearer) to the blocklist, keyed off
+        auth_type so any future SDK-cred provider is covered automatically."""
+        assert "AWS_BEARER_TOKEN_BEDROCK" in _HERMES_PROVIDER_ENV_BLOCKLIST
+
+    def test_general_aws_chain_not_in_blocklist(self):
+        """The general AWS credential chain must NOT be in the blocklist —
+        no-regression guard for #32314. These belong to the user's trusted
+        operator shell (SECURITY.md §3.2), not to Hermes, and blocklisting
+        them would be unrecoverable via env_passthrough (GHSA-rhgp-j443-p4rf).
+        """
+        general_chain = {
+            "AWS_ACCESS_KEY_ID",
+            "AWS_SECRET_ACCESS_KEY",
+            "AWS_SESSION_TOKEN",
+            "AWS_PROFILE",
+            "AWS_DEFAULT_REGION",
+            "AWS_REGION",
+            "AWS_SHARED_CREDENTIALS_FILE",
+            "AWS_CONFIG_FILE",
+            "AWS_WEB_IDENTITY_TOKEN_FILE",
+            "AWS_ROLE_ARN",
+        }
+        leaked_block = general_chain & _HERMES_PROVIDER_ENV_BLOCKLIST
+        assert not leaked_block, (
+            f"General AWS chain vars must stay inheritable, but these are "
+            f"blocklisted: {sorted(leaked_block)} (capability regression, #32314)"
+        )
+
     def test_extra_auth_vars_covered(self):
         """Non-registry auth vars (ANTHROPIC_TOKEN, CLAUDE_CODE_OAUTH_TOKEN)
         must also be in the blocklist."""

tests/tools/test_local_env_windows_msys.py

@@ -18,10 +18,8 @@ and ``os.path.isdir`` so the MSYS path tests as "missing" exactly like
 on the real OS.
 """
 
-import os
 from unittest.mock import patch
 
-import pytest
 
 from tools.environments import local as local_mod
 from tools.environments.local import (

tests/tools/test_local_interrupt_cleanup.py

@@ -161,7 +161,6 @@ def test_wait_for_process_kills_subprocess_on_keyboardinterrupt():
         # way CPython's signal machinery would.  We use ctypes.PyThreadState_SetAsyncExc
         # which is how signal delivery to non-main threads is simulated.
         import ctypes
-        import sys as _sys
         # py-thread-state exception targets need the ident, not the Thread
         tid = t.ident
         assert tid is not None

tests/tools/test_local_shell_init.py

@@ -14,7 +14,6 @@ import pytest
 from tools.environments.local import (
     LocalEnvironment,
     _prepend_shell_init,
-    _read_terminal_shell_init_config,
     _resolve_shell_init_files,
 )
 

tests/tools/test_managed_browserbase_and_modal.py

@@ -9,6 +9,8 @@ from unittest.mock import patch
 
 import pytest
 
+from hermes_cli.nous_account import NousPortalAccountInfo
+
 
 REPO_ROOT = Path(__file__).resolve().parents[2]
 TOOLS_DIR = REPO_ROOT / "tools"
@@ -69,10 +71,17 @@ def _enable_managed_nous_tools(monkeypatch):
     The _install_fake_tools_package() helper resets and reimports tool modules,
     so a simple monkeypatch on tool_backend_helpers doesn't survive.  We patch
     the *source* modules that the reimported modules will import from — both
-    hermes_cli.auth and hermes_cli.models — so the function body returns True.
+    hermes_cli.nous_account — so the function body returns True.
     """
-    monkeypatch.setattr("hermes_cli.auth.get_nous_auth_status", lambda: {"logged_in": True})
-    monkeypatch.setattr("hermes_cli.models.check_nous_free_tier", lambda: False)
+    monkeypatch.setattr(
+        "hermes_cli.nous_account.get_nous_portal_account_info",
+        lambda: NousPortalAccountInfo(
+            logged_in=True,
+            source="jwt",
+            fresh=False,
+            paid_service_access=True,
+        ),
+    )
 
 
 def _install_fake_tools_package():

tests/tools/test_managed_media_gateways.py

@@ -5,6 +5,8 @@ from pathlib import Path
 
 import pytest
 
+from hermes_cli.nous_account import NousPortalAccountInfo
+
 
 TOOLS_DIR = Path(__file__).resolve().parents[2] / "tools"
 
@@ -48,8 +50,15 @@ def _restore_tool_and_agent_modules():
 def _enable_managed_nous_tools(monkeypatch):
     """Patch the source modules so managed_nous_tools_enabled() returns True
     even after tool modules are dynamically reloaded."""
-    monkeypatch.setattr("hermes_cli.auth.get_nous_auth_status", lambda: {"logged_in": True})
-    monkeypatch.setattr("hermes_cli.models.check_nous_free_tier", lambda: False)
+    monkeypatch.setattr(
+        "hermes_cli.nous_account.get_nous_portal_account_info",
+        lambda: NousPortalAccountInfo(
+            logged_in=True,
+            source="jwt",
+            fresh=False,
+            paid_service_access=True,
+        ),
+    )
 
 
 def _install_fake_tools_package():
@@ -296,3 +305,214 @@ def test_transcription_uses_model_specific_response_formats(monkeypatch, tmp_pat
     assert json_result["transcript"] == "hello from gpt-4o"
     assert json_capture["transcription_kwargs"]["response_format"] == "json"
     assert json_capture["close_calls"] == 1
+
+
+PLUGINS_DIR = Path(__file__).resolve().parents[2] / "plugins"
+
+
+def _load_video_gen_plugin(monkeypatch):
+    """Load the FAL video gen plugin in isolation."""
+    _install_fake_tools_package()
+
+    # Also need the agent.video_gen_provider ABC
+    agent_dir = Path(__file__).resolve().parents[2] / "agent"
+    spec = spec_from_file_location(
+        "agent.video_gen_provider",
+        agent_dir / "video_gen_provider.py",
+    )
+    assert spec and spec.loader
+    mod = module_from_spec(spec)
+    sys.modules["agent.video_gen_provider"] = mod
+    spec.loader.exec_module(mod)
+
+    # Load the plugin
+    plugin_init = PLUGINS_DIR / "video_gen" / "fal" / "__init__.py"
+    spec = spec_from_file_location("plugins.video_gen.fal", plugin_init)
+    assert spec and spec.loader
+    plugin_mod = module_from_spec(spec)
+    sys.modules["plugins.video_gen.fal"] = plugin_mod
+    spec.loader.exec_module(plugin_mod)
+    return plugin_mod
+
+
+def test_video_gen_managed_fal_submit_uses_gateway(monkeypatch):
+    """Video gen routes through the managed gateway when FAL_KEY is absent."""
+    captured = {}
+    fake_fal = _install_fake_fal_client(captured)
+    monkeypatch.delenv("FAL_KEY", raising=False)
+    monkeypatch.setenv("FAL_QUEUE_GATEWAY_URL", "http://127.0.0.1:3009")
+    monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-video-token")
+
+    plugin = _load_video_gen_plugin(monkeypatch)
+
+    # Patch uuid for deterministic idempotency key
+    monkeypatch.setattr(plugin.uuid, "uuid4", lambda: "video-submit-456")
+
+    plugin._submit_fal_video_request(
+        "fal-ai/pixverse/v6/text-to-video",
+        {"prompt": "a cat riding a bicycle", "duration": "5"},
+    )
+
+    assert captured["submit_via"] == "managed_client"
+    assert captured["client_key"] == "nous-video-token"
+    assert captured["submit_url"] == "http://127.0.0.1:3009/fal-ai/pixverse/v6/text-to-video"
+    assert captured["method"] == "POST"
+    assert captured["arguments"] == {"prompt": "a cat riding a bicycle", "duration": "5"}
+    assert captured["headers"] == {"x-idempotency-key": "video-submit-456"}
+    assert captured["sync_client_inits"] == 1
+
+
+def test_video_gen_managed_client_reused_across_calls(monkeypatch):
+    """The managed video client is cached and reused across requests."""
+    captured = {}
+    _install_fake_fal_client(captured)
+    monkeypatch.delenv("FAL_KEY", raising=False)
+    monkeypatch.setenv("FAL_QUEUE_GATEWAY_URL", "http://127.0.0.1:3009")
+    monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-video-token")
+
+    plugin = _load_video_gen_plugin(monkeypatch)
+
+    plugin._submit_fal_video_request("fal-ai/pixverse/v6/text-to-video", {"prompt": "first"})
+    first_client = captured["http_client"]
+    plugin._submit_fal_video_request("fal-ai/pixverse/v6/text-to-video", {"prompt": "second"})
+
+    assert captured["sync_client_inits"] == 1
+    assert captured["http_client"] is first_client
+
+
+def test_video_gen_direct_mode_when_fal_key_set(monkeypatch):
+    """When FAL_KEY is set and gateway not preferred, uses direct fal_client.submit."""
+    captured = {}
+    _install_fake_fal_client(captured)
+    monkeypatch.setenv("FAL_KEY", "direct-fal-key-123")
+    monkeypatch.delenv("FAL_QUEUE_GATEWAY_URL", raising=False)
+    monkeypatch.delenv("TOOL_GATEWAY_USER_TOKEN", raising=False)
+
+    plugin = _load_video_gen_plugin(monkeypatch)
+    monkeypatch.setattr(plugin.uuid, "uuid4", lambda: "direct-456")
+
+    # Trigger the lazy load so _fal_client is populated from our fake
+    plugin._load_fal_client()
+
+    # In direct mode, fal_client.submit is the module-level function.
+    # Our fake raises AssertionError from the managed path, so we need
+    # to patch it to actually capture the call.
+    direct_captured = {}
+
+    def direct_submit(endpoint, arguments=None, headers=None):
+        direct_captured["endpoint"] = endpoint
+        direct_captured["arguments"] = arguments
+        direct_captured["headers"] = headers
+        # Return a mock handle
+        class FakeHandle:
+            def get(self):
+                return {"video": {"url": "https://fal.media/result.mp4"}}
+        return FakeHandle()
+
+    plugin._fal_client.submit = direct_submit
+
+    plugin._submit_fal_video_request(
+        "fal-ai/pixverse/v6/text-to-video",
+        {"prompt": "test direct"},
+    )
+
+    assert direct_captured["endpoint"] == "fal-ai/pixverse/v6/text-to-video"
+    assert direct_captured["arguments"] == {"prompt": "test direct"}
+    assert direct_captured["headers"] == {"x-idempotency-key": "direct-456"}
+    # Managed client should NOT have been initialized
+    assert "submit_via" not in captured
+
+
+def test_video_gen_gateway_4xx_raises_actionable_valueerror(monkeypatch):
+    """A 4xx from the managed gateway surfaces a clear ValueError with remediation hints."""
+    captured = {}
+    _install_fake_fal_client(captured)
+    monkeypatch.delenv("FAL_KEY", raising=False)
+    monkeypatch.setenv("FAL_QUEUE_GATEWAY_URL", "http://127.0.0.1:3009")
+    monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-video-token")
+
+    plugin = _load_video_gen_plugin(monkeypatch)
+
+    # Make _maybe_retry_request raise an exception with a 403 status
+    class FakeResponse:
+        status_code = 403
+
+    class GatewayRejectError(Exception):
+        def __init__(self):
+            super().__init__("forbidden")
+            self.response = FakeResponse()
+
+    original_retry = sys.modules["fal_client"].client._maybe_retry_request
+
+    def raising_retry(client, method, url, json=None, timeout=None, headers=None):
+        raise GatewayRejectError()
+
+    sys.modules["fal_client"].client._maybe_retry_request = raising_retry
+
+    with pytest.raises(ValueError, match=r"gateway rejected endpoint.*HTTP 403"):
+        plugin._submit_fal_video_request(
+            "fal-ai/pixverse/v6/text-to-video",
+            {"prompt": "test 4xx"},
+        )
+
+
+def test_video_gen_is_available_true_via_gateway(monkeypatch):
+    """is_available() returns True when FAL_KEY is absent but managed gateway is configured."""
+    _install_fake_fal_client({})
+    monkeypatch.delenv("FAL_KEY", raising=False)
+    monkeypatch.setenv("FAL_QUEUE_GATEWAY_URL", "http://127.0.0.1:3009")
+    monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-video-token")
+
+    plugin = _load_video_gen_plugin(monkeypatch)
+    provider = plugin.FALVideoGenProvider()
+    assert provider.is_available() is True
+
+
+def test_video_gen_prefers_gateway_overrides_direct_key(monkeypatch):
+    """When FAL_KEY is set but prefers_gateway('video_gen') is True, routes through gateway."""
+    captured = {}
+    _install_fake_fal_client(captured)
+    monkeypatch.setenv("FAL_KEY", "direct-key-present")
+    monkeypatch.setenv("FAL_QUEUE_GATEWAY_URL", "http://127.0.0.1:3009")
+    monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "nous-video-token")
+
+    plugin = _load_video_gen_plugin(monkeypatch)
+
+    # Patch prefers_gateway to return True for video_gen
+    tb_helpers = sys.modules["tools.tool_backend_helpers"]
+    original_pg = tb_helpers.prefers_gateway
+    monkeypatch.setattr(tb_helpers, "prefers_gateway", lambda section: section == "video_gen")
+
+    plugin._submit_fal_video_request(
+        "fal-ai/pixverse/v6/text-to-video",
+        {"prompt": "gateway preferred"},
+    )
+
+    assert captured["submit_via"] == "managed_client"
+    assert captured["client_key"] == "nous-video-token"
+
+
+def test_video_gen_happy_horse_uses_alibaba_namespace():
+    """Verify the happy-horse family uses alibaba/ not fal-ai/ endpoints."""
+    _install_fake_tools_package()
+
+    # Load just the plugin module to check the catalog
+    plugin_init = PLUGINS_DIR / "video_gen" / "fal" / "__init__.py"
+
+    agent_dir = Path(__file__).resolve().parents[2] / "agent"
+    spec = spec_from_file_location(
+        "agent.video_gen_provider",
+        agent_dir / "video_gen_provider.py",
+    )
+    mod = module_from_spec(spec)
+    sys.modules["agent.video_gen_provider"] = mod
+    spec.loader.exec_module(mod)
+
+    spec = spec_from_file_location("plugins.video_gen.fal", plugin_init)
+    plugin_mod = module_from_spec(spec)
+    sys.modules["plugins.video_gen.fal"] = plugin_mod
+    spec.loader.exec_module(plugin_mod)
+
+    hh = plugin_mod.FAL_FAMILIES["happy-horse"]
+    assert hh["text_endpoint"] == "alibaba/happy-horse/text-to-video"
+    assert hh["image_endpoint"] == "alibaba/happy-horse/image-to-video"

tests/tools/test_managed_modal_environment.py

@@ -1,4 +1,3 @@
-import json
 import sys
 import tempfile
 import threading

tests/tools/test_mcp_cancelled_error_propagation.py

@@ -20,7 +20,6 @@ from __future__ import annotations
 import asyncio
 from unittest.mock import patch
 
-import pytest
 
 
 async def _hanging_run(self, cfg):

tests/tools/test_mcp_client_cert.py

@@ -0,0 +1,521 @@
+"""Tests for mTLS client certificate config on MCP HTTP/SSE transports.
+
+Covers:
+
+1. ``_resolve_client_cert`` helper — string, tuple, encrypted-key, validation
+   errors, missing-file errors.
+
+2. HTTP (new SDK ``streamable_http_client``) path forwards ``cert=`` into the
+   user-owned ``httpx.AsyncClient``.
+
+3. SSE path forwards ``cert`` and ``ssl_verify`` via an ``httpx_client_factory``
+   without breaking the OAuth/headers/timeout passthrough.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# _resolve_client_cert helper
+# ---------------------------------------------------------------------------
+
+
+class TestResolveClientCert:
+    def test_returns_none_when_unset(self):
+        from tools.mcp_tool import _resolve_client_cert
+
+        assert _resolve_client_cert("srv", {}) is None
+        assert _resolve_client_cert("srv", {"url": "https://x"}) is None
+
+    def test_string_form_single_pem(self, tmp_path):
+        from tools.mcp_tool import _resolve_client_cert
+
+        pem = tmp_path / "combined.pem"
+        pem.write_text("dummy")
+
+        result = _resolve_client_cert("srv", {"client_cert": str(pem)})
+        assert result == str(pem)
+
+    def test_string_cert_with_separate_key(self, tmp_path):
+        from tools.mcp_tool import _resolve_client_cert
+
+        cert = tmp_path / "client.crt"
+        key = tmp_path / "client.key"
+        cert.write_text("cert")
+        key.write_text("key")
+
+        result = _resolve_client_cert("srv", {
+            "client_cert": str(cert),
+            "client_key": str(key),
+        })
+        assert result == (str(cert), str(key))
+
+    def test_list_form_two_elements(self, tmp_path):
+        from tools.mcp_tool import _resolve_client_cert
+
+        cert = tmp_path / "client.crt"
+        key = tmp_path / "client.key"
+        cert.write_text("cert")
+        key.write_text("key")
+
+        result = _resolve_client_cert("srv", {
+            "client_cert": [str(cert), str(key)],
+        })
+        assert result == (str(cert), str(key))
+
+    def test_list_form_with_passphrase(self, tmp_path):
+        from tools.mcp_tool import _resolve_client_cert
+
+        cert = tmp_path / "client.crt"
+        key = tmp_path / "client.key"
+        cert.write_text("cert")
+        key.write_text("key")
+
+        result = _resolve_client_cert("srv", {
+            "client_cert": [str(cert), str(key), "passphrase"],
+        })
+        assert result == (str(cert), str(key), "passphrase")
+
+    def test_tilde_expansion(self, tmp_path, monkeypatch):
+        from tools.mcp_tool import _resolve_client_cert
+
+        monkeypatch.setenv("HOME", str(tmp_path))
+        pem = tmp_path / "client.pem"
+        pem.write_text("dummy")
+
+        result = _resolve_client_cert("srv", {"client_cert": "~/client.pem"})
+        assert result == str(pem)
+
+    def test_missing_file_raises(self, tmp_path):
+        from tools.mcp_tool import _resolve_client_cert
+
+        with pytest.raises(FileNotFoundError, match=r"srv.*client_cert.*not found"):
+            _resolve_client_cert("srv", {
+                "client_cert": str(tmp_path / "nope.pem"),
+            })
+
+    def test_missing_key_file_raises(self, tmp_path):
+        from tools.mcp_tool import _resolve_client_cert
+
+        cert = tmp_path / "client.crt"
+        cert.write_text("cert")
+
+        with pytest.raises(FileNotFoundError, match=r"srv.*client_key.*not found"):
+            _resolve_client_cert("srv", {
+                "client_cert": str(cert),
+                "client_key": str(tmp_path / "missing.key"),
+            })
+
+    def test_list_with_bad_length_raises(self, tmp_path):
+        from tools.mcp_tool import _resolve_client_cert
+
+        with pytest.raises(ValueError, match=r"list form must have 2 or 3"):
+            _resolve_client_cert("srv", {"client_cert": [str(tmp_path / "x")]})
+
+    def test_list_plus_client_key_rejected(self, tmp_path):
+        from tools.mcp_tool import _resolve_client_cert
+
+        cert = tmp_path / "client.crt"
+        key = tmp_path / "client.key"
+        cert.write_text("cert")
+        key.write_text("key")
+
+        with pytest.raises(ValueError, match=r"either client_cert as a list"):
+            _resolve_client_cert("srv", {
+                "client_cert": [str(cert), str(key)],
+                "client_key": str(key),
+            })
+
+    def test_non_string_path_rejected(self):
+        from tools.mcp_tool import _resolve_client_cert
+
+        with pytest.raises(ValueError, match=r"client_cert must be a non-empty string"):
+            _resolve_client_cert("srv", {"client_cert": 123})
+
+    def test_password_must_be_string(self, tmp_path):
+        from tools.mcp_tool import _resolve_client_cert
+
+        cert = tmp_path / "client.crt"
+        key = tmp_path / "client.key"
+        cert.write_text("cert")
+        key.write_text("key")
+
+        with pytest.raises(ValueError, match=r"key passphrase.*must be a string"):
+            _resolve_client_cert("srv", {
+                "client_cert": [str(cert), str(key), 42],
+            })
+
+
+# ---------------------------------------------------------------------------
+# HTTP transport — cert forwarded into httpx.AsyncClient
+# ---------------------------------------------------------------------------
+
+
+class TestHTTPClientCert:
+    def test_cert_forwarded_to_async_client(self, tmp_path):
+        """When client_cert is set, the new-SDK HTTP path passes ``cert=``
+        into ``httpx.AsyncClient``."""
+        from tools.mcp_tool import MCPServerTask
+
+        cert = tmp_path / "client.pem"
+        cert.write_text("dummy")
+
+        server = MCPServerTask("remote")
+        captured: dict = {}
+
+        class DummyAsyncClient:
+            def __init__(self, **kwargs):
+                captured.update(kwargs)
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *a):
+                return False
+
+        class DummyTransportCtx:
+            async def __aenter__(self):
+                return MagicMock(), MagicMock(), (lambda: None)
+
+            async def __aexit__(self, *a):
+                return False
+
+        class DummySession:
+            def __init__(self, *args, **kwargs):
+                pass
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *a):
+                return False
+
+            async def initialize(self):
+                return None
+
+        async def _discover_tools(self):
+            self._shutdown_event.set()
+
+        async def _drive():
+            with patch("tools.mcp_tool._MCP_HTTP_AVAILABLE", True), \
+                 patch("tools.mcp_tool._MCP_NEW_HTTP", True), \
+                 patch("httpx.AsyncClient", DummyAsyncClient), \
+                 patch("tools.mcp_tool.streamable_http_client",
+                       return_value=DummyTransportCtx()), \
+                 patch("tools.mcp_tool.ClientSession", DummySession), \
+                 patch.object(MCPServerTask, "_discover_tools", _discover_tools):
+                await server._run_http({
+                    "url": "https://example.com/mcp",
+                    "client_cert": str(cert),
+                })
+
+        asyncio.run(_drive())
+        assert captured.get("cert") == str(cert)
+
+    def test_cert_tuple_forwarded(self, tmp_path):
+        """List/tuple form resolves to a tuple in ``cert=``."""
+        from tools.mcp_tool import MCPServerTask
+
+        cert = tmp_path / "client.crt"
+        key = tmp_path / "client.key"
+        cert.write_text("cert")
+        key.write_text("key")
+
+        server = MCPServerTask("remote")
+        captured: dict = {}
+
+        class DummyAsyncClient:
+            def __init__(self, **kwargs):
+                captured.update(kwargs)
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *a):
+                return False
+
+        class DummyTransportCtx:
+            async def __aenter__(self):
+                return MagicMock(), MagicMock(), (lambda: None)
+
+            async def __aexit__(self, *a):
+                return False
+
+        class DummySession:
+            def __init__(self, *args, **kwargs):
+                pass
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *a):
+                return False
+
+            async def initialize(self):
+                return None
+
+        async def _discover_tools(self):
+            self._shutdown_event.set()
+
+        async def _drive():
+            with patch("tools.mcp_tool._MCP_HTTP_AVAILABLE", True), \
+                 patch("tools.mcp_tool._MCP_NEW_HTTP", True), \
+                 patch("httpx.AsyncClient", DummyAsyncClient), \
+                 patch("tools.mcp_tool.streamable_http_client",
+                       return_value=DummyTransportCtx()), \
+                 patch("tools.mcp_tool.ClientSession", DummySession), \
+                 patch.object(MCPServerTask, "_discover_tools", _discover_tools):
+                await server._run_http({
+                    "url": "https://example.com/mcp",
+                    "client_cert": [str(cert), str(key)],
+                })
+
+        asyncio.run(_drive())
+        assert captured.get("cert") == (str(cert), str(key))
+
+    def test_no_cert_means_no_cert_kwarg(self):
+        """When client_cert is unset, ``cert`` is not passed to ``httpx.AsyncClient``
+        (matches SDK defaults)."""
+        from tools.mcp_tool import MCPServerTask
+
+        server = MCPServerTask("remote")
+        captured: dict = {}
+
+        class DummyAsyncClient:
+            def __init__(self, **kwargs):
+                captured.update(kwargs)
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *a):
+                return False
+
+        class DummyTransportCtx:
+            async def __aenter__(self):
+                return MagicMock(), MagicMock(), (lambda: None)
+
+            async def __aexit__(self, *a):
+                return False
+
+        class DummySession:
+            def __init__(self, *args, **kwargs):
+                pass
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *a):
+                return False
+
+            async def initialize(self):
+                return None
+
+        async def _discover_tools(self):
+            self._shutdown_event.set()
+
+        async def _drive():
+            with patch("tools.mcp_tool._MCP_HTTP_AVAILABLE", True), \
+                 patch("tools.mcp_tool._MCP_NEW_HTTP", True), \
+                 patch("httpx.AsyncClient", DummyAsyncClient), \
+                 patch("tools.mcp_tool.streamable_http_client",
+                       return_value=DummyTransportCtx()), \
+                 patch("tools.mcp_tool.ClientSession", DummySession), \
+                 patch.object(MCPServerTask, "_discover_tools", _discover_tools):
+                await server._run_http({"url": "https://example.com/mcp"})
+
+        asyncio.run(_drive())
+        assert "cert" not in captured
+
+    def test_missing_cert_file_surfaces_clear_error(self, tmp_path):
+        """A missing cert file fails fast with a server-scoped error message."""
+        from tools.mcp_tool import MCPServerTask
+
+        server = MCPServerTask("remote")
+
+        async def _drive():
+            with patch("tools.mcp_tool._MCP_HTTP_AVAILABLE", True), \
+                 patch("tools.mcp_tool._MCP_NEW_HTTP", True):
+                await server._run_http({
+                    "url": "https://example.com/mcp",
+                    "client_cert": str(tmp_path / "nope.pem"),
+                })
+
+        with pytest.raises(FileNotFoundError, match=r"remote.*client_cert.*not found"):
+            asyncio.run(_drive())
+
+
+# ---------------------------------------------------------------------------
+# SSE transport — cert + verify routed via httpx_client_factory
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def patch_sse_client():
+    """Replace ``sse_client`` with a MagicMock that records its kwargs.
+
+    Returns the captured kwargs dict so tests can assert how ``_run_http``
+    called it.
+    """
+    captured_kwargs: dict = {}
+
+    class _FakeStream:
+        def __init__(self):
+            self._read = AsyncMock()
+            self._write = AsyncMock()
+
+        async def __aenter__(self):
+            return (self._read, self._write)
+
+        async def __aexit__(self, *a):
+            return False
+
+    def fake_sse_client(**kwargs):
+        captured_kwargs.clear()
+        captured_kwargs.update(kwargs)
+        return _FakeStream()
+
+    class _FakeSession:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def __aenter__(self):
+            mock_session = MagicMock()
+            mock_session.initialize = AsyncMock()
+            return mock_session
+
+        async def __aexit__(self, *a):
+            return False
+
+    with patch("tools.mcp_tool.sse_client", new=fake_sse_client), \
+         patch("tools.mcp_tool.ClientSession", new=_FakeSession):
+        yield captured_kwargs
+
+
+class TestSSEClientCert:
+    def test_no_factory_when_defaults(self, patch_sse_client):
+        """With no cert and ssl_verify=True (default), the SDK's own factory is
+        used — we don't inject one."""
+        from tools.mcp_tool import MCPServerTask
+
+        server = MCPServerTask("sse-test")
+        server._auth_type = ""
+        server._sampling = None
+
+        async def drive():
+            with patch.object(MCPServerTask, "_wait_for_lifecycle_event",
+                              new=AsyncMock(return_value="shutdown")), \
+                 patch.object(MCPServerTask, "_discover_tools", new=AsyncMock()):
+                try:
+                    await asyncio.wait_for(
+                        server._run_http({
+                            "url": "https://example.com/mcp/sse",
+                            "transport": "sse",
+                        }),
+                        timeout=2.0,
+                    )
+                except (asyncio.TimeoutError, StopAsyncIteration, Exception):
+                    pass
+
+        asyncio.run(drive())
+        assert "httpx_client_factory" not in patch_sse_client
+
+    def test_factory_injected_when_cert_set(self, patch_sse_client, tmp_path):
+        """With client_cert set, an httpx_client_factory is injected that
+        applies the cert (and follow_redirects=True to match the SDK)."""
+        from tools.mcp_tool import MCPServerTask
+
+        cert = tmp_path / "client.pem"
+        cert.write_text("dummy")
+
+        server = MCPServerTask("sse-test")
+        server._auth_type = ""
+        server._sampling = None
+
+        async def drive():
+            with patch.object(MCPServerTask, "_wait_for_lifecycle_event",
+                              new=AsyncMock(return_value="shutdown")), \
+                 patch.object(MCPServerTask, "_discover_tools", new=AsyncMock()):
+                try:
+                    await asyncio.wait_for(
+                        server._run_http({
+                            "url": "https://example.com/mcp/sse",
+                            "transport": "sse",
+                            "client_cert": str(cert),
+                        }),
+                        timeout=2.0,
+                    )
+                except (asyncio.TimeoutError, StopAsyncIteration, Exception):
+                    pass
+
+        asyncio.run(drive())
+
+        factory = patch_sse_client.get("httpx_client_factory")
+        assert factory is not None, "expected httpx_client_factory to be injected"
+
+        # Invoke the factory the way the SDK would; capture the resulting
+        # httpx.AsyncClient kwargs.
+        captured_client_kwargs: dict = {}
+
+        class DummyAsyncClient:
+            def __init__(self, **kwargs):
+                captured_client_kwargs.update(kwargs)
+
+        import httpx
+        with patch.object(httpx, "AsyncClient", DummyAsyncClient):
+            factory(headers={"x": "y"}, timeout=httpx.Timeout(30.0), auth=None)
+
+        assert captured_client_kwargs["cert"] == str(cert)
+        assert captured_client_kwargs["verify"] is True
+        assert captured_client_kwargs["follow_redirects"] is True
+        assert captured_client_kwargs["headers"] == {"x": "y"}
+
+    def test_factory_forwards_custom_ca_bundle(self, patch_sse_client, tmp_path):
+        """ssl_verify as a path is forwarded to the factory's httpx client."""
+        from tools.mcp_tool import MCPServerTask
+
+        ca_bundle = tmp_path / "ca.pem"
+        ca_bundle.write_text("dummy")
+
+        server = MCPServerTask("sse-test")
+        server._auth_type = ""
+        server._sampling = None
+
+        async def drive():
+            with patch.object(MCPServerTask, "_wait_for_lifecycle_event",
+                              new=AsyncMock(return_value="shutdown")), \
+                 patch.object(MCPServerTask, "_discover_tools", new=AsyncMock()):
+                try:
+                    await asyncio.wait_for(
+                        server._run_http({
+                            "url": "https://example.com/mcp/sse",
+                            "transport": "sse",
+                            "ssl_verify": str(ca_bundle),
+                        }),
+                        timeout=2.0,
+                    )
+                except (asyncio.TimeoutError, StopAsyncIteration, Exception):
+                    pass
+
+        asyncio.run(drive())
+
+        factory = patch_sse_client.get("httpx_client_factory")
+        assert factory is not None
+
+        captured_client_kwargs: dict = {}
+
+        class DummyAsyncClient:
+            def __init__(self, **kwargs):
+                captured_client_kwargs.update(kwargs)
+
+        import httpx
+        with patch.object(httpx, "AsyncClient", DummyAsyncClient):
+            factory(headers=None, timeout=None, auth=None)
+
+        assert captured_client_kwargs["verify"] == str(ca_bundle)
+        assert "cert" not in captured_client_kwargs

tests/tools/test_mcp_empty_error_message.py

@@ -7,11 +7,7 @@ nothing after the colon, making debugging impossible.
 Fix: ``_exc_str()`` falls back to ``repr(exc)`` when ``str(exc)`` is empty.
 """
 
-import json
-from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
 
-import pytest
 
 from tools.mcp_tool import _exc_str, _sanitize_error
 

tests/tools/test_mcp_image_content.py

@@ -18,9 +18,7 @@ from __future__ import annotations
 
 import base64
 from types import SimpleNamespace
-from unittest.mock import patch
 
-import pytest
 
 
 def _png_bytes():

tests/tools/test_mcp_oauth.py

@@ -5,8 +5,7 @@ import os
 import stat
 import sys
 from io import BytesIO
-from pathlib import Path
-from unittest.mock import patch, MagicMock, AsyncMock
+from unittest.mock import patch, MagicMock
 
 import pytest
 

tests/tools/test_mcp_stability.py

@@ -1,12 +1,9 @@
 """Tests for MCP stability fixes — event loop handler, PID tracking, shutdown robustness."""
 
 import asyncio
-import os
 import signal
-import threading
 from unittest.mock import patch, MagicMock
 
-import pytest
 
 
 # ---------------------------------------------------------------------------
@@ -227,7 +224,7 @@ class TestMCPInitialConnectionRetry:
 
     def test_initial_connect_retry_succeeds_on_second_attempt(self):
         """Server succeeds after one transient initial failure."""
-        from tools.mcp_tool import MCPServerTask, _MAX_INITIAL_CONNECT_RETRIES
+        from tools.mcp_tool import MCPServerTask
 
         call_count = 0
 

tests/tools/test_mcp_tool.py

@@ -5,7 +5,6 @@ All tests use mocks -- no real MCP servers or subprocesses are started.
 
 import asyncio
 import json
-import os
 import threading
 import time
 from types import SimpleNamespace
@@ -1770,7 +1769,7 @@ class TestConfigurableTimeouts:
 
     def test_timeout_passed_to_handler(self):
         """The tool handler uses the server's configured timeout."""
-        from tools.mcp_tool import _make_tool_handler, _servers, MCPServerTask
+        from tools.mcp_tool import _make_tool_handler, _servers
 
         mock_session = MagicMock()
         mock_session.call_tool = AsyncMock(
@@ -2225,8 +2224,6 @@ class TestUtilityToolRegistration:
 # SamplingHandler tests
 # ===========================================================================
 
-import math
-import time
 
 class _CompatType:
     def __init__(self, **kwargs):

tests/tools/test_mcp_tool_issue_948.py

@@ -1,10 +1,8 @@
 import asyncio
 import os
-import sys
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
-import pytest
 
 from tools.mcp_tool import MCPServerTask, _format_connect_error, _resolve_stdio_command, _MCP_AVAILABLE
 
@@ -34,6 +32,39 @@ def test_resolve_stdio_command_falls_back_to_hermes_node_bin(tmp_path):
     assert env["PATH"].split(os.pathsep)[0] == str(node_bin)
 
 
+def test_resolve_stdio_command_falls_back_to_usr_local_bin():
+    """When ``npx`` isn't on the filtered PATH and isn't under ``$HERMES_HOME/node/bin``
+    or ``~/.local/bin``, the resolver should still locate it at ``/usr/local/bin/npx``.
+
+    This is the canonical install location for Node on Linux from-source builds,
+    the upstream ``node:bookworm-slim`` image (which the Hermes Docker image
+    copies ``node + npm + corepack`` from since #4977), and macOS Homebrew on
+    Intel. Without this candidate, MCP servers run with an ``env.PATH`` that
+    omits ``/usr/local/bin`` (common when users hand-author PATH for sandboxing)
+    fail with ENOENT at ``execvp``.
+    """
+    target = os.path.join(os.sep, "usr", "local", "bin", "npx")
+
+    # Pretend ONLY the /usr/local/bin/npx candidate exists and is executable —
+    # the other candidates ($HERMES_HOME/node/bin/npx and ~/.local/bin/npx)
+    # should fail isfile() and the resolver must fall through to /usr/local/bin.
+    def _fake_isfile(path):
+        return path == target
+
+    def _fake_access(path, _mode):
+        return path == target
+
+    with patch("tools.mcp_tool.shutil.which", return_value=None), \
+         patch("tools.mcp_tool.os.path.isfile", side_effect=_fake_isfile), \
+         patch("tools.mcp_tool.os.access", side_effect=_fake_access):
+        command, env = _resolve_stdio_command("npx", {"PATH": "/opt/data/bin:/usr/bin:/bin"})
+
+    assert command == target
+    # /usr/local/bin must be prepended so npx's shebang (`/usr/bin/env node`)
+    # can find node in the same directory.
+    assert env["PATH"].split(os.pathsep)[0] == os.path.dirname(target)
+
+
 def test_resolve_stdio_command_respects_explicit_empty_path():
     seen_paths = []
 

tests/tools/test_mcp_tool_session_expired.py

@@ -12,8 +12,7 @@ affected MCP server failed until the gateway was manually restarted.
 """
 import json
 import threading
-import time
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import MagicMock
 
 import pytest
 

tests/tools/test_mcp_utility_capability_gating.py

@@ -29,7 +29,6 @@ from __future__ import annotations
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
-import pytest
 
 
 def _make_init_result(*, resources: bool, prompts: bool):

tests/tools/test_memory_tool.py

@@ -8,7 +8,6 @@ from tools.memory_tool import (
     MemoryStore,
     memory_tool,
     _scan_memory_content,
-    ENTRY_DELIMITER,
     MEMORY_SCHEMA,
 )
 

tests/tools/test_modal_bulk_upload.py

@@ -4,8 +4,7 @@ import asyncio
 import base64
 import io
 import tarfile
-from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 

tests/tools/test_notify_on_complete.py

@@ -10,11 +10,9 @@ Covers:
 
 import json
 import os
-import queue
 import time
 import pytest
-from pathlib import Path
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 from tools.process_registry import (
     ProcessRegistry,

tests/tools/test_process_registry.py

@@ -7,14 +7,12 @@ import subprocess
 import sys
 import time
 import pytest
-from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 from tools.environments.local import _HERMES_PROVIDER_ENV_FORCE_PREFIX
 from tools.process_registry import (
     ProcessRegistry,
     ProcessSession,
-    MAX_OUTPUT_CHARS,
     FINISHED_TTL_SECONDS,
     MAX_PROCESSES,
 )
@@ -563,9 +561,18 @@ class TestPopenLeakOnSetupFailure:
         def boom(*args, **kwargs):
             raise RuntimeError("Thread creation failed")
 
+        # proc.pid is a MagicMock-backed fake; os.getpgid(fake_pid) would query
+        # the real OS for an arbitrary PID. On a busy host that PID may exist,
+        # in which case spawn_local's primary cleanup path
+        # (os.killpg(os.getpgid(pid), SIGKILL)) succeeds against an UNRELATED
+        # real process group and proc.kill() is never reached — flaky failure,
+        # and a real risk of SIGKILLing an innocent process group. Force the
+        # ProcessLookupError fallback so the test deterministically exercises
+        # proc.kill() and never issues a real killpg.
         with patch("tools.process_registry._find_shell", return_value="/bin/bash"), \
              patch("subprocess.Popen", return_value=proc), \
              patch("threading.Thread", side_effect=boom), \
+             patch("os.getpgid", side_effect=ProcessLookupError), \
              patch.object(registry, "_write_checkpoint"):
             with pytest.raises(RuntimeError, match="Thread creation failed"):
                 registry.spawn_local("echo hello", cwd="/tmp")
@@ -590,9 +597,14 @@ class TestPopenLeakOnSetupFailure:
 
         fake_thread = MagicMock()
 
+        # See note in test_popen_killed_when_thread_creation_fails: force the
+        # ProcessLookupError fallback so cleanup deterministically calls
+        # proc.kill() instead of issuing a real os.killpg against whatever
+        # process group happens to own the fake PID on the host.
         with patch("tools.process_registry._find_shell", return_value="/bin/bash"), \
              patch("subprocess.Popen", return_value=proc), \
              patch("threading.Thread", return_value=fake_thread), \
+             patch("os.getpgid", side_effect=ProcessLookupError), \
              patch.object(registry, "_write_checkpoint", side_effect=OSError("disk full")):
             with pytest.raises(OSError, match="disk full"):
                 registry.spawn_local("echo hello", cwd="/tmp")

tests/tools/test_resolve_path.py

@@ -4,7 +4,6 @@ import os
 from pathlib import Path
 from types import SimpleNamespace
 
-import pytest
 
 
 class TestResolvePath:

tests/tools/test_search_hidden_dirs.py

@@ -13,7 +13,6 @@ Fix: _search_files (find) and _search_with_grep both now exclude hidden
 directories, matching ripgrep's default behavior.
 """
 
-import os
 import subprocess
 
 import pytest

tests/tools/test_send_message_tool.py

@@ -4,7 +4,6 @@ import asyncio
 import json
 import os
 import sys
-from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -40,7 +39,6 @@ from tools.send_message_tool import (
 # and provide a thin ``_send_discord(token, ...)`` shim that mirrors the
 # pre-migration signature so the existing test bodies keep working.
 from plugins.platforms.discord.adapter import (
-    _DISCORD_CHANNEL_TYPE_PROBE_CACHE,
     _derive_forum_thread_name,
     _probe_is_forum_cached,
     _remember_channel_is_forum,
@@ -378,9 +376,12 @@ class TestSendMessageTool:
         )
 
     def test_media_tag_outside_allowed_roots_is_not_sent(self, tmp_path, monkeypatch):
-        # This test exercises the strict-allowlist path; disable recency trust
-        # so the freshly-written tmp_path file is not auto-accepted by the
-        # trust window. (Recency trust is covered in test_platform_base.py.)
+        # This test exercises the strict-allowlist path; force strict mode on
+        # and disable recency trust so the freshly-written tmp_path file is
+        # not auto-accepted by the trust window. (Recency trust is covered
+        # in test_platform_base.py. The public default flipped to non-strict
+        # in 2026-05; this test pins strict on explicitly.)
+        monkeypatch.setenv("HERMES_MEDIA_DELIVERY_STRICT", "1")
         monkeypatch.setenv("HERMES_MEDIA_TRUST_RECENT_FILES", "0")
         config, telegram_cfg = _make_config()
         secret = tmp_path / "secret.pdf"
@@ -1514,7 +1515,6 @@ class TestSendMatrixUrlEncoding:
 
     def test_room_id_is_percent_encoded_in_url(self):
         """Matrix room IDs with ! and : are percent-encoded in the PUT URL."""
-        import aiohttp
 
         mock_resp = MagicMock()
         mock_resp.status = 200
@@ -1891,10 +1891,6 @@ class TestForumProbeCache:
         discord_adapter._DISCORD_CHANNEL_TYPE_PROBE_CACHE.clear()
 
     def test_cache_round_trip(self):
-        from plugins.platforms.discord.adapter import (
-            _probe_is_forum_cached,
-            _remember_channel_is_forum,
-        )
         assert _probe_is_forum_cached("xyz") is None
         _remember_channel_is_forum("xyz", True)
         assert _probe_is_forum_cached("xyz") is True

tests/tools/test_signal_media.py

@@ -2,7 +2,6 @@
 
 import asyncio
 import sys
-from pathlib import Path
 from types import ModuleType
 from unittest.mock import MagicMock, AsyncMock, patch
 

tests/tools/test_skill_env_passthrough.py

@@ -1,8 +1,6 @@
 """Test that skill_view registers required env vars in the passthrough registry."""
 
 import json
-import os
-from pathlib import Path
 from unittest.mock import patch
 
 import pytest

tests/tools/test_skill_improvements.py

@@ -1,9 +1,6 @@
 """Tests for skill fuzzy patching via tools.fuzzy_match."""
 
 import json
-import os
-from pathlib import Path
-from unittest.mock import patch
 
 import pytest
 

tests/tools/test_skill_manager_tool.py

@@ -12,8 +12,6 @@ from tools.skill_manager_tool import (
     _validate_category,
     _validate_frontmatter,
     _validate_file_path,
-    _find_skill,
-    _resolve_skill_dir,
     _create_skill,
     _edit_skill,
     _patch_skill,
@@ -21,8 +19,6 @@ from tools.skill_manager_tool import (
     _write_file,
     _remove_file,
     skill_manage,
-    VALID_NAME_RE,
-    ALLOWED_SUBDIRS,
     MAX_NAME_LENGTH,
 )
 

tests/tools/test_skill_provenance.py

@@ -2,7 +2,6 @@
 
 import contextvars
 
-import pytest
 
 
 

tests/tools/test_skill_size_limits.py

@@ -6,15 +6,11 @@ Hand-placed and hub-installed skills have no hard limit.
 """
 
 import json
-import os
-from pathlib import Path
-from unittest.mock import patch
 
 import pytest
 
 from tools.skill_manager_tool import (
     MAX_SKILL_CONTENT_CHARS,
-    MAX_SKILL_FILE_BYTES,
     _validate_content_size,
     skill_manage,
 )

tests/tools/test_skill_usage.py

@@ -339,7 +339,7 @@ def test_agent_created_skips_archive_and_hub_dirs(skills_home):
 # ---------------------------------------------------------------------------
 
 def test_archive_skill_moves_directory(skills_home):
-    from tools.skill_usage import archive_skill, get_record, STATE_ARCHIVED
+    from tools.skill_usage import archive_skill, get_record
     skills_dir = skills_home / "skills"
     skill_dir = _write_skill(skills_dir, "old-skill")
     assert skill_dir.exists()

tests/tools/test_skill_view_traversal.py

@@ -6,7 +6,6 @@ reading arbitrary files (e.g., ~/.hermes/.env) via path traversal.
 
 import json
 import pytest
-from pathlib import Path
 from unittest.mock import patch
 
 from tools.skills_tool import skill_view

tests/tools/test_skills_ast_audit.py

@@ -1,7 +1,6 @@
 """Tests for tools.skills_ast_audit — opt-in AST diagnostic scanner."""
 
 import sys
-from pathlib import Path
 
 from tools.skills_ast_audit import ast_scan_path, format_ast_report
 

tests/tools/test_skills_guard.py

@@ -1,7 +1,5 @@
 """Tests for tools/skills_guard.py - security scanner for skills."""
 
-import os
-import stat
 import tempfile
 from pathlib import Path
 
@@ -33,8 +31,6 @@ from tools.skills_guard import (
     _resolve_trust_level,
     _check_structure,
     _unicode_char_name,
-    INSTALL_POLICY,
-    INVISIBLE_CHARS,
     MAX_FILE_COUNT,
     MAX_SINGLE_FILE_KB,
 )
@@ -54,6 +50,14 @@ class TestResolveTrustLevel:
         assert _resolve_trust_level("anthropics/skills") == "trusted"
         assert _resolve_trust_level("openai/skills/some-skill") == "trusted"
 
+    def test_nvidia_skills_is_trusted(self):
+        # NVIDIA/skills ships NVIDIA-verified skills with detached OMS
+        # signatures and governance skill cards. It's wired through the
+        # same trust path as the OpenAI / Anthropic / HuggingFace taps.
+        assert _resolve_trust_level("NVIDIA/skills") == "trusted"
+        assert _resolve_trust_level("NVIDIA/skills/aiq-deploy") == "trusted"
+        assert _resolve_trust_level("skills-sh/NVIDIA/skills/cuopt") == "trusted"
+
     def test_trusted_repo_sibling_prefixes_are_not_trusted(self):
         assert _resolve_trust_level("openai/skills-evil") == "community"
         assert _resolve_trust_level("anthropics/skills-foo/frontend-design") == "community"

tests/tools/test_skills_hub.py

@@ -1,7 +1,6 @@
 """Tests for tools/skills_hub.py — source adapters, lock file, taps, dedup logic."""
 
 import json
-from pathlib import Path
 from unittest.mock import patch, MagicMock
 
 import httpx
@@ -71,6 +70,143 @@ class TestParseFrontmatterQuick:
         assert fm == {}
 
 
+# ---------------------------------------------------------------------------
+# GitHubSource skills.sh.json grouping sidecar (category support)
+# ---------------------------------------------------------------------------
+
+
+class TestSkillsShGroupings:
+    """Parsing + stamping of the skills.sh.json grouping sidecar.
+
+    A tap can ship a repo-root ``skills.sh.json`` declaring category
+    groupings; we flatten it to {skill_name: title} and stamp the title onto
+    each SkillMeta's ``extra["category"]``. This is the generic cross-ecosystem
+    mechanism behind NVIDIA-style categorization — not NVIDIA-specific.
+    """
+
+    def test_parse_basic_groupings(self):
+        content = json.dumps({
+            "$schema": "https://skills.sh/schemas/skills.sh.schema.json",
+            "groupings": [
+                {"title": "Inference AI", "skills": ["dynamo-router", "dynamo-recipe"]},
+                {"title": "Decision Optimization", "skills": ["cuopt-developer"]},
+            ],
+        })
+        mapping = GitHubSource._parse_skillsh_groupings(content)
+        assert mapping == {
+            "dynamo-router": "Inference AI",
+            "dynamo-recipe": "Inference AI",
+            "cuopt-developer": "Decision Optimization",
+        }
+
+    def test_parse_invalid_json_returns_none(self):
+        assert GitHubSource._parse_skillsh_groupings("not json{{") is None
+
+    def test_parse_non_dict_returns_none(self):
+        assert GitHubSource._parse_skillsh_groupings("[1, 2, 3]") is None
+
+    def test_parse_missing_groupings_returns_none(self):
+        assert GitHubSource._parse_skillsh_groupings('{"foo": 1}') is None
+
+    def test_parse_empty_groupings_returns_empty_map(self):
+        assert GitHubSource._parse_skillsh_groupings('{"groupings": []}') == {}
+
+    def test_parse_tolerates_malformed_group(self):
+        # A group missing its skills list is skipped; the valid one survives.
+        content = json.dumps({"groupings": [
+            {"title": "X"},                              # no skills -> skipped
+            {"skills": ["a"]},                           # no title -> skipped
+            {"title": "Y", "skills": ["b", 5, None]},    # only valid string members kept
+        ]})
+        assert GitHubSource._parse_skillsh_groupings(content) == {"b": "Y"}
+
+    def test_parse_first_grouping_wins_on_duplicate(self):
+        content = json.dumps({"groupings": [
+            {"title": "First", "skills": ["dup"]},
+            {"title": "Second", "skills": ["dup"]},
+        ]})
+        assert GitHubSource._parse_skillsh_groupings(content) == {"dup": "First"}
+
+    def test_get_groupings_caches_per_repo(self):
+        auth = MagicMock()
+        src = GitHubSource(auth=auth)
+        content = json.dumps({"groupings": [{"title": "T", "skills": ["s"]}]})
+        with patch.object(src, "_fetch_file_content", return_value=content) as mock_fetch:
+            first = src._get_skillsh_groupings("acme/skills")
+            second = src._get_skillsh_groupings("acme/skills")
+        assert first == {"s": "T"}
+        assert second == {"s": "T"}
+        # Second call must hit the per-repo cache, not GitHub again.
+        mock_fetch.assert_called_once_with("acme/skills", "skills.sh.json")
+
+    def test_get_groupings_no_sidecar_returns_none_and_caches(self):
+        auth = MagicMock()
+        src = GitHubSource(auth=auth)
+        with patch.object(src, "_fetch_file_content", return_value=None) as mock_fetch:
+            assert src._get_skillsh_groupings("acme/skills") is None
+            assert src._get_skillsh_groupings("acme/skills") is None
+        mock_fetch.assert_called_once()
+
+    def test_list_skills_stamps_category_from_sidecar(self):
+        auth = MagicMock()
+        src = GitHubSource(auth=auth)
+
+        meta = SkillMeta(
+            name="cuopt-developer", description="d", source="github",
+            identifier="NVIDIA/skills/skills/cuopt-developer", trust_level="trusted",
+        )
+        contents = [{"type": "dir", "name": "cuopt-developer"}]
+        groupings = {"cuopt-developer": "Decision Optimization"}
+
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.json.return_value = contents
+
+        with patch.object(src, "_read_cache", return_value=None), \
+             patch.object(src, "_write_cache"), \
+             patch.object(src, "_get_skillsh_groupings", return_value=groupings), \
+             patch.object(src, "inspect", return_value=meta), \
+             patch("tools.skills_hub.httpx.get", return_value=resp):
+            skills = src._list_skills_in_repo("NVIDIA/skills", "skills/")
+
+        assert len(skills) == 1
+        assert skills[0].extra["category"] == "Decision Optimization"
+
+    def test_list_skills_no_sidecar_leaves_extra_empty(self):
+        auth = MagicMock()
+        src = GitHubSource(auth=auth)
+
+        meta = SkillMeta(
+            name="foo", description="d", source="github",
+            identifier="acme/skills/skills/foo", trust_level="community",
+        )
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.json.return_value = [{"type": "dir", "name": "foo"}]
+
+        with patch.object(src, "_read_cache", return_value=None), \
+             patch.object(src, "_write_cache"), \
+             patch.object(src, "_get_skillsh_groupings", return_value=None), \
+             patch.object(src, "inspect", return_value=meta), \
+             patch("tools.skills_hub.httpx.get", return_value=resp):
+            skills = src._list_skills_in_repo("acme/skills", "skills/")
+
+        assert len(skills) == 1
+        assert "category" not in skills[0].extra
+
+    def test_meta_to_dict_roundtrip_preserves_extra(self):
+        meta = SkillMeta(
+            name="x", description="d", source="github",
+            identifier="acme/skills/x", trust_level="trusted",
+            extra={"category": "Inference AI"},
+        )
+        d = GitHubSource._meta_to_dict(meta)
+        assert d["extra"] == {"category": "Inference AI"}
+        # Round-trips back through the cache deserialization path.
+        restored = SkillMeta(**d)
+        assert restored.extra == {"category": "Inference AI"}
+
+
 # ---------------------------------------------------------------------------
 # GitHubSource.trust_level_for
 # ---------------------------------------------------------------------------
@@ -103,6 +239,36 @@ class TestTrustLevelFor:
         # No path part — still resolves repo correctly
         assert result in {"trusted", "community"}
 
+    def test_nvidia_skills_tap_is_registered_and_trusted(self):
+        # Invariant: every trusted repo in TRUSTED_REPOS that we want
+        # browseable/searchable through `hermes skills browse` must also
+        # appear as a default tap on GitHubSource. Without the tap, the
+        # repo's skills don't show up in search results or the docs-site
+        # Skills Hub page even though the trust level is correct.
+        from tools.skills_guard import TRUSTED_REPOS
+
+        assert "NVIDIA/skills" in TRUSTED_REPOS
+        tap_repos = {tap["repo"] for tap in GitHubSource.DEFAULT_TAPS}
+        assert "NVIDIA/skills" in tap_repos
+
+        src = self._source()
+        assert src.trust_level_for("NVIDIA/skills/aiq-deploy") == "trusted"
+
+    def test_browseable_trusted_repos_have_taps(self):
+        # General invariant covering all current and future trusted repos
+        # that publish under a single `skills/`-style path. openai/skills
+        # is the deliberate exception — it has two taps (`.curated/` and
+        # `.system/`) — so we just assert membership not path equality.
+        from tools.skills_guard import TRUSTED_REPOS
+
+        tap_repos = {tap["repo"] for tap in GitHubSource.DEFAULT_TAPS}
+        for repo in TRUSTED_REPOS:
+            assert repo in tap_repos, (
+                f"Trusted repo {repo!r} is in TRUSTED_REPOS but missing "
+                "from GitHubSource.DEFAULT_TAPS — its skills will not be "
+                "browsable via `hermes skills browse`."
+            )
+
 
 # ---------------------------------------------------------------------------
 # SkillsShSource
@@ -472,6 +638,68 @@ class TestSkillsShSource:
         requested_urls = [call.args[0] for call in mock_get.call_args_list]
         assert root_url not in requested_urls
 
+    @patch("tools.skills_hub._write_index_cache")
+    @patch("tools.skills_hub._read_index_cache", return_value=None)
+    @patch("tools.skills_hub.httpx.get")
+    def test_empty_query_walks_sitemap_not_homepage(
+        self, mock_get, _mock_read_cache, _mock_write_cache,
+    ):
+        """Empty query must walk the full sitemap.
+
+        Regression for skills.sh shipping ~858/20000 skills: the previous
+        empty-query path scraped the homepage's featured strip (~200 entries),
+        and build_skills_index.py supplemented it with 28 popular keyword
+        searches to drag the count to ~850. The sitemap walker hits the
+        full ~20k catalog in one pass.
+        """
+        index_xml = """<?xml version="1.0" encoding="UTF-8"?>
+<sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
+  <sitemap><loc>https://www.skills.sh/sitemap-misc.xml</loc></sitemap>
+  <sitemap><loc>https://www.skills.sh/sitemap-skills-1.xml</loc></sitemap>
+  <sitemap><loc>https://www.skills.sh/sitemap-skills-2.xml</loc></sitemap>
+</sitemapindex>"""
+        skills_1_xml = """<?xml version="1.0" encoding="UTF-8"?>
+<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
+  <url><loc>https://www.skills.sh/anthropics/skills/frontend-design</loc></url>
+  <url><loc>https://www.skills.sh/anthropics/skills/pdf</loc></url>
+  <url><loc>https://www.skills.sh/vercel-labs/agent-skills/react-best-practices</loc></url>
+</urlset>"""
+        skills_2_xml = """<?xml version="1.0" encoding="UTF-8"?>
+<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
+  <url><loc>https://www.skills.sh/microsoft/azure-skills/azure-ai</loc></url>
+  <url><loc>https://www.skills.sh/anthropics/skills/frontend-design</loc></url>
+</urlset>"""
+
+        def side_effect(url, *args, **kwargs):
+            resp = MagicMock(status_code=200)
+            if url.endswith("/sitemap.xml"):
+                resp.text = index_xml
+            elif "sitemap-skills-1" in url:
+                resp.text = skills_1_xml
+            elif "sitemap-skills-2" in url:
+                resp.text = skills_2_xml
+            else:
+                resp.status_code = 404
+                resp.text = ""
+            return resp
+
+        mock_get.side_effect = side_effect
+
+        results = self._source().search("", limit=0)
+
+        # 4 unique skills (the frontend-design dup across sitemaps collapsed).
+        assert len(results) == 4
+        identifiers = {r.identifier for r in results}
+        assert identifiers == {
+            "skills-sh/anthropics/skills/frontend-design",
+            "skills-sh/anthropics/skills/pdf",
+            "skills-sh/vercel-labs/agent-skills/react-best-practices",
+            "skills-sh/microsoft/azure-skills/azure-ai",
+        }
+        # Homepage was NOT fetched — the sitemap path is taken on empty query.
+        urls_called = [call.args[0] for call in mock_get.call_args_list]
+        assert not any(u == "https://skills.sh" or u == "https://skills.sh/" for u in urls_called)
+
 
 class TestFindSkillInRepoTree:
     """Tests for GitHubSource._find_skill_in_repo_tree."""

tests/tools/test_skills_hub_clawhub.py

@@ -298,6 +298,58 @@ class TestClawHubSource(unittest.TestCase):
         self.assertIsNone(bundle)
         self.assertEqual(mock_get.call_count, 3)
 
+    @patch("tools.skills_hub._write_index_cache")
+    @patch("tools.skills_hub._read_index_cache", return_value=None)
+    @patch("tools.skills_hub.httpx.get")
+    def test_search_empty_query_paginates_full_catalog(
+        self, mock_get, _mock_read_cache, _mock_write_cache
+    ):
+        """Empty query must walk the cursor-paginated catalog.
+
+        Regression for the silent 200-skill truncation: ClawHub's listing
+        endpoint caps any single page at 200 items + returns a `nextCursor`.
+        The build_skills_index.py crawler calls `search("", limit=N)` with a
+        large N to dump the full catalog. Before the fix, that hit a single
+        unpaginated request and silently dropped 99% of the catalog.
+        """
+        # Three pages: 200 + 200 + 50 items, then no cursor → stop.
+        page_calls = {"n": 0}
+        pages = [
+            {
+                "items": [{"slug": f"a-skill-{i}", "displayName": f"A {i}"} for i in range(200)],
+                "nextCursor": "cursor-page-2",
+            },
+            {
+                "items": [{"slug": f"b-skill-{i}", "displayName": f"B {i}"} for i in range(200)],
+                "nextCursor": "cursor-page-3",
+            },
+            {
+                "items": [{"slug": f"c-skill-{i}", "displayName": f"C {i}"} for i in range(50)],
+                "nextCursor": None,
+            },
+        ]
+
+        def side_effect(url, *args, **kwargs):
+            if url.endswith("/skills"):
+                idx = page_calls["n"]
+                page_calls["n"] += 1
+                if idx < len(pages):
+                    return _MockResponse(status_code=200, json_data=pages[idx])
+                return _MockResponse(status_code=200, json_data={"items": []})
+            return _MockResponse(status_code=404, json_data={})
+
+        mock_get.side_effect = side_effect
+
+        results = self.src.search("", limit=10_000)
+
+        # 200 + 200 + 50 = 450 unique skills, all retrieved via cursor pagination.
+        self.assertEqual(len(results), 450)
+        self.assertEqual(page_calls["n"], 3, "expected exactly 3 cursor-paginated pages")
+        identifiers = {meta.identifier for meta in results}
+        self.assertIn("a-skill-0", identifiers)
+        self.assertIn("b-skill-199", identifiers)
+        self.assertIn("c-skill-49", identifiers)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/tools/test_skills_sync.py

@@ -15,8 +15,6 @@ from tools.skills_sync import (
     sync_skills,
     reset_bundled_skill,
     restore_official_optional_skill,
-    MANIFEST_FILE,
-    SKILLS_DIR,
 )
 
 

tests/tools/test_ssh_bulk_upload.py

@@ -2,7 +2,6 @@
 
 import os
 import subprocess
-from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest

tests/tools/test_stage2_hook_puid_pgid.py

@@ -0,0 +1,86 @@
+"""Contract test: the s6-overlay stage2 hook accepts PUID/PGID as aliases for
+HERMES_UID/HERMES_GID.
+
+Regression guard for #15290.  NAS platforms (UGOS, Synology, unRAID) bind-mount
+/opt/data from a host directory owned by the user's own UID and expect the
+LinuxServer.io PUID/PGID convention.  Without the alias those vars are silently
+ignored, the s6-setuidgid drop lands on UID 10000, and the runtime cannot read
+the volume.  HERMES_UID/HERMES_GID must still take precedence when both are
+set.
+
+The s6-overlay rework moved bootstrap from docker/entrypoint.sh (now a shim)
+to docker/stage2-hook.sh, which is installed as /etc/cont-init.d/01-hermes-setup
+by the Dockerfile.  This test targets the post-rework location.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+STAGE2_HOOK = REPO_ROOT / "docker" / "stage2-hook.sh"
+
+
+@pytest.fixture(scope="module")
+def stage2_text() -> str:
+    if not STAGE2_HOOK.exists():
+        pytest.skip("docker/stage2-hook.sh not present in this checkout")
+    return STAGE2_HOOK.read_text()
+
+
+def _alias_lines(text: str) -> list[str]:
+    """The stage2 hook lines that resolve HERMES_UID/HERMES_GID from aliases."""
+    return [
+        line.strip()
+        for line in text.splitlines()
+        if line.strip().startswith(("HERMES_UID=", "HERMES_GID="))
+    ]
+
+
+def test_stage2_hook_resolves_puid_pgid_aliases(stage2_text: str) -> None:
+    alias_lines = _alias_lines(stage2_text)
+    assert any("PUID" in line for line in alias_lines), (
+        "docker/stage2-hook.sh must resolve HERMES_UID from a PUID alias; see #15290"
+    )
+    assert any("PGID" in line for line in alias_lines), (
+        "docker/stage2-hook.sh must resolve HERMES_GID from a PGID alias; see #15290"
+    )
+
+
+def _resolve(stage2_text: str, env: dict[str, str]) -> str:
+    """Run the stage2 hook's alias-resolution lines in isolation and report the
+    resolved ``HERMES_UID:HERMES_GID`` pair."""
+    bash = shutil.which("bash")
+    if bash is None:
+        pytest.skip("bash not available")
+    script = "\n".join(_alias_lines(stage2_text))
+    script += '\necho "${HERMES_UID:-}:${HERMES_GID:-}"\n'
+    proc = subprocess.run(
+        [bash, "-ec", script],
+        env={"PATH": os.environ.get("PATH", "")} | env,
+        capture_output=True,
+        text=True,
+    )
+    assert proc.returncode == 0, proc.stderr
+    return proc.stdout.strip()
+
+
+def test_puid_pgid_populate_hermes_uid_gid(stage2_text: str) -> None:
+    assert _resolve(stage2_text, {"PUID": "1000", "PGID": "10"}) == "1000:10"
+
+
+def test_hermes_uid_gid_take_precedence_over_aliases(stage2_text: str) -> None:
+    resolved = _resolve(
+        stage2_text,
+        {"HERMES_UID": "2000", "HERMES_GID": "2001", "PUID": "1000", "PGID": "10"},
+    )
+    assert resolved == "2000:2001"
+
+
+def test_no_uid_vars_leaves_values_empty(stage2_text: str) -> None:
+    # An empty resolution means the stage2 hook keeps the default hermes user.
+    assert _resolve(stage2_text, {}) == ":"

tests/tools/test_symlink_prefix_confusion.py

@@ -6,7 +6,6 @@ for 'axolotl/' because the string prefix matched. Now uses
 Path.is_relative_to() which handles directory boundaries correctly.
 """
 
-import os
 import pytest
 from pathlib import Path
 

tests/tools/test_sync_back_backends.py

@@ -3,7 +3,7 @@
 import asyncio
 import subprocess
 from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock, call, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 

tests/tools/test_terminal_compound_background.py

@@ -12,7 +12,6 @@ The rewriter fixes this by wrapping the tail in a brace group —
 the current shell. No subshell fork, no wait.
 """
 
-import pytest
 
 from tools.terminal_tool import _rewrite_compound_background as rewrite
 

tests/tools/test_terminal_config_env_sync.py

@@ -224,3 +224,39 @@ def test_docker_env_is_bridged_everywhere():
     assert "docker_env" in _gateway_env_map_keys()
     assert "docker_env" in _save_config_env_sync_keys()
     assert "TERMINAL_DOCKER_ENV" in _terminal_tool_env_var_names()
+
+
+def test_docker_persist_across_processes_is_bridged_everywhere():
+    """Regression pin for the cross-process container reuse toggle.
+
+    ``terminal.docker_persist_across_processes`` (issue #20561) controls
+    whether ``DockerEnvironment.__init__`` probes for and reuses an existing
+    labeled container at startup, and whether ``cleanup()`` removes the
+    container on Hermes exit or just stops it (keeping it for the next
+    process).  Same four-bridge invariant as docker_run_as_host_user /
+    docker_env / docker_mount_cwd_to_workspace — drift between any of the
+    four sites means ``terminal.docker_persist_across_processes: false`` in
+    config.yaml silently does nothing for that entry point, leaving the
+    user unable to opt out of the documented "ONE long-lived container
+    shared across sessions" behavior.
+    """
+    assert "docker_persist_across_processes" in _cli_env_map_keys()
+    assert "docker_persist_across_processes" in _gateway_env_map_keys()
+    assert "docker_persist_across_processes" in _save_config_env_sync_keys()
+    assert "TERMINAL_DOCKER_PERSIST_ACROSS_PROCESSES" in _terminal_tool_env_var_names()
+
+
+def test_docker_orphan_reaper_is_bridged_everywhere():
+    """Regression pin for the startup orphan reaper toggle (issue #20561).
+
+    ``terminal.docker_orphan_reaper`` controls whether Hermes sweeps stale
+    Exited containers from prior SIGKILL'd processes at startup.  Same
+    four-site bridge invariant — drift means
+    ``terminal.docker_orphan_reaper: false`` silently does nothing for one
+    entry point, and the reaper either runs when the operator disabled it
+    or fails to run when they enabled it.
+    """
+    assert "docker_orphan_reaper" in _cli_env_map_keys()
+    assert "docker_orphan_reaper" in _gateway_env_map_keys()
+    assert "docker_orphan_reaper" in _save_config_env_sync_keys()
+    assert "TERMINAL_DOCKER_ORPHAN_REAPER" in _terminal_tool_env_var_names()

tests/tools/test_terminal_foreground_timeout_cap.py

@@ -4,7 +4,6 @@ Ensures that foreground commands with timeout > FOREGROUND_MAX_TIMEOUT
 are rejected with an error suggesting background=true.
 """
 import json
-import os
 from unittest.mock import patch, MagicMock
 
 
@@ -123,7 +122,7 @@ class TestForegroundTimeoutCap:
         Only the model's explicit timeout parameter triggers rejection,
         not the user's configured default.
         """
-        from tools.terminal_tool import terminal_tool, FOREGROUND_MAX_TIMEOUT
+        from tools.terminal_tool import terminal_tool
 
         # User configured TERMINAL_TIMEOUT=900 in their env
         with patch("tools.terminal_tool._get_env_config",