Merge remote-tracking branch 'origin/main' into bb/gui

# Conflicts:
#	apps/dashboard/src/i18n/af.ts
#	apps/dashboard/src/i18n/de.ts
#	apps/dashboard/src/i18n/es.ts
#	apps/dashboard/src/i18n/fr.ts
#	apps/dashboard/src/i18n/ga.ts
#	apps/dashboard/src/i18n/hu.ts
#	apps/dashboard/src/i18n/it.ts
#	apps/dashboard/src/i18n/ja.ts
#	apps/dashboard/src/i18n/ko.ts
#	apps/dashboard/src/i18n/pt.ts
#	apps/dashboard/src/i18n/ru.ts
#	apps/dashboard/src/i18n/tr.ts
#	apps/dashboard/src/i18n/uk.ts
#	apps/dashboard/src/i18n/zh-hant.ts
#	gateway/config.py
#	hermes_cli/main.py
#	plugins/strike-freedom-cockpit/README.md
#	tui_gateway/server.py

tests/tools/test_approval.py

@@ -965,3 +965,140 @@ class TestFailClosedUnderPromptToolkit:
             assert result == "once"
         finally:
             ptc.get_app_or_none = orig
+
+
+class TestDetectSudoStdin:
+    """Sudo with stdin / askpass / shell / list-privileges flags (#17873 cat 4).
+
+    An LLM-driven agent has no TTY, so the sudo invocations that succeed
+    without human interaction are those reading the password from stdin
+    (-S / --stdin) or via an askpass helper (-A / --askpass). The
+    shell-launch (-s) and list-privileges (-a) flags are also gated since
+    they are privilege-relevant invocations the agent can chain after
+    acquiring the password.
+
+    `_normalize_command_for_detection` lowercases input before pattern
+    matching, so -S/-s and -A/-a are indistinguishable at the regex
+    layer; both letter-pairs are gated.
+    """
+
+    # Positive cases (must match)
+
+    def test_canonical_pipe_to_sudo_S_detected(self):
+        is_dangerous, _, desc = detect_dangerous_command(
+            "echo pwd | sudo -S whoami"
+        )
+        assert is_dangerous is True
+        assert "sudo" in desc.lower()
+
+    def test_long_flag_stdin_detected(self):
+        is_dangerous, _, _ = detect_dangerous_command("sudo --stdin id")
+        assert is_dangerous is True
+
+    def test_non_interactive_plus_stdin_detected(self):
+        is_dangerous, _, _ = detect_dangerous_command("sudo -n -S id")
+        assert is_dangerous is True
+
+    def test_user_then_stdin_detected(self):
+        # Codex audit caught that the original "leading flags only" regex
+        # missed this form because `-u root` has a flag-argument (`root`)
+        # that broke the (?:\s+-[^\s]+)* loop. The lazy [^;|&\n]*? class
+        # consumes flag-args without spanning command separators.
+        is_dangerous, _, _ = detect_dangerous_command(
+            "sudo -u root -S whoami"
+        )
+        assert is_dangerous is True
+
+    def test_long_non_interactive_plus_stdin_detected(self):
+        is_dangerous, _, _ = detect_dangerous_command(
+            "sudo --non-interactive -S whoami"
+        )
+        assert is_dangerous is True
+
+    def test_long_user_equals_stdin_detected(self):
+        is_dangerous, _, _ = detect_dangerous_command(
+            "sudo --user=root -S id"
+        )
+        assert is_dangerous is True
+
+    def test_herestring_input_detected(self):
+        is_dangerous, _, _ = detect_dangerous_command(
+            "sudo -S id <<< 'mypwd'"
+        )
+        assert is_dangerous is True
+
+    def test_combined_short_flags_nS_detected(self):
+        # `-nS` packs `-n` and `-S` into one arg; second pattern catches.
+        is_dangerous, _, _ = detect_dangerous_command("sudo -nS id")
+        assert is_dangerous is True
+
+    def test_printf_form_detected(self):
+        is_dangerous, _, _ = detect_dangerous_command(
+            'printf "%s\\n" "$PW" | sudo -S id'
+        )
+        assert is_dangerous is True
+
+    def test_askpass_short_flag_detected(self):
+        is_dangerous, _, _ = detect_dangerous_command("sudo -A id")
+        assert is_dangerous is True
+
+    def test_askpass_long_flag_detected(self):
+        is_dangerous, _, _ = detect_dangerous_command("sudo --askpass id")
+        assert is_dangerous is True
+
+    def test_two_sudo_invocations_second_caught(self):
+        # The first sudo here is benign (no -S); the second has -S.
+        # Lazy [^;|&\n]*? does NOT span past `;`, so re.search anchors
+        # on the second sudo invocation independently.
+        is_dangerous, _, _ = detect_dangerous_command(
+            "sudo whoami; sudo -S id"
+        )
+        assert is_dangerous is True
+
+    # Negative cases (must NOT match)
+
+    def test_plain_sudo_safe(self):
+        is_dangerous, _, _ = detect_dangerous_command("sudo whoami")
+        assert is_dangerous is False
+
+    def test_sudo_interactive_shell_safe(self):
+        is_dangerous, _, _ = detect_dangerous_command("sudo -i")
+        assert is_dangerous is False
+
+    def test_sudo_with_user_no_stdin_flag_safe(self):
+        is_dangerous, _, _ = detect_dangerous_command("sudo -u root -i")
+        assert is_dangerous is False
+
+    def test_man_sudo_safe(self):
+        is_dangerous, _, _ = detect_dangerous_command("man sudo")
+        assert is_dangerous is False
+
+    def test_which_sudo_safe(self):
+        is_dangerous, _, _ = detect_dangerous_command("which sudo")
+        assert is_dangerous is False
+
+    def test_sudo_user_env_reference_safe(self):
+        is_dangerous, _, _ = detect_dangerous_command(
+            "echo SUDO_USER=$SUDO_USER"
+        )
+        assert is_dangerous is False
+
+    def test_apt_install_sudo_safe(self):
+        is_dangerous, _, _ = detect_dangerous_command("apt install sudo")
+        assert is_dangerous is False
+
+    def test_ls_etc_sudoers_safe(self):
+        is_dangerous, _, _ = detect_dangerous_command("ls /etc/sudoers")
+        assert is_dangerous is False
+
+    def test_pseudosudo_safe_word_boundary(self):
+        # `\bsudo\b` requires a word boundary; `pseudosudo` has none
+        # before `sudo`, so should not trigger.
+        is_dangerous, _, _ = detect_dangerous_command("pseudosudo -S id")
+        assert is_dangerous is False
+
+    def test_unrelated_redirection_safe(self):
+        is_dangerous, _, _ = detect_dangerous_command(
+            "make 2>&1 | tee build.log"
+        )
+        assert is_dangerous is False

tests/tools/test_browser_eval_supervisor_path.py

@@ -0,0 +1,363 @@
+"""Unit tests for the supervisor-WS fast path in browser_console / _browser_eval.
+
+These exercise the dispatch logic in ``tools.browser_tool._browser_eval`` and
+the response shaping in ``CDPSupervisor.evaluate_runtime`` using mocks — no
+real browser, no real WebSocket.  Real-CDP coverage lives in
+``tests/tools/test_browser_supervisor.py`` (gated on Chrome being installed).
+"""
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Fast-path dispatch: tools.browser_tool._browser_eval
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _disable_camofox(monkeypatch):
+    """Force the non-camofox path so our supervisor branch is reached."""
+    import tools.browser_tool as bt
+
+    monkeypatch.setattr(bt, "_is_camofox_mode", lambda: False)
+    monkeypatch.setattr(bt, "_last_session_key", lambda task_id: "test-task")
+
+
+def _patch_supervisor(monkeypatch, supervisor):
+    """Wire SUPERVISOR_REGISTRY.get to return ``supervisor`` for any task_id."""
+    import tools.browser_supervisor as bs
+
+    registry = MagicMock()
+    registry.get.return_value = supervisor
+    monkeypatch.setattr(bs, "SUPERVISOR_REGISTRY", registry)
+    return registry
+
+
+class TestBrowserEvalSupervisorPath:
+    """The supervisor fast path replaces the agent-browser subprocess hop."""
+
+    def test_primitive_result_routes_through_supervisor(self, monkeypatch):
+        import tools.browser_tool as bt
+
+        sup = MagicMock()
+        sup.evaluate_runtime.return_value = {
+            "ok": True,
+            "result": 42,
+            "result_type": "number",
+        }
+        _patch_supervisor(monkeypatch, sup)
+        # If the subprocess path is hit we want a loud failure.
+        monkeypatch.setattr(
+            bt, "_run_browser_command",
+            lambda *a, **kw: pytest.fail("subprocess path must not run when supervisor is healthy"),
+        )
+
+        out = json.loads(bt._browser_eval("1 + 41"))
+        assert out["success"] is True
+        assert out["result"] == 42
+        assert out["method"] == "cdp_supervisor"
+        sup.evaluate_runtime.assert_called_once_with("1 + 41")
+
+    def test_json_string_result_is_parsed(self, monkeypatch):
+        """Match agent-browser semantics: JSON-string results get parsed."""
+        import tools.browser_tool as bt
+
+        sup = MagicMock()
+        sup.evaluate_runtime.return_value = {
+            "ok": True,
+            "result": '{"a": 1, "b": [2, 3]}',
+            "result_type": "string",
+        }
+        _patch_supervisor(monkeypatch, sup)
+        monkeypatch.setattr(
+            bt, "_run_browser_command",
+            lambda *a, **kw: pytest.fail("subprocess path must not run"),
+        )
+
+        out = json.loads(bt._browser_eval('JSON.stringify({a:1,b:[2,3]})'))
+        assert out["success"] is True
+        assert out["result"] == {"a": 1, "b": [2, 3]}
+        # result_type reflects the parsed Python type, not the raw JS type.
+        assert out["result_type"] == "dict"
+
+    def test_non_json_string_result_kept_as_string(self, monkeypatch):
+        import tools.browser_tool as bt
+
+        sup = MagicMock()
+        sup.evaluate_runtime.return_value = {
+            "ok": True,
+            "result": "hello world",
+            "result_type": "string",
+        }
+        _patch_supervisor(monkeypatch, sup)
+        monkeypatch.setattr(bt, "_run_browser_command", lambda *a, **kw: pytest.fail("nope"))
+
+        out = json.loads(bt._browser_eval('"hello world"'))
+        assert out["result"] == "hello world"
+        assert out["result_type"] == "str"
+
+    def test_js_exception_surfaces_without_subprocess_fallthrough(self, monkeypatch):
+        """A JS-side error must NOT trigger a (slow + redundant) subprocess retry."""
+        import tools.browser_tool as bt
+
+        sup = MagicMock()
+        sup.evaluate_runtime.return_value = {
+            "ok": False,
+            "error": "Uncaught ReferenceError: foo is not defined",
+        }
+        _patch_supervisor(monkeypatch, sup)
+        called = {"subprocess": False}
+
+        def _fake_subprocess(*a, **kw):
+            called["subprocess"] = True
+            return {"success": True, "data": {"result": "should-not-be-used"}}
+
+        monkeypatch.setattr(bt, "_run_browser_command", _fake_subprocess)
+
+        out = json.loads(bt._browser_eval("foo.bar"))
+        assert out["success"] is False
+        assert "ReferenceError" in out["error"]
+        assert called["subprocess"] is False, \
+            "JS exception should be surfaced, not retried via subprocess"
+
+    def test_supervisor_loop_down_falls_through_to_subprocess(self, monkeypatch):
+        """When the supervisor itself is unavailable, fall back to the subprocess."""
+        import tools.browser_tool as bt
+
+        sup = MagicMock()
+        sup.evaluate_runtime.return_value = {
+            "ok": False,
+            "error": "supervisor loop is not running",
+        }
+        _patch_supervisor(monkeypatch, sup)
+
+        called = {"subprocess": False}
+
+        def _fake_subprocess(task_id, cmd, args):
+            called["subprocess"] = True
+            assert cmd == "eval"
+            return {"success": True, "data": {"result": "fallback-result"}}
+
+        monkeypatch.setattr(bt, "_run_browser_command", _fake_subprocess)
+
+        out = json.loads(bt._browser_eval("anything"))
+        assert called["subprocess"] is True
+        assert out["success"] is True
+        assert out["result"] == "fallback-result"
+        # Subprocess path doesn't tag the response with method=cdp_supervisor.
+        assert out.get("method") != "cdp_supervisor"
+
+    def test_no_active_supervisor_falls_through_to_subprocess(self, monkeypatch):
+        """When SUPERVISOR_REGISTRY.get returns None, subprocess path runs."""
+        import tools.browser_tool as bt
+
+        _patch_supervisor(monkeypatch, None)
+        called = {"subprocess": False}
+
+        def _fake_subprocess(task_id, cmd, args):
+            called["subprocess"] = True
+            return {"success": True, "data": {"result": "agent-browser-result"}}
+
+        monkeypatch.setattr(bt, "_run_browser_command", _fake_subprocess)
+
+        out = json.loads(bt._browser_eval("1+1"))
+        assert called["subprocess"] is True
+        assert out["success"] is True
+        assert out.get("method") != "cdp_supervisor"
+
+    def test_supervisor_no_session_falls_through(self, monkeypatch):
+        """A supervisor without an attached page session must fall through cleanly."""
+        import tools.browser_tool as bt
+
+        sup = MagicMock()
+        sup.evaluate_runtime.return_value = {
+            "ok": False,
+            "error": "supervisor has no attached page session",
+        }
+        _patch_supervisor(monkeypatch, sup)
+        called = {"subprocess": False}
+
+        def _fake_subprocess(*a, **kw):
+            called["subprocess"] = True
+            return {"success": True, "data": {"result": "fallback"}}
+
+        monkeypatch.setattr(bt, "_run_browser_command", _fake_subprocess)
+        json.loads(bt._browser_eval("1+1"))
+        assert called["subprocess"] is True
+
+
+# ---------------------------------------------------------------------------
+# Response shaping: CDPSupervisor.evaluate_runtime
+# ---------------------------------------------------------------------------
+
+
+def _make_supervisor_with_cdp(cdp_response):
+    """Build a CDPSupervisor instance that mocks ``_cdp`` to return ``cdp_response``.
+
+    Bypasses ``__init__`` entirely so we don't need a real WS connection.  We
+    set just the state ``evaluate_runtime`` reads.
+    """
+    import asyncio
+    import threading
+
+    from tools.browser_supervisor import CDPSupervisor
+
+    sup = object.__new__(CDPSupervisor)
+    sup._state_lock = threading.Lock()
+    sup._active = True
+    sup._page_session_id = "test-session-id"
+
+    # Build a real running event loop on a background thread so
+    # asyncio.run_coroutine_threadsafe has somewhere to dispatch.
+    loop = asyncio.new_event_loop()
+
+    def _runner():
+        asyncio.set_event_loop(loop)
+        loop.run_forever()
+
+    thread = threading.Thread(target=_runner, daemon=True)
+    thread.start()
+
+    async def _fake_cdp(method, params=None, *, session_id=None, timeout=10.0):
+        return cdp_response
+
+    sup._cdp = _fake_cdp  # type: ignore[method-assign]
+    sup._loop = loop
+    sup._thread = thread
+    return sup
+
+
+def _stop_supervisor(sup):
+    sup._loop.call_soon_threadsafe(sup._loop.stop)
+    sup._thread.join(timeout=2)
+
+
+class TestEvaluateRuntimeResponseShaping:
+    """CDPSupervisor.evaluate_runtime decodes the Runtime.evaluate response correctly."""
+
+    def test_primitive_value(self):
+        sup = _make_supervisor_with_cdp({
+            "id": 1,
+            "result": {"result": {"type": "number", "value": 42}},
+        })
+        try:
+            out = sup.evaluate_runtime("1 + 41")
+            assert out == {"ok": True, "result": 42, "result_type": "number"}
+        finally:
+            _stop_supervisor(sup)
+
+    def test_object_value_returned_by_value(self):
+        sup = _make_supervisor_with_cdp({
+            "id": 1,
+            "result": {
+                "result": {
+                    "type": "object",
+                    "value": {"foo": "bar", "n": 7},
+                }
+            },
+        })
+        try:
+            out = sup.evaluate_runtime('({foo:"bar", n:7})')
+            assert out["ok"] is True
+            assert out["result"] == {"foo": "bar", "n": 7}
+            assert out["result_type"] == "object"
+        finally:
+            _stop_supervisor(sup)
+
+    def test_undefined_value(self):
+        sup = _make_supervisor_with_cdp({
+            "id": 1,
+            "result": {"result": {"type": "undefined"}},
+        })
+        try:
+            out = sup.evaluate_runtime("undefined")
+            assert out == {"ok": True, "result": None, "result_type": "undefined"}
+        finally:
+            _stop_supervisor(sup)
+
+    def test_dom_node_returns_description(self):
+        """Non-serializable values (DOM nodes, functions) come back as description strings."""
+        sup = _make_supervisor_with_cdp({
+            "id": 1,
+            "result": {
+                "result": {
+                    "type": "object",
+                    "subtype": "node",
+                    "description": "div#main.app",
+                    # No 'value' key — returnByValue couldn't serialize it.
+                }
+            },
+        })
+        try:
+            out = sup.evaluate_runtime("document.querySelector('#main')")
+            assert out["ok"] is True
+            assert out["result"] == "div#main.app"
+            assert out["result_type"] == "object"
+        finally:
+            _stop_supervisor(sup)
+
+    def test_js_exception_returns_error(self):
+        sup = _make_supervisor_with_cdp({
+            "id": 1,
+            "result": {
+                "result": {"type": "undefined"},
+                "exceptionDetails": {
+                    "text": "Uncaught",
+                    "exception": {
+                        "description": "ReferenceError: foo is not defined",
+                    },
+                },
+            },
+        })
+        try:
+            out = sup.evaluate_runtime("foo.bar")
+            assert out["ok"] is False
+            assert "ReferenceError" in out["error"]
+        finally:
+            _stop_supervisor(sup)
+
+    def test_inactive_supervisor_returns_error_without_dispatch(self):
+        """Inactive supervisor short-circuits before even touching the loop."""
+        import threading
+        from tools.browser_supervisor import CDPSupervisor
+
+        sup = object.__new__(CDPSupervisor)
+        sup._state_lock = threading.Lock()
+        sup._active = False  # ← key
+        sup._page_session_id = None
+        sup._loop = None
+
+        out = sup.evaluate_runtime("1+1")
+        assert out["ok"] is False
+        # Either "loop is not running" or "is not active" is acceptable —
+        # both are caught by the supervisor-side error branch in _browser_eval.
+        assert "supervisor" in out["error"].lower()
+
+    def test_no_session_attached_returns_error(self):
+        import asyncio
+        import threading
+        from tools.browser_supervisor import CDPSupervisor
+
+        sup = object.__new__(CDPSupervisor)
+        sup._state_lock = threading.Lock()
+        sup._active = True
+        sup._page_session_id = None  # ← attach hasn't happened yet
+
+        loop = asyncio.new_event_loop()
+        thread = threading.Thread(
+            target=lambda: (asyncio.set_event_loop(loop), loop.run_forever()),
+            daemon=True,
+        )
+        thread.start()
+        sup._loop = loop
+        try:
+            out = sup.evaluate_runtime("1+1")
+            assert out["ok"] is False
+            assert "session" in out["error"].lower()
+        finally:
+            loop.call_soon_threadsafe(loop.stop)
+            thread.join(timeout=2)

tests/tools/test_browser_supervisor.py

@@ -561,3 +561,80 @@ def test_bridge_captures_prompt_and_returns_reply_text(chrome_cdp, supervisor_re
 
     value = asyncio.run(nav_and_read())
     assert value == "AGENT-SUPPLIED-REPLY", f"expected AGENT-SUPPLIED-REPLY, got {value!r}"
+
+
+def test_evaluate_runtime_primitive(chrome_cdp, supervisor_registry):
+    """evaluate_runtime returns primitive values via the supervisor's live WS."""
+    cdp_url, _port = chrome_cdp
+    supervisor = supervisor_registry.get_or_start(task_id="pytest-eval-1", cdp_url=cdp_url)
+
+    # Need a page to evaluate against.
+    _fire_on_page(cdp_url, "void 0")
+    time.sleep(0.5)
+
+    out = supervisor.evaluate_runtime("1 + 41")
+    assert out["ok"] is True
+    assert out["result"] == 42
+    assert out["result_type"] == "number"
+
+
+def test_evaluate_runtime_object(chrome_cdp, supervisor_registry):
+    """Plain objects come back JSON-serialized via returnByValue=True."""
+    cdp_url, _port = chrome_cdp
+    supervisor = supervisor_registry.get_or_start(task_id="pytest-eval-2", cdp_url=cdp_url)
+
+    _fire_on_page(cdp_url, "void 0")
+    time.sleep(0.5)
+
+    out = supervisor.evaluate_runtime('({foo: "bar", n: 7})')
+    assert out["ok"] is True
+    assert out["result"] == {"foo": "bar", "n": 7}
+    assert out["result_type"] == "object"
+
+
+def test_evaluate_runtime_js_exception(chrome_cdp, supervisor_registry):
+    """JS exceptions surface as ok=False with the exception message."""
+    cdp_url, _port = chrome_cdp
+    supervisor = supervisor_registry.get_or_start(task_id="pytest-eval-3", cdp_url=cdp_url)
+
+    _fire_on_page(cdp_url, "void 0")
+    time.sleep(0.5)
+
+    out = supervisor.evaluate_runtime("nonExistentVar.nope")
+    assert out["ok"] is False
+    assert "ReferenceError" in out["error"] or "not defined" in out["error"]
+
+
+def test_evaluate_runtime_dom_node_returns_empty_object(chrome_cdp, supervisor_registry):
+    """DOM nodes with returnByValue=true serialize to ``{}`` (Chrome quirk).
+
+    This is honest — DOM nodes can't be deeply JSON-serialized — and matches
+    DevTools console behaviour for the same expression.  Documenting the
+    contract here so a future change that "fixes" it (e.g. switching to
+    returnByValue=false + DOM.describeNode) doesn't break callers expecting
+    the current shape.
+    """
+    cdp_url, _port = chrome_cdp
+    supervisor = supervisor_registry.get_or_start(task_id="pytest-eval-4", cdp_url=cdp_url)
+
+    _fire_on_page(cdp_url, "void 0")
+    time.sleep(0.5)
+
+    out = supervisor.evaluate_runtime("document.querySelector('h1')")
+    assert out["ok"] is True
+    assert out["result_type"] == "object"
+    # Empty dict — Chrome can't deeply-serialize a DOM node through returnByValue.
+    assert out["result"] == {}
+
+
+def test_evaluate_runtime_unserializable_value(chrome_cdp, supervisor_registry):
+    """``Infinity``/``NaN``/``BigInt`` come back via ``unserializableValue``."""
+    cdp_url, _port = chrome_cdp
+    supervisor = supervisor_registry.get_or_start(task_id="pytest-eval-5", cdp_url=cdp_url)
+
+    _fire_on_page(cdp_url, "void 0")
+    time.sleep(0.5)
+
+    out = supervisor.evaluate_runtime("Infinity")
+    assert out["ok"] is True
+    assert out["result"] == "Infinity"

tests/tools/test_hardline_blocklist.py

@@ -288,3 +288,91 @@ def test_hardline_list_is_small():
         f"HARDLINE_PATTERNS has grown to {len(HARDLINE_PATTERNS)} entries; "
         "only truly unrecoverable commands belong here."
     )
+
+
+# =========================================================================
+# Sudo stdin guard — blocks "sudo -S" without SUDO_PASSWORD
+# =========================================================================
+
+_SUDO_STDIN_BLOCK = [
+    "sudo -S whoami",
+    "echo hunter2 | sudo -S whoami",
+    "sudo -S -u root whoami",
+    "sudo -S apt-get install foo",
+    "echo password | sudo -S systemctl restart nginx",
+    "sudo -k && sudo -S whoami",
+]
+
+_SUDO_STDIN_ALLOW = [
+    # Plain sudo without -S — goes through normal approval
+    "sudo whoami",
+    "sudo apt-get update",
+    "sudo -u root whoami",
+    # -S flag not attached to sudo
+    "echo -S hello",
+    "some_tool -S thing",
+    # Literal text mention of sudo
+    "echo 'use sudo -S to pipe passwords'",
+]
+
+_SUDO_STDIN_BLOCK_YOLO = [
+    "sudo -S whoami",
+    "echo hunter2 | sudo -S apt-get install",
+]
+
+
+def test_sudo_stdin_guard_detects_without_password():
+    """sudo -S is dangerous when SUDO_PASSWORD is not configured."""
+    import tools.approval as approval_mod
+
+    for cmd in _SUDO_STDIN_BLOCK:
+        is_blocked, desc = approval_mod._check_sudo_stdin_guard(cmd)
+        assert is_blocked, f"expected sudo stdin guard to block {cmd!r}"
+        assert "sudo" in desc.lower()
+
+
+def test_sudo_stdin_guard_allows_benign_commands():
+    """Commands without explicit sudo -S are not blocked."""
+    import tools.approval as approval_mod
+
+    for cmd in _SUDO_STDIN_ALLOW:
+        is_blocked, desc = approval_mod._check_sudo_stdin_guard(cmd)
+        assert not is_blocked, f"expected sudo stdin guard NOT to block {cmd!r}"
+
+
+def test_sudo_stdin_guard_bypassed_when_password_configured(monkeypatch):
+    """When SUDO_PASSWORD is set, sudo -S is legitimate (injected by transform)."""
+    import tools.approval as approval_mod
+
+    monkeypatch.setenv("SUDO_PASSWORD", "testpass")
+    for cmd in _SUDO_STDIN_BLOCK:
+        is_blocked, _ = approval_mod._check_sudo_stdin_guard(cmd)
+        assert not is_blocked, f"with SUDO_PASSWORD set, {cmd!r} should NOT be blocked"
+
+
+def test_sudo_stdin_guard_blocks_via_check_all_command_guards(clean_session):
+    """Integration: check_all_command_guards returns block for sudo -S."""
+    for cmd in _SUDO_STDIN_BLOCK:
+        result = check_all_command_guards(cmd, "local")
+        assert result["approved"] is False, f"expected block on {cmd!r}"
+        # Should NOT be marked as hardline (it's sudo-specific)
+        assert result.get("hardline") is not True
+        assert "BLOCKED" in result["message"]
+        assert "sudo -S" in result["message"].lower() or "sudo password" in result["message"].lower()
+
+
+def test_sudo_stdin_guard_not_blocked_by_yolo(clean_session, monkeypatch):
+    """yolo/approvals.mode=off must NOT bypass sudo stdin guard."""
+    monkeypatch.setenv("HERMES_YOLO_MODE", "1")
+
+    for cmd in _SUDO_STDIN_BLOCK_YOLO:
+        result = check_all_command_guards(cmd, "local")
+        assert result["approved"] is False, f"yolo leaked sudo guard on {cmd!r}"
+
+
+def test_sudo_stdin_guard_container_bypass(clean_session):
+    """Containerized backends still bypass — they can't touch the host."""
+    for env in ("docker", "singularity", "modal", "daytona", "vercel_sandbox"):
+        for cmd in _SUDO_STDIN_BLOCK:
+            result = check_all_command_guards(cmd, env)
+            assert result["approved"] is True, f"container {env} should bypass sudo guard on {cmd!r}"

tests/tools/test_kanban_tools.py

@@ -2,7 +2,7 @@
 
 Verifies:
   - Tools are gated on HERMES_KANBAN_TASK: a normal chat session sees
-    zero kanban tools in its schema; a worker session sees all seven.
+    zero kanban tools in its schema; a worker session sees the kanban set.
   - Each handler's happy path.
   - Error paths (missing required args, bad metadata type, etc).
 """
@@ -27,9 +27,10 @@ def test_kanban_tools_hidden_without_env_var(monkeypatch, tmp_path):
     monkeypatch.setenv("HERMES_HOME", str(home))
 
     import tools.kanban_tools  # ensure registered
-    from tools.registry import registry
+    from tools.registry import invalidate_check_fn_cache, registry
     from toolsets import resolve_toolset
 
+    invalidate_check_fn_cache()
     schema = registry.get_definitions(set(resolve_toolset("hermes-cli")), quiet=True)
     names = {s["function"].get("name") for s in schema if "function" in s}
     kanban = {n for n in names if n and n.startswith("kanban_")}
@@ -39,16 +40,17 @@ def test_kanban_tools_hidden_without_env_var(monkeypatch, tmp_path):
 
 
 def test_kanban_tools_visible_with_env_var(monkeypatch, tmp_path):
-    """Worker sessions (HERMES_KANBAN_TASK set) must have all 7 tools."""
+    """Worker sessions get task lifecycle tools, not board-routing tools."""
     monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake")
     home = tmp_path / ".hermes"
     home.mkdir()
     monkeypatch.setenv("HERMES_HOME", str(home))
 
     import tools.kanban_tools  # ensure registered
-    from tools.registry import registry
+    from tools.registry import invalidate_check_fn_cache, registry
     from toolsets import resolve_toolset
 
+    invalidate_check_fn_cache()
     schema = registry.get_definitions(set(resolve_toolset("hermes-cli")), quiet=True)
     names = {s["function"].get("name") for s in schema if "function" in s}
     kanban = {n for n in names if n and n.startswith("kanban_")}
@@ -59,6 +61,61 @@ def test_kanban_tools_visible_with_env_var(monkeypatch, tmp_path):
     assert kanban == expected, f"expected {expected}, got {kanban}"
 
 
+def test_worker_with_kanban_toolset_still_hides_board_routing(monkeypatch, tmp_path):
+    """Task scope wins over profile config for board-routing tools.
+
+    Even if a worker process happens to also have ``toolsets: [kanban]``
+    in its config, the HERMES_KANBAN_TASK env var means it's a focused
+    worker and must not see kanban_list / kanban_unblock.
+    """
+    monkeypatch.setenv("HERMES_KANBAN_TASK", "t_fake")
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    (home / "config.yaml").write_text("toolsets:\n  - kanban\n")
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    import tools.kanban_tools  # ensure registered
+    from tools.registry import invalidate_check_fn_cache, registry
+    from toolsets import resolve_toolset
+
+    invalidate_check_fn_cache()
+    schema = registry.get_definitions(set(resolve_toolset("hermes-cli")), quiet=True)
+    names = {s["function"].get("name") for s in schema if "function" in s}
+    kanban = {n for n in names if n and n.startswith("kanban_")}
+    assert {
+        "kanban_list",
+        "kanban_unblock",
+    }.isdisjoint(kanban), (
+        f"Board-routing tools leaked into worker schema: "
+        f"{kanban & {'kanban_list', 'kanban_unblock'}}"
+    )
+
+
+def test_kanban_tools_visible_with_toolset_config(monkeypatch, tmp_path):
+    """Orchestrator profiles with toolsets: [kanban] see all kanban tools."""
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    (home / "config.yaml").write_text("toolsets:\n  - kanban\n")
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    import tools.kanban_tools  # ensure registered
+    from tools.registry import invalidate_check_fn_cache, registry
+    from toolsets import resolve_toolset
+
+    invalidate_check_fn_cache()
+    schema = registry.get_definitions(set(resolve_toolset("hermes-cli")), quiet=True)
+    names = {s["function"].get("name") for s in schema if "function" in s}
+    kanban = {n for n in names if n and n.startswith("kanban_")}
+    expected = {
+        "kanban_list",
+        "kanban_show", "kanban_complete", "kanban_block", "kanban_heartbeat",
+        "kanban_comment", "kanban_create", "kanban_link",
+        "kanban_unblock",
+    }
+    assert kanban == expected, f"expected {expected}, got {kanban}"
+
+
 # ---------------------------------------------------------------------------
 # Handler happy paths
 # ---------------------------------------------------------------------------
@@ -112,6 +169,100 @@ def test_show_explicit_task_id(worker_env):
     assert d["task"]["id"] == other
 
 
+def test_list_filters_tasks(monkeypatch, worker_env):
+    """kanban_list gives orchestrators filtered board discovery."""
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        a = kb.create_task(conn, title="alpha", assignee="factory", priority=5)
+        b = kb.create_task(conn, title="beta", assignee="reviewer")
+        c = kb.create_task(conn, title="gamma", assignee="factory", tenant="other")
+    finally:
+        conn.close()
+
+    from tools import kanban_tools as kt
+    out = kt._handle_list({"assignee": "factory", "status": "ready", "limit": 10})
+    d = json.loads(out)
+    ids = [t["id"] for t in d["tasks"]]
+    assert ids == [a, c]
+    assert d["count"] == 2
+    assert d["tasks"][0]["title"] == "alpha"
+    assert d["tasks"][0]["parent_count"] == 0
+    assert b not in ids
+
+    tenant_out = kt._handle_list({
+        "assignee": "factory",
+        "status": "ready",
+        "tenant": "other",
+    })
+    tenant_ids = [t["id"] for t in json.loads(tenant_out)["tasks"]]
+    assert tenant_ids == [c]
+
+
+def test_list_rejects_invalid_status(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from tools import kanban_tools as kt
+    out = kt._handle_list({"status": "not-a-state"})
+    assert "status must be one of" in json.loads(out).get("error", "")
+
+
+def test_list_rejects_bad_limit(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from tools import kanban_tools as kt
+    assert json.loads(kt._handle_list({"limit": "nope"})).get("error")
+    assert json.loads(kt._handle_list({"limit": 0})).get("error")
+
+
+def test_list_parses_include_archived_string_false(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        live = kb.create_task(conn, title="live task", assignee="factory")
+        archived = kb.create_task(conn, title="archived task", assignee="factory")
+        assert kb.archive_task(conn, archived)
+    finally:
+        conn.close()
+
+    from tools import kanban_tools as kt
+    out = kt._handle_list({
+        "assignee": "factory",
+        "include_archived": "false",
+    })
+    ids = [t["id"] for t in json.loads(out)["tasks"]]
+    assert live in ids
+    assert archived not in ids
+
+
+def test_list_parses_include_archived_string_true(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        live = kb.create_task(conn, title="live task", assignee="factory")
+        archived = kb.create_task(conn, title="archived task", assignee="factory")
+        assert kb.archive_task(conn, archived)
+    finally:
+        conn.close()
+
+    from tools import kanban_tools as kt
+    out = kt._handle_list({
+        "assignee": "factory",
+        "include_archived": "true",
+    })
+    ids = [t["id"] for t in json.loads(out)["tasks"]]
+    assert live in ids
+    assert archived in ids
+
+
+def test_list_rejects_bad_include_archived(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from tools import kanban_tools as kt
+    out = kt._handle_list({"include_archived": "sometimes"})
+    assert "include_archived must be" in json.loads(out).get("error", "")
+
+
 def test_complete_happy_path(worker_env):
     from tools import kanban_tools as kt
     out = kt._handle_complete({
@@ -179,6 +330,106 @@ def test_complete_rejects_non_dict_metadata(worker_env):
     assert json.loads(out).get("error")
 
 
+def test_complete_phantom_card_message_advertises_retry(worker_env):
+    """A phantom-card rejection must surface a tool_error that explicitly
+    tells the worker the task is still in-flight and how to retry — the
+    worker has no other channel to discover that. Regression for #22923,
+    where the previous wording read like a terminal failure and workers
+    routinely abandoned the run instead of trying again.
+    """
+    from hermes_cli import kanban_db as kb
+    from tools import kanban_tools as kt
+
+    out = kt._handle_complete({
+        "summary": "oops claimed a phantom",
+        "created_cards": ["t_phantomdeadbeef"],
+    })
+    err = json.loads(out).get("error", "")
+    assert err, f"expected an error, got {out!r}"
+    # Phantom id surfaced verbatim.
+    assert "t_phantomdeadbeef" in err
+    # The retry-is-supported phrasing — these are the literal cues a
+    # worker reads to decide whether to retry vs block/abandon. If a
+    # future change rewords the message, these checks will catch the
+    # regression. See #22923 for the failure mode.
+    assert "still in-flight" in err
+    assert "Retry kanban_complete" in err
+    assert "created_cards=[]" in err
+
+    # Critically: the task is genuinely still in-flight — the gate
+    # rejection did not mutate state, so the worker's retry can land.
+    conn = kb.connect()
+    try:
+        assert kb.get_task(conn, worker_env).status == "running"
+    finally:
+        conn.close()
+
+
+def test_complete_retry_with_empty_created_cards_succeeds(worker_env):
+    """After a phantom rejection, retrying kanban_complete with
+    created_cards=[] (the documented escape hatch) must complete the
+    task. Regression for #22923."""
+    from hermes_cli import kanban_db as kb
+    from tools import kanban_tools as kt
+
+    # Hit the gate first.
+    rejected = json.loads(kt._handle_complete({
+        "summary": "oops",
+        "created_cards": ["t_phantomdeadbeef"],
+    }))
+    assert rejected.get("error")
+
+    # Retry with the escape hatch.
+    ok = json.loads(kt._handle_complete({
+        "summary": "retry without claims",
+        "created_cards": [],
+    }))
+    assert ok.get("ok") is True
+
+    conn = kb.connect()
+    try:
+        assert kb.get_task(conn, worker_env).status == "done"
+    finally:
+        conn.close()
+
+
+def test_complete_retry_with_corrected_created_cards_succeeds(worker_env):
+    """After a phantom rejection, retrying kanban_complete with a
+    corrected created_cards list (phantom ids removed) must complete the
+    task. Regression for #22923."""
+    from hermes_cli import kanban_db as kb
+    from tools import kanban_tools as kt
+
+    # Create a real child via the tool so it gets the worker-profile
+    # attribution the gate trusts.
+    child = json.loads(kt._handle_create({
+        "title": "real child", "assignee": "peer",
+    }))
+    assert child["ok"]
+    real_id = child["task_id"]
+
+    # First attempt mixes real + phantom — gate rejects.
+    rejected = json.loads(kt._handle_complete({
+        "summary": "oops",
+        "created_cards": [real_id, "t_phantomdeadbeef"],
+    }))
+    assert rejected.get("error")
+    assert "t_phantomdeadbeef" in rejected["error"]
+
+    # Retry with corrected list.
+    ok = json.loads(kt._handle_complete({
+        "summary": "retry with corrected list",
+        "created_cards": [real_id],
+    }))
+    assert ok.get("ok") is True
+
+    conn = kb.connect()
+    try:
+        assert kb.get_task(conn, worker_env).status == "done"
+    finally:
+        conn.close()
+
+
 def test_block_happy_path(worker_env):
     from tools import kanban_tools as kt
     out = kt._handle_block({"reason": "need clarification"})
@@ -368,6 +619,52 @@ def test_create_rejects_non_list_parents(worker_env):
     assert json.loads(out).get("error")
 
 
+def test_create_parses_triage_string_false(worker_env):
+    from tools import kanban_tools as kt
+    from hermes_cli import kanban_db as kb
+    out = kt._handle_create({
+        "title": "not triage",
+        "assignee": "peer",
+        "triage": "false",
+    })
+    d = json.loads(out)
+    assert d["ok"] is True
+    conn = kb.connect()
+    try:
+        task = kb.get_task(conn, d["task_id"])
+        assert task.status == "ready"
+    finally:
+        conn.close()
+
+
+def test_create_parses_triage_string_true(worker_env):
+    from tools import kanban_tools as kt
+    from hermes_cli import kanban_db as kb
+    out = kt._handle_create({
+        "title": "needs triage",
+        "assignee": "peer",
+        "triage": "true",
+    })
+    d = json.loads(out)
+    assert d["ok"] is True
+    conn = kb.connect()
+    try:
+        task = kb.get_task(conn, d["task_id"])
+        assert task.status == "triage"
+    finally:
+        conn.close()
+
+
+def test_create_rejects_bad_triage(worker_env):
+    from tools import kanban_tools as kt
+    out = kt._handle_create({
+        "title": "bad triage",
+        "assignee": "peer",
+        "triage": "sometimes",
+    })
+    assert "triage must be" in json.loads(out).get("error", "")
+
+
 def test_create_accepts_string_parent(worker_env):
     """Convenience: a single parent id as string is coerced to [id]."""
     from tools import kanban_tools as kt
@@ -458,9 +755,35 @@ def test_link_rejects_cycle(worker_env):
     assert json.loads(out).get("error")
 
 
-# ---------------------------------------------------------------------------
-# End-to-end: simulate a full worker lifecycle through the tools
-# ---------------------------------------------------------------------------
+def test_unblock_happy_path(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="blocked", assignee="worker")
+        kb.block_task(conn, tid, reason="waiting")
+    finally:
+        conn.close()
+
+    from tools import kanban_tools as kt
+    out = kt._handle_unblock({"task_id": tid})
+    d = json.loads(out)
+    assert d["ok"] is True
+    assert d["status"] == "ready"
+
+    conn = kb.connect()
+    try:
+        assert kb.get_task(conn, tid).status == "ready"
+    finally:
+        conn.close()
+
+
+def test_unblock_rejects_non_blocked_task(monkeypatch, worker_env):
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    from tools import kanban_tools as kt
+    out = kt._handle_unblock({"task_id": worker_env})
+    assert json.loads(out).get("error")
+
 
 def test_worker_lifecycle_through_tools(worker_env):
     """Drive the full claim -> heartbeat -> comment -> complete lifecycle
@@ -599,11 +922,12 @@ def test_kanban_guidance_prompt_size_bounded(monkeypatch, tmp_path):
 # ---------------------------------------------------------------------------
 #
 # A worker process has HERMES_KANBAN_TASK set to its own task id. The
-# destructive tools (kanban_complete, kanban_block, kanban_heartbeat)
-# must refuse to operate on any OTHER task id, even if the caller
-# supplies an explicit `task_id` argument. Workers legitimately call
-# kanban_show / kanban_comment / kanban_create / kanban_link on other
-# tasks, so those are unrestricted.
+# destructive tools (kanban_complete, kanban_block, kanban_heartbeat,
+# kanban_unblock) must refuse to operate
+# on any OTHER task id, even if the caller supplies an explicit `task_id`
+# argument. Workers legitimately call kanban_show / kanban_list /
+# kanban_comment / kanban_create / kanban_link on other tasks, so those
+# are unrestricted.
 #
 # Orchestrator profiles (no HERMES_KANBAN_TASK in env) are intentionally
 # exempt — their job is routing, and they sometimes close out child
@@ -712,6 +1036,37 @@ def test_worker_can_comment_on_foreign_task(worker_env):
         conn.close()
 
 
+def test_worker_unblock_rejects_foreign_task_id(worker_env):
+    """A worker cannot unblock any task — kanban_unblock is orchestrator-only.
+
+    The check fires before the per-task ownership check, so the error
+    surface is the orchestrator-only refusal rather than the
+    cross-task-ownership refusal. Either is fine — the property we're
+    pinning is "worker cannot mutate foreign task via kanban_unblock".
+    """
+    from hermes_cli import kanban_db as kb
+    conn = kb.connect()
+    try:
+        other = kb.create_task(conn, title="blocked sibling", assignee="peer")
+        kb.block_task(conn, other, reason="waiting")
+    finally:
+        conn.close()
+
+    from tools import kanban_tools as kt
+    out = kt._handle_unblock({"task_id": other})
+    d = json.loads(out)
+    err = d.get("error", "")
+    assert "orchestrator-only" in err or "refusing to mutate" in err, (
+        f"expected worker-rejection error, got {err}"
+    )
+
+    conn = kb.connect()
+    try:
+        assert kb.get_task(conn, other).status == "blocked"
+    finally:
+        conn.close()
+
+
 def test_worker_complete_own_task_still_works(worker_env):
     """The ownership check doesn't break the normal own-task happy path."""
     from tools import kanban_tools as kt

tests/tools/test_send_message_tool.py

@@ -2229,3 +2229,106 @@ class TestSendViaAdapterStandaloneFallback:
         assert result["success"] is True
         assert result["message_id"] == "abc-123"
         assert result["extra_field"] == "preserved"
+
+
+# ---------------------------------------------------------------------------
+# _check_send_message — availability gating
+# ---------------------------------------------------------------------------
+
+class TestCheckSendMessage:
+    """The tool's check_fn governs whether the model sees ``send_message`` as
+    callable for a given session. The four passing conditions are:
+
+    1. ``HERMES_KANBAN_TASK`` is set (worker spawned by the kanban dispatcher
+       — parent gateway is by definition running, but the worker's
+       ``HERMES_HOME`` may be a profile dir without a ``gateway.pid``).
+    2. ``HERMES_SESSION_PLATFORM`` resolves to a non-empty, non-``local`` value
+       (the session is wired to a messaging platform like Telegram).
+    3. ``is_gateway_running()`` returns True (CLI / orchestrator profile with
+       a live gateway colocated under the same ``HERMES_HOME``).
+    4. None of the above → False, tool is hidden.
+    """
+
+    def test_kanban_task_env_grants_access(self, monkeypatch):
+        """Workers spawned by the dispatcher (HERMES_KANBAN_TASK set) must be
+        allowed regardless of session_platform / gateway-pid state."""
+        from tools.send_message_tool import _check_send_message
+
+        monkeypatch.setenv("HERMES_KANBAN_TASK", "t_abc12345")
+        monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False)
+
+        with patch("gateway.session_context.get_session_env", return_value=""), \
+             patch("gateway.status.is_gateway_running", return_value=False):
+            assert _check_send_message() is True
+
+    def test_kanban_task_env_short_circuits_before_gateway_check(self, monkeypatch):
+        """Honoring HERMES_KANBAN_TASK must not depend on importing or calling
+        gateway.status — the worker may run with a HERMES_HOME that has no
+        gateway.pid, and we don't want that import path to be load-bearing."""
+        from tools.send_message_tool import _check_send_message
+
+        monkeypatch.setenv("HERMES_KANBAN_TASK", "t_abc12345")
+
+        with patch("gateway.session_context.get_session_env",
+                   side_effect=AssertionError("session_context not consulted "
+                                              "when HERMES_KANBAN_TASK is set")), \
+             patch("gateway.status.is_gateway_running",
+                   side_effect=AssertionError("gateway.status not consulted "
+                                              "when HERMES_KANBAN_TASK is set")):
+            assert _check_send_message() is True
+
+    def test_messaging_platform_session_grants_access(self, monkeypatch):
+        """Telegram/Discord/etc. sessions pass via the platform branch even
+        without HERMES_KANBAN_TASK."""
+        from tools.send_message_tool import _check_send_message
+
+        monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+        with patch("gateway.session_context.get_session_env", return_value="telegram"), \
+             patch("gateway.status.is_gateway_running", return_value=False):
+            assert _check_send_message() is True
+
+    def test_local_platform_falls_through_to_gateway_check(self, monkeypatch):
+        """``HERMES_SESSION_PLATFORM=local`` means CLI-style — must defer to
+        is_gateway_running() rather than auto-grant."""
+        from tools.send_message_tool import _check_send_message
+
+        monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+        with patch("gateway.session_context.get_session_env", return_value="local"), \
+             patch("gateway.status.is_gateway_running", return_value=True) as gw_mock:
+            assert _check_send_message() is True
+            gw_mock.assert_called_once()
+
+    def test_running_gateway_grants_access(self, monkeypatch):
+        """Plain CLI session (no kanban task, empty platform) with a live
+        gateway: tool is callable."""
+        from tools.send_message_tool import _check_send_message
+
+        monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+        with patch("gateway.session_context.get_session_env", return_value=""), \
+             patch("gateway.status.is_gateway_running", return_value=True):
+            assert _check_send_message() is True
+
+    def test_no_signals_means_unavailable(self, monkeypatch):
+        """No kanban task, no platform, no gateway: tool is hidden."""
+        from tools.send_message_tool import _check_send_message
+
+        monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+        with patch("gateway.session_context.get_session_env", return_value=""), \
+             patch("gateway.status.is_gateway_running", return_value=False):
+            assert _check_send_message() is False
+
+    def test_gateway_status_import_error_is_swallowed(self, monkeypatch):
+        """If gateway.status can't be imported (unusual deployment / partial
+        install), the check returns False rather than raising."""
+        from tools.send_message_tool import _check_send_message
+
+        monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+
+        with patch("gateway.session_context.get_session_env", return_value=""), \
+             patch("gateway.status.is_gateway_running",
+                   side_effect=ImportError("simulated")):
+            assert _check_send_message() is False