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

tests/acp/test_server.py

@@ -13,6 +13,7 @@ from acp.schema import (
     AgentCapabilities,
     AgentMessageChunk,
     AgentPlanUpdate,
+    AgentThoughtChunk,
     AuthenticateResponse,
     AvailableCommandsUpdate,
     Implementation,
@@ -467,25 +468,296 @@ class TestSessionOps:
         )
 
     @pytest.mark.asyncio
-    async def test_load_session_schedules_history_replay_after_response(self, agent):
-        """Zed only attaches replayed updates after session/load has completed."""
+    async def test_load_session_replays_reasoning_thought_before_message(self, agent):
+        """Thinking-model thoughts must be replayed via ``agent_thought_chunk``.
+
+        Regression for #12285 — when a session is loaded, persisted assistant
+        ``reasoning_content`` / ``reasoning`` fields must surface as ACP
+        ``AgentThoughtChunk`` notifications in the same relative position they
+        had live (thought streams before the assistant message text), so Zed's
+        collapsed Thinking pane rebuilds instead of vanishing on reconnect.
+        """
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
+        agent._conn = mock_conn
+
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [
+            {"role": "user", "content": "Walk me through it."},
+            {
+                "role": "assistant",
+                "reasoning_content": "Let me think step by step about the request.",
+                "content": "Here is the plan.",
+            },
+            {"role": "user", "content": "And the legacy case?"},
+            {
+                "role": "assistant",
+                # No reasoning_content — exercise the legacy "reasoning" fallback
+                # path so sessions persisted before #16892 still replay thoughts.
+                "reasoning": "Older sessions stored the trace under the internal key.",
+                "content": "Same idea, older field name.",
+            },
+        ]
+
+        mock_conn.session_update.reset_mock()
+        resp = await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+        assert isinstance(resp, LoadSessionResponse)
+
+        replay_kinds = [
+            getattr(call.kwargs.get("update"), "session_update", None)
+            for call in mock_conn.session_update.await_args_list
+            if getattr(call.kwargs.get("update"), "session_update", None)
+            in {"user_message_chunk", "agent_message_chunk", "agent_thought_chunk"}
+        ]
+        assert replay_kinds == [
+            "user_message_chunk",
+            "agent_thought_chunk",
+            "agent_message_chunk",
+            "user_message_chunk",
+            "agent_thought_chunk",
+            "agent_message_chunk",
+        ]
+
+        thought_updates = [
+            call.kwargs["update"]
+            for call in mock_conn.session_update.await_args_list
+            if isinstance(call.kwargs.get("update"), AgentThoughtChunk)
+        ]
+        assert len(thought_updates) == 2
+        assert thought_updates[0].content.text == "Let me think step by step about the request."
+        assert thought_updates[1].content.text == "Older sessions stored the trace under the internal key."
+
+    @pytest.mark.asyncio
+    async def test_load_session_replays_reasoning_only_turn(self, agent):
+        """Assistant turns with reasoning but no content should still emit a thought.
+
+        Pure reasoning-only assistant entries (e.g. a thinking step before a
+        tool-call turn) commonly carry ``reasoning_content`` with empty
+        ``content``. The replay must still surface the thought so the editor's
+        Thinking pane rebuilds, even when there is no message text to follow.
+        """
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
+        agent._conn = mock_conn
+
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [
+            {
+                "role": "assistant",
+                "reasoning_content": "I should call the search tool next.",
+                "content": "",
+            },
+        ]
+
+        mock_conn.session_update.reset_mock()
+        await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+        thought_updates = [
+            call.kwargs["update"]
+            for call in mock_conn.session_update.await_args_list
+            if isinstance(call.kwargs.get("update"), AgentThoughtChunk)
+        ]
+        message_updates = [
+            call.kwargs["update"]
+            for call in mock_conn.session_update.await_args_list
+            if isinstance(call.kwargs.get("update"), AgentMessageChunk)
+        ]
+        assert len(thought_updates) == 1
+        assert thought_updates[0].content.text == "I should call the search tool next."
+        assert message_updates == []
+
+    @pytest.mark.asyncio
+    async def test_load_session_skips_empty_reasoning_fields(self, agent):
+        """Empty/whitespace reasoning fields must not produce notifications."""
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
+        agent._conn = mock_conn
+
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [
+            {
+                "role": "assistant",
+                "reasoning_content": "",
+                "reasoning": "   \n\t",
+                "content": "Just a regular answer.",
+            },
+        ]
+
+        mock_conn.session_update.reset_mock()
+        await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+        thought_updates = [
+            call.kwargs["update"]
+            for call in mock_conn.session_update.await_args_list
+            if isinstance(call.kwargs.get("update"), AgentThoughtChunk)
+        ]
+        assert thought_updates == []
+
+    @pytest.mark.asyncio
+    async def test_load_session_replays_thought_then_tool_call_without_message(self, agent):
+        """Canonical thinking-model shape: reasoning + tool_call + no body text.
+
+        Thinking models commonly emit a pre-tool thought followed by a
+        tool_calls turn with empty ``content``. Replay must emit:
+        ``agent_thought_chunk`` then ``tool_call`` then ``tool_call_update``
+        for the matching tool result — and crucially, NO ``agent_message_chunk``
+        for the empty-text assistant body. Regression for the canonical
+        thinking-then-tool flow on #12285.
+        """
+        mock_conn = MagicMock(spec=acp.Client)
+        mock_conn.session_update = AsyncMock()
+        agent._conn = mock_conn
+
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [
+            {"role": "user", "content": "Find the bug."},
+            {
+                "role": "assistant",
+                "reasoning_content": "I should grep for the function name first.",
+                "content": "",
+                "tool_calls": [
+                    {
+                        "id": "call_grep_1",
+                        "type": "function",
+                        "function": {
+                            "name": "search_files",
+                            "arguments": '{"pattern":"foo","path":"."}',
+                        },
+                    }
+                ],
+            },
+            {
+                "role": "tool",
+                "tool_call_id": "call_grep_1",
+                "content": '{"total_count":1,"matches":[{"path":"x.py","line":1,"content":"foo"}]}',
+            },
+        ]
+
+        mock_conn.session_update.reset_mock()
+        await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+        kinds = [
+            getattr(call.kwargs.get("update"), "session_update", None)
+            for call in mock_conn.session_update.await_args_list
+            if getattr(call.kwargs.get("update"), "session_update", None)
+            in {
+                "user_message_chunk",
+                "agent_thought_chunk",
+                "agent_message_chunk",
+                "tool_call",
+                "tool_call_update",
+            }
+        ]
+        # No agent_message_chunk for the empty-content assistant turn.
+        assert "agent_message_chunk" not in kinds
+        # Thought must precede the tool_call_start within the assistant turn,
+        # and the tool result follows.
+        assert kinds == [
+            "user_message_chunk",
+            "agent_thought_chunk",
+            "tool_call",
+            "tool_call_update",
+        ]
+
+    @pytest.mark.asyncio
+    async def test_load_session_replays_history_before_returning_response(self, agent):
+        """Per ACP spec, replay must complete BEFORE load_session returns.
+
+        Spec-compliant ACP clients (Codex, Claude Code, OpenCode, Pi, Zed)
+        attach their ``session/update`` listeners before awaiting the
+        ``loadSession`` RPC and rely on receiving the full transcript within
+        the request's lifetime. Deferring replay via ``loop.call_soon`` (the
+        prior behavior in May 2026) broke clients that read notification
+        counts synchronously against the load response — see #12285 follow-up.
+        """
         new_resp = await agent.new_session(cwd="/tmp")
         state = agent.session_manager.get_session(new_resp.session_id)
         state.history = [{"role": "user", "content": "hello from history"}]
-        events = []
+        events: list[str] = []
 
-        async def replay_after_response(_state):
+        async def replay_records(_state):
             events.append("replay")
 
-        with patch.object(agent, "_replay_session_history", side_effect=replay_after_response):
+        with patch.object(agent, "_replay_session_history", side_effect=replay_records):
             resp = await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
             events.append("returned")
 
         assert isinstance(resp, LoadSessionResponse)
-        assert events == ["returned"]
-        await asyncio.sleep(0)
-        await asyncio.sleep(0)
-        assert events == ["returned", "replay"]
+        # Replay must have happened BEFORE the response was constructed —
+        # i.e. before the `events.append("returned")` after the await resolves.
+        assert events == ["replay", "returned"]
+
+    @pytest.mark.asyncio
+    async def test_resume_session_replays_history_before_returning_response(self, agent):
+        """Same spec rationale as ``load_session`` — replay before responding."""
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [{"role": "user", "content": "hello from history"}]
+        events: list[str] = []
+
+        async def replay_records(_state):
+            events.append("replay")
+
+        with patch.object(agent, "_replay_session_history", side_effect=replay_records):
+            resp = await agent.resume_session(cwd="/tmp", session_id=new_resp.session_id)
+            events.append("returned")
+
+        assert isinstance(resp, ResumeSessionResponse)
+        assert events == ["replay", "returned"]
+
+    @pytest.mark.asyncio
+    async def test_load_session_survives_replay_helper_exception(self, agent, caplog):
+        """A replay helper raising must not turn load_session into an error.
+
+        With awaited replay, an exception in ``_replay_session_history`` now
+        propagates into the ``load_session`` handler. The defensive try/except
+        guard at the call site must catch and log it so the JSON-RPC client
+        still receives a ``LoadSessionResponse`` — partial transcripts are
+        acceptable, total load failure is not.
+        """
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [{"role": "user", "content": "hi"}]
+
+        async def boom(_state):
+            raise RuntimeError("simulated replay helper crash")
+
+        with caplog.at_level("WARNING", logger="acp_adapter.server"):
+            with patch.object(agent, "_replay_session_history", side_effect=boom):
+                resp = await agent.load_session(cwd="/tmp", session_id=new_resp.session_id)
+
+        assert isinstance(resp, LoadSessionResponse)
+        assert "history replay raised during session/load" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_resume_session_survives_replay_helper_exception(self, agent, caplog):
+        """Same guarantee as ``load_session`` for the resume path."""
+        new_resp = await agent.new_session(cwd="/tmp")
+        state = agent.session_manager.get_session(new_resp.session_id)
+        state.history = [{"role": "user", "content": "hi"}]
+
+        async def boom(_state):
+            raise RuntimeError("simulated replay helper crash")
+
+        with caplog.at_level("WARNING", logger="acp_adapter.server"):
+            with patch.object(agent, "_replay_session_history", side_effect=boom):
+                resp = await agent.resume_session(cwd="/tmp", session_id=new_resp.session_id)
+
+        assert isinstance(resp, ResumeSessionResponse)
+        assert "history replay raised during session/resume" in caplog.text
 
     @pytest.mark.asyncio
     async def test_resume_session_creates_new_if_missing(self, agent):

tests/agent/test_anthropic_oauth_pkce.py

@@ -0,0 +1,170 @@
+"""Regression tests for the Anthropic OAuth PKCE flow.
+
+Guards against re-introducing the bug where the PKCE ``code_verifier`` was
+reused as the OAuth ``state`` parameter, leaking the verifier via the
+authorization URL (browser history, Referer headers, auth-server logs) and
+removing CSRF protection on the callback path.
+
+History:
+  - PR #1775 first fixed this on ``run_hermes_oauth_login()``.
+  - PR #2647 (b17e5c10) added ``run_hermes_oauth_login_pure()`` and silently
+    copy-pasted the pre-#1775 vulnerable pattern.
+  - PR #3107 removed the old function, leaving only the regressed copy.
+  - PR #10699 (issue #10693) fixed the regression on the surviving function.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+from typing import Any, Dict
+from urllib.parse import parse_qs, urlparse
+
+
+def _patch_oauth_flow(
+    monkeypatch,
+    *,
+    callback_code: str,
+    token_response: Dict[str, Any] | None = None,
+    capture_token_request: Dict[str, Any] | None = None,
+    capture_auth_url: Dict[str, str] | None = None,
+) -> None:
+    """Wire up monkeypatches that let ``run_hermes_oauth_login_pure()`` run
+    end-to-end without touching a real browser, stdin, or HTTP endpoint.
+
+    ``callback_code`` is the literal string the user would paste back into the
+    terminal (``"<code>#<state>"`` format).
+    ``capture_token_request`` and ``capture_auth_url`` are out-dict captures
+    so the test can introspect what was sent to the auth URL and the token
+    endpoint, respectively.
+    """
+    import urllib.request
+
+    if token_response is None:
+        token_response = {
+            "access_token": "sk-ant-test-access",
+            "refresh_token": "sk-ant-test-refresh",
+            "expires_in": 3600,
+        }
+
+    def fake_open(url):
+        if capture_auth_url is not None:
+            capture_auth_url["url"] = url
+        return True
+
+    monkeypatch.setattr("webbrowser.open", fake_open)
+    monkeypatch.setattr("builtins.input", lambda *_a, **_kw: callback_code)
+
+    class _FakeResponse:
+        def __init__(self, body: bytes) -> None:
+            self._body = body
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_exc):
+            return False
+
+        def read(self):
+            return self._body
+
+    def fake_urlopen(req, *_a, **_kw):
+        if capture_token_request is not None:
+            capture_token_request["url"] = req.full_url
+            capture_token_request["data"] = json.loads(req.data.decode())
+            capture_token_request["headers"] = dict(req.headers)
+        return _FakeResponse(json.dumps(token_response).encode())
+
+    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
+
+
+def test_authorization_url_state_is_not_pkce_verifier(monkeypatch, tmp_path):
+    """The ``state`` parameter in the authorization URL must NOT equal the
+    PKCE ``code_verifier``.
+
+    Reusing the verifier as state leaks the verifier into browser history,
+    Referer headers, and auth-server access logs — defeating RFC 7636.
+    """
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+    captured_url: Dict[str, str] = {}
+    captured_token: Dict[str, Any] = {}
+    _patch_oauth_flow(
+        monkeypatch,
+        # state echoed back unchanged so the CSRF guard passes
+        callback_code="auth-code-from-anthropic#PLACEHOLDER",
+        capture_auth_url=captured_url,
+        capture_token_request=captured_token,
+    )
+
+    # Stub the callback parse: we need the state echoed back to match. To do
+    # that without hardcoding the state value, override input() AFTER seeing
+    # the auth URL.
+    import builtins
+
+    real_input_calls = {"count": 0}
+
+    def fake_input(*_a, **_kw):
+        real_input_calls["count"] += 1
+        # First (and only) call is the "Authorization code:" prompt.
+        url = captured_url.get("url", "")
+        qs = parse_qs(urlparse(url).query)
+        state = qs.get("state", [""])[0]
+        return f"auth-code-from-anthropic#{state}"
+
+    monkeypatch.setattr(builtins, "input", fake_input)
+
+    from agent.anthropic_adapter import run_hermes_oauth_login_pure
+
+    result = run_hermes_oauth_login_pure()
+    assert result is not None, "OAuth flow should succeed with matching state"
+
+    url = captured_url["url"]
+    qs = parse_qs(urlparse(url).query)
+
+    assert "state" in qs and qs["state"][0], "authorization URL must include state"
+    assert "code_challenge" in qs, "authorization URL must include code_challenge"
+
+    state_in_url = qs["state"][0]
+    verifier_sent = captured_token["data"]["code_verifier"]
+
+    # The whole point: state and verifier must be independent values.
+    assert state_in_url != verifier_sent, (
+        "PKCE code_verifier was reused as OAuth state — regression of #10693 / "
+        "#1775. The verifier is supposed to be a secret known only to the "
+        "client; placing it in the authorization URL leaks it via browser "
+        "history, Referer headers, and auth-server logs."
+    )
+
+    # And the verifier MUST NOT appear anywhere in the URL.
+    assert verifier_sent not in url, (
+        "PKCE verifier leaked into authorization URL — regression of #10693"
+    )
+
+
+def test_callback_state_mismatch_aborts(monkeypatch, tmp_path, caplog):
+    """If the state returned in the callback does not match the one we sent
+    in the authorization URL, the flow must abort before exchanging the code.
+
+    Without this check, an attacker who tricks the user into pasting a
+    crafted ``<code>#<state>`` string can complete the token exchange — the
+    CSRF protection that ``state`` is supposed to provide (RFC 6749 §10.12)
+    would be absent.
+    """
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+
+    captured_token: Dict[str, Any] = {}
+    _patch_oauth_flow(
+        monkeypatch,
+        callback_code="attacker-code#attacker-state-does-not-match",
+        capture_token_request=captured_token,
+    )
+
+    from agent.anthropic_adapter import run_hermes_oauth_login_pure
+
+    result = run_hermes_oauth_login_pure()
+
+    assert result is None, "mismatched state must abort the flow"
+    assert "url" not in captured_token, (
+        "token exchange must NOT happen when state mismatches"
+    )

tests/agent/test_copilot_acp_deprecation.py

@@ -0,0 +1,77 @@
+"""Tests for gh-copilot CLI deprecation detection and GitHub Models Azure URL mapping."""
+
+import pytest
+
+from agent.copilot_acp_client import _is_gh_copilot_deprecation_message
+
+
+class TestDeprecationPatternDetection:
+    """Verify that stderr from the deprecated `gh copilot` extension is caught
+    without false-positiving on the new `@github/copilot` CLI."""
+
+    _REAL_DEPRECATION_STDERR = (
+        "The gh-copilot extension has been deprecated in favor of the newer "
+        "GitHub Copilot CLI.\nFor more information, visit:\n"
+        "- Copilot CLI: https://github.com/github/copilot-cli\n"
+        "- Deprecation announcement: https://github.blog/changelog/"
+        "2025-09-25-upcoming-deprecation-of-gh-copilot-cli-extension\n"
+        "No commands will be executed."
+    )
+
+    def test_real_deprecation_message_matches(self):
+        assert _is_gh_copilot_deprecation_message(self._REAL_DEPRECATION_STDERR)
+
+    @pytest.mark.parametrize(
+        "stderr_text",
+        [
+            # The deprecation banner uses both halves of the fingerprint.
+            "The gh-copilot extension has been deprecated.",
+            "gh-copilot: no commands will be executed.",
+            # Mixed casing — match is case-insensitive.
+            "The GH-Copilot Extension HAS BEEN DEPRECATED.",
+        ],
+    )
+    def test_genuine_deprecation_variants_match(self, stderr_text: str):
+        assert _is_gh_copilot_deprecation_message(stderr_text)
+
+    @pytest.mark.parametrize(
+        "stderr_text",
+        [
+            # Generic errors — no fingerprint at all.
+            "Error: connection refused",
+            "",
+            # The NEW @github/copilot CLI's repo is github.com/github/copilot-cli.
+            # Its stderr can legitimately mention "copilot-cli" or "deprecation"
+            # in unrelated contexts; neither alone should trip the detector.
+            "copilot-cli: failed to authenticate with the API",
+            "warning: the --foo flag is scheduled for deprecation in v3",
+            "See https://github.com/github/copilot-cli/issues for support",
+            # Half the fingerprint without the other half.
+            "gh-copilot: command not found",
+            "extension has been deprecated (some other extension)",
+        ],
+    )
+    def test_does_not_false_positive(self, stderr_text: str):
+        assert not _is_gh_copilot_deprecation_message(stderr_text)
+
+
+class TestGitHubModelsAzureUrl:
+    """Verify that the Azure GitHub Models URL is recognised."""
+
+    def test_url_to_provider_contains_azure_models(self):
+        from agent.model_metadata import _URL_TO_PROVIDER
+
+        # Maps to the canonical "copilot" provider (same convention as the
+        # other GitHub-family entries) — not the "github-models" alias.
+        assert _URL_TO_PROVIDER.get("models.inference.ai.azure.com") == "copilot"
+
+    def test_is_github_models_base_url_recognises_azure(self):
+        from hermes_cli.models import _is_github_models_base_url
+
+        assert _is_github_models_base_url("https://models.inference.ai.azure.com")
+        assert _is_github_models_base_url("https://models.inference.ai.azure.com/v1/chat")
+
+    def test_is_github_models_base_url_still_recognises_github_ai(self):
+        from hermes_cli.models import _is_github_models_base_url
+
+        assert _is_github_models_base_url("https://models.github.ai/inference")

tests/gateway/test_active_session_text_merge.py

@@ -0,0 +1,152 @@
+"""Regression test for #4469.
+
+When the agent is actively running (session present in
+``adapter._active_sessions``) and the user fires off multiple TEXT
+follow-ups in rapid succession, the previous behaviour was a single-slot
+replacement at ``gateway/platforms/base.py``:
+
+    self._pending_messages[session_key] = event
+
+So three rapid messages ``A``, ``B``, ``C`` arriving while the agent was
+still working on the initial turn produced a pending slot containing only
+``C``; ``A`` and ``B`` were silently dropped.
+
+The fix routes the follow-up through ``merge_pending_message_event(...,
+merge_text=True)`` so TEXT events accumulate into the existing pending
+event's text instead of clobbering it.  Photo / media bursts continue to
+merge through the same helper (they always did).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+import types
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+# Minimal telegram stub so importing gateway.platforms.base does not pull
+# in the real python-telegram-bot dependency.
+_tg = sys.modules.get("telegram") or types.ModuleType("telegram")
+_tg.constants = sys.modules.get("telegram.constants") or types.ModuleType("telegram.constants")
+_ct = MagicMock()
+_ct.PRIVATE = "private"
+_ct.GROUP = "group"
+_ct.SUPERGROUP = "supergroup"
+_tg.constants.ChatType = _ct
+sys.modules.setdefault("telegram", _tg)
+sys.modules.setdefault("telegram.constants", _tg.constants)
+sys.modules.setdefault("telegram.ext", types.ModuleType("telegram.ext"))
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+)
+from gateway.session import SessionSource, build_session_key
+
+
+def _make_event(text: str, chat_id: str = "12345") -> MessageEvent:
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id=chat_id,
+        chat_type="dm",
+        user_id="u1",
+    )
+    return MessageEvent(
+        text=text,
+        message_type=MessageType.TEXT,
+        source=source,
+        message_id=f"msg-{text[:8]}",
+    )
+
+
+def _make_adapter() -> BasePlatformAdapter:
+    """Build a BasePlatformAdapter without running its heavy __init__.
+
+    We only need the bits ``handle_message`` touches on the active-session
+    path: ``_active_sessions``, ``_pending_messages``,
+    ``_message_handler``, ``_busy_session_handler``, ``config``, ``platform``.
+    """
+
+    class _DummyAdapter(BasePlatformAdapter):  # type: ignore[misc]
+        async def connect(self):
+            pass
+
+        async def disconnect(self):
+            pass
+
+        async def get_chat_info(self, chat_id):
+            return None
+
+        async def send(self, *args, **kwargs):
+            return MagicMock(success=True, message_id="x", retryable=False)
+
+    adapter = object.__new__(_DummyAdapter)
+    adapter.config = PlatformConfig(enabled=True, token="***")
+    adapter.platform = Platform.TELEGRAM
+    adapter._message_handler = AsyncMock(return_value=None)
+    adapter._busy_session_handler = None
+    adapter._active_sessions = {}
+    adapter._pending_messages = {}
+    adapter._session_tasks = {}
+    adapter._background_tasks = set()
+    adapter._post_delivery_callbacks = {}
+    adapter._expected_cancelled_tasks = set()
+    adapter._fatal_error_code = None
+    adapter._fatal_error_message = None
+    adapter._fatal_error_retryable = True
+    adapter._fatal_error_handler = None
+    adapter._running = True
+    adapter._auto_tts_default = False
+    adapter._auto_tts_enabled_chats = set()
+    adapter._auto_tts_disabled_chats = set()
+    adapter._typing_paused = set()
+    return adapter
+
+
+@pytest.mark.asyncio
+async def test_rapid_text_followups_accumulate_instead_of_replacing():
+    """Three rapid TEXT follow-ups during an active session must all
+    survive in ``adapter._pending_messages[session_key].text``."""
+    adapter = _make_adapter()
+    first = _make_event("part one")
+    session_key = build_session_key(first.source)
+
+    # Mark the session as active so subsequent messages take the
+    # "already running" branch in handle_message.
+    adapter._active_sessions[session_key] = asyncio.Event()
+
+    second = _make_event("part two")
+    third = _make_event("part three")
+
+    await adapter.handle_message(second)
+    await adapter.handle_message(third)
+
+    # Both rapid follow-ups must be preserved, not just the last one.
+    pending = adapter._pending_messages[session_key]
+    assert pending.text == "part two\npart three", (
+        f"expected accumulated text, got {pending.text!r}"
+    )
+    # Interrupt event must be signalled exactly like before.
+    assert adapter._active_sessions[session_key].is_set()
+
+
+@pytest.mark.asyncio
+async def test_single_followup_is_stored_as_is():
+    """One TEXT follow-up still lands as the event object itself
+    (no spurious wrapping / mutation) — guards against the merge path
+    breaking the simple case."""
+    adapter = _make_adapter()
+    first = _make_event("only one")
+    session_key = build_session_key(first.source)
+
+    adapter._active_sessions[session_key] = asyncio.Event()
+    await adapter.handle_message(first)
+
+    pending = adapter._pending_messages[session_key]
+    assert pending is first
+    assert pending.text == "only one"
+    assert adapter._active_sessions[session_key].is_set()

tests/hermes_cli/test_doctor.py

@@ -839,3 +839,108 @@ class TestGitHubTokenCheck:
 
         assert "gh auth" in str(call_log) or any(c[0] == "gh" for c in call_log), f"gh not called: {call_log}"
         assert "GitHub authenticated via gh CLI" in out or "token configured" in out
+
+
+def _run_doctor_with_healthy_oauth_fallback(
+    monkeypatch,
+    tmp_path,
+    *,
+    env_key: str,
+    bad_key: str,
+    failing_host: str,
+    gemini_oauth_status: dict,
+    minimax_oauth_status: dict,
+) -> str:
+    home = tmp_path / ".hermes"
+    home.mkdir(parents=True, exist_ok=True)
+    (home / "config.yaml").write_text(
+        "model:\n"
+        "  provider: nous\n"
+        "  default: moonshotai/kimi-k2.6\n",
+        encoding="utf-8",
+    )
+    project = tmp_path / "project"
+    project.mkdir(exist_ok=True)
+
+    monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
+    monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", project)
+    monkeypatch.setattr(doctor_mod, "_DHH", str(home))
+    monkeypatch.setenv(env_key, bad_key)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
+    monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+    monkeypatch.delenv("GOOGLE_API_KEY", raising=False)
+    monkeypatch.delenv("MINIMAX_API_KEY", raising=False)
+    monkeypatch.delenv("MINIMAX_CN_API_KEY", raising=False)
+    monkeypatch.setenv(env_key, bad_key)
+
+    fake_model_tools = types.SimpleNamespace(
+        check_tool_availability=lambda *a, **kw: ([], []),
+        TOOLSET_REQUIREMENTS={},
+    )
+    monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
+
+    from hermes_cli import auth as _auth_mod
+
+    monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {"logged_in": True})
+    monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
+    monkeypatch.setattr(_auth_mod, "get_gemini_oauth_auth_status", lambda: gemini_oauth_status)
+    monkeypatch.setattr(_auth_mod, "get_minimax_oauth_auth_status", lambda: minimax_oauth_status)
+
+    def fake_get(url, headers=None, timeout=None):
+        status = 401 if failing_host in url else 200
+        return types.SimpleNamespace(status_code=status)
+
+    import httpx
+
+    monkeypatch.setattr(httpx, "get", fake_get)
+
+    buf = io.StringIO()
+    with contextlib.redirect_stdout(buf):
+        doctor_mod.run_doctor(Namespace(fix=False))
+    return buf.getvalue()
+
+
+@pytest.mark.parametrize(
+    ("env_key", "bad_key", "failing_host", "gemini_oauth_status", "minimax_oauth_status", "unexpected_issue"),
+    [
+        (
+            "GOOGLE_API_KEY",
+            "bad-gemini-key",
+            "googleapis.com",
+            {"logged_in": True, "email": "user@example.com"},
+            {},
+            "Check GOOGLE_API_KEY in .env",
+        ),
+        (
+            "MINIMAX_API_KEY",
+            "bad-minimax-key",
+            "minimax.io",
+            {},
+            {"logged_in": True, "region": "global"},
+            "Check MINIMAX_API_KEY in .env",
+        ),
+    ],
+)
+def test_run_doctor_ignores_invalid_direct_keys_when_oauth_fallback_is_healthy(
+    monkeypatch,
+    tmp_path,
+    env_key,
+    bad_key,
+    failing_host,
+    gemini_oauth_status,
+    minimax_oauth_status,
+    unexpected_issue,
+):
+    out = _run_doctor_with_healthy_oauth_fallback(
+        monkeypatch,
+        tmp_path,
+        env_key=env_key,
+        bad_key=bad_key,
+        failing_host=failing_host,
+        gemini_oauth_status=gemini_oauth_status,
+        minimax_oauth_status=minimax_oauth_status,
+    )
+
+    assert "invalid API key" in out
+    assert unexpected_issue not in out

tests/hermes_cli/test_plugins.py

@@ -662,6 +662,129 @@ class TestPluginContext:
         from tools.registry import registry
         assert "plugin_echo" in registry._tools
 
+    def test_register_tool_rejects_shadow_without_override(self, tmp_path, monkeypatch, caplog):
+        """Without override=True, registering a tool name claimed by a different toolset is rejected."""
+        from tools.registry import registry
+
+        # Seed an existing entry from a non-plugin toolset.
+        registry.register(
+            name="shadow_target",
+            toolset="terminal",
+            schema={"name": "shadow_target", "description": "Built-in", "parameters": {"type": "object", "properties": {}}},
+            handler=lambda args, **kw: "built-in",
+        )
+        original_handler = registry._tools["shadow_target"].handler
+        try:
+            plugins_dir = tmp_path / "hermes_test" / "plugins"
+            plugin_dir = plugins_dir / "shadow_plugin"
+            plugin_dir.mkdir(parents=True)
+            (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "shadow_plugin"}))
+            (plugin_dir / "__init__.py").write_text(
+                'def register(ctx):\n'
+                '    ctx.register_tool(\n'
+                '        name="shadow_target",\n'
+                '        toolset="plugin_shadow_plugin",\n'
+                '        schema={"name": "shadow_target", "description": "Plugin", "parameters": {"type": "object", "properties": {}}},\n'
+                '        handler=lambda args, **kw: "plugin",\n'
+                '    )\n'
+            )
+            hermes_home = tmp_path / "hermes_test"
+            (hermes_home / "config.yaml").write_text(
+                yaml.safe_dump({"plugins": {"enabled": ["shadow_plugin"]}})
+            )
+            monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+            with caplog.at_level(logging.ERROR, logger="tools.registry"):
+                mgr = PluginManager()
+                mgr.discover_and_load()
+
+            # Original handler must still be in place — registration was rejected.
+            assert registry._tools["shadow_target"].handler is original_handler
+            assert registry._tools["shadow_target"].toolset == "terminal"
+            # And an ERROR was logged explaining why and how to opt in.
+            assert any("override=True" in r.message for r in caplog.records)
+        finally:
+            registry.deregister("shadow_target")
+
+    def test_register_tool_override_replaces_existing(self, tmp_path, monkeypatch, caplog):
+        """override=True lets a plugin replace an existing built-in tool."""
+        from tools.registry import registry
+
+        registry.register(
+            name="override_target",
+            toolset="terminal",
+            schema={"name": "override_target", "description": "Built-in", "parameters": {"type": "object", "properties": {}}},
+            handler=lambda args, **kw: "built-in",
+        )
+        try:
+            plugins_dir = tmp_path / "hermes_test" / "plugins"
+            plugin_dir = plugins_dir / "override_plugin"
+            plugin_dir.mkdir(parents=True)
+            (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "override_plugin"}))
+            (plugin_dir / "__init__.py").write_text(
+                'def register(ctx):\n'
+                '    ctx.register_tool(\n'
+                '        name="override_target",\n'
+                '        toolset="plugin_override_plugin",\n'
+                '        schema={"name": "override_target", "description": "Plugin", "parameters": {"type": "object", "properties": {}}},\n'
+                '        handler=lambda args, **kw: "plugin",\n'
+                '        override=True,\n'
+                '    )\n'
+            )
+            hermes_home = tmp_path / "hermes_test"
+            (hermes_home / "config.yaml").write_text(
+                yaml.safe_dump({"plugins": {"enabled": ["override_plugin"]}})
+            )
+            monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+            with caplog.at_level(logging.INFO, logger="tools.registry"):
+                mgr = PluginManager()
+                mgr.discover_and_load()
+
+            # Plugin handler replaced the built-in one.
+            assert registry._tools["override_target"].toolset == "plugin_override_plugin"
+            assert registry._tools["override_target"].handler({}, ) == "plugin"
+            # Override is audit-logged at INFO.
+            assert any(
+                "overriding existing" in r.message and "override_target" in r.message
+                for r in caplog.records
+            )
+            # Plugin tracks it.
+            assert "override_target" in mgr._plugin_tool_names
+        finally:
+            registry.deregister("override_target")
+
+    def test_register_tool_override_on_new_name_is_noop_path(self, tmp_path, monkeypatch):
+        """override=True on a brand-new name still registers cleanly (no existing entry to replace)."""
+        from tools.registry import registry
+
+        plugins_dir = tmp_path / "hermes_test" / "plugins"
+        plugin_dir = plugins_dir / "new_override_plugin"
+        plugin_dir.mkdir(parents=True)
+        (plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "new_override_plugin"}))
+        (plugin_dir / "__init__.py").write_text(
+            'def register(ctx):\n'
+            '    ctx.register_tool(\n'
+            '        name="brand_new_override_tool",\n'
+            '        toolset="plugin_new_override_plugin",\n'
+            '        schema={"name": "brand_new_override_tool", "description": "New", "parameters": {"type": "object", "properties": {}}},\n'
+            '        handler=lambda args, **kw: "ok",\n'
+            '        override=True,\n'
+            '    )\n'
+        )
+        hermes_home = tmp_path / "hermes_test"
+        (hermes_home / "config.yaml").write_text(
+            yaml.safe_dump({"plugins": {"enabled": ["new_override_plugin"]}})
+        )
+        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+        try:
+            mgr = PluginManager()
+            mgr.discover_and_load()
+            assert "brand_new_override_tool" in registry._tools
+        finally:
+            registry.deregister("brand_new_override_tool")
+
 
 # ── TestPluginToolVisibility ───────────────────────────────────────────────
 

tests/run_agent/test_run_agent.py

@@ -2269,6 +2269,60 @@ class TestParallelScopePathNormalization:
         assert not _should_parallelize_tool_batch([tc1, tc2])
 
 
+class TestMcpParallelToolBatch:
+    """Integration test: _should_parallelize_tool_batch respects MCP parallel flag."""
+
+    def test_mcp_tools_default_sequential(self):
+        """MCP tools without supports_parallel_tool_calls are sequential."""
+        from run_agent import _should_parallelize_tool_batch
+        tc1 = _mock_tool_call(name="mcp_github_list_repos", arguments='{"org":"openai"}', call_id="c1")
+        tc2 = _mock_tool_call(name="mcp_github_search_code", arguments='{"q":"test"}', call_id="c2")
+        assert not _should_parallelize_tool_batch([tc1, tc2])
+
+    def test_mcp_tools_parallel_when_server_opted_in(self):
+        """MCP tools from a parallel-safe server can run concurrently."""
+        from run_agent import _should_parallelize_tool_batch
+        from tools.mcp_tool import _parallel_safe_servers, _lock
+        with _lock:
+            _parallel_safe_servers.add("github")
+        try:
+            tc1 = _mock_tool_call(name="mcp_github_list_repos", arguments='{"org":"openai"}', call_id="c1")
+            tc2 = _mock_tool_call(name="mcp_github_search_code", arguments='{"q":"test"}', call_id="c2")
+            assert _should_parallelize_tool_batch([tc1, tc2])
+        finally:
+            with _lock:
+                _parallel_safe_servers.discard("github")
+
+    def test_mixed_mcp_and_builtin_parallel(self):
+        """MCP parallel tools mixed with built-in parallel-safe tools."""
+        from run_agent import _should_parallelize_tool_batch
+        from tools.mcp_tool import _parallel_safe_servers, _lock
+        with _lock:
+            _parallel_safe_servers.add("docs")
+        try:
+            tc1 = _mock_tool_call(name="mcp_docs_search", arguments='{"query":"api"}', call_id="c1")
+            tc2 = _mock_tool_call(name="web_search", arguments='{"query":"test"}', call_id="c2")
+            assert _should_parallelize_tool_batch([tc1, tc2])
+        finally:
+            with _lock:
+                _parallel_safe_servers.discard("docs")
+
+    def test_mixed_parallel_and_serial_mcp_servers(self):
+        """One parallel MCP server + one non-parallel MCP server = sequential."""
+        from run_agent import _should_parallelize_tool_batch
+        from tools.mcp_tool import _parallel_safe_servers, _lock
+        with _lock:
+            _parallel_safe_servers.add("docs")
+            # "github" is NOT in _parallel_safe_servers
+        try:
+            tc1 = _mock_tool_call(name="mcp_docs_search", arguments='{"query":"api"}', call_id="c1")
+            tc2 = _mock_tool_call(name="mcp_github_list_repos", arguments='{"org":"openai"}', call_id="c2")
+            assert not _should_parallelize_tool_batch([tc1, tc2])
+        finally:
+            with _lock:
+                _parallel_safe_servers.discard("docs")
+
+
 class TestHandleMaxIterations:
     def test_returns_summary(self, agent):
         resp = _mock_response(content="Here is a summary of what I did.")

tests/run_agent/test_streaming.py

@@ -999,6 +999,88 @@ class TestAnthropicStreamCallbacks:
 
         assert touch_calls.count("receiving stream response") == len(events)
 
+    @patch("run_agent.AIAgent._replace_primary_openai_client")
+    def test_anthropic_stream_parser_valueerror_retries_before_delivery(
+        self, mock_replace, monkeypatch,
+    ):
+        """Malformed Anthropic event-stream frames retry instead of surfacing HTTP None."""
+        from run_agent import AIAgent
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://api.minimax.io/anthropic",
+            provider="minimax",
+            model="MiniMax-M2.7",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+        agent.api_mode = "anthropic_messages"
+        agent._interrupt_requested = False
+        monkeypatch.setenv("HERMES_STREAM_RETRIES", "1")
+
+        class _BadStream:
+            response = None
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                return False
+
+            def __iter__(self):
+                raise ValueError("expected ident at line 1 column 149")
+
+        final_message = SimpleNamespace(content=[], stop_reason="end_turn")
+        good_stream = MagicMock()
+        good_stream.__enter__ = MagicMock(return_value=good_stream)
+        good_stream.__exit__ = MagicMock(return_value=False)
+        good_stream.__iter__ = MagicMock(return_value=iter([]))
+        good_stream.get_final_message.return_value = final_message
+
+        agent._anthropic_client = MagicMock()
+        agent._anthropic_client.messages.stream.side_effect = [
+            _BadStream(),
+            good_stream,
+        ]
+
+        response = agent._interruptible_streaming_api_call({})
+
+        assert response is final_message
+        assert agent._anthropic_client.messages.stream.call_count == 2
+        assert mock_replace.call_count == 1
+
+    @patch("run_agent.AIAgent._replace_primary_openai_client")
+    def test_generic_anthropic_valueerror_still_propagates_without_stream_retry(
+        self, mock_replace, monkeypatch,
+    ):
+        """Only known provider stream parser ValueErrors are treated as transient."""
+        from run_agent import AIAgent
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://api.minimax.io/anthropic",
+            provider="minimax",
+            model="MiniMax-M2.7",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+        agent.api_mode = "anthropic_messages"
+        agent._interrupt_requested = False
+        monkeypatch.setenv("HERMES_STREAM_RETRIES", "1")
+
+        agent._anthropic_client = MagicMock()
+        agent._anthropic_client.messages.stream.side_effect = ValueError(
+            "invalid local request shape"
+        )
+
+        with pytest.raises(ValueError, match="invalid local request shape"):
+            agent._interruptible_streaming_api_call({})
+
+        assert agent._anthropic_client.messages.stream.call_count == 1
+        assert mock_replace.call_count == 0
+
 
 class TestPartialToolCallWarning:
     """Regression: when a stream dies mid tool-call argument generation after
@@ -1504,4 +1586,3 @@ class TestCopilotACPStreamingDecision:
             _use_streaming = False
 
         assert _use_streaming is True
-

tests/skills/test_darwinian_evolver_skill.py

@@ -0,0 +1,102 @@
+"""
+Smoke tests for the darwinian-evolver optional skill.
+
+We can't actually run the evolution loop in CI (it needs network + a paid LLM),
+so these tests verify:
+  - SKILL.md frontmatter conforms to the hardline format
+  - shipped scripts parse as valid Python
+  - the scripts reference the right env var / module paths
+"""
+from __future__ import annotations
+
+import ast
+import re
+from pathlib import Path
+
+import pytest
+import yaml
+
+SKILL_DIR = Path(__file__).resolve().parents[2] / "optional-skills" / "research" / "darwinian-evolver"
+
+
+@pytest.fixture(scope="module")
+def frontmatter() -> dict:
+    src = (SKILL_DIR / "SKILL.md").read_text()
+    m = re.search(r"^---\n(.*?)\n---", src, re.DOTALL)
+    assert m, "SKILL.md missing YAML frontmatter"
+    return yaml.safe_load(m.group(1))
+
+
+def test_skill_dir_exists() -> None:
+    assert SKILL_DIR.is_dir(), f"missing skill dir: {SKILL_DIR}"
+
+
+def test_skill_md_present() -> None:
+    assert (SKILL_DIR / "SKILL.md").is_file()
+
+
+def test_description_under_60_chars(frontmatter) -> None:
+    desc = frontmatter["description"]
+    assert len(desc) <= 60, f"description is {len(desc)} chars (hardline ≤60): {desc!r}"
+
+
+def test_name_matches_dir(frontmatter) -> None:
+    assert frontmatter["name"] == "darwinian-evolver"
+
+
+def test_platforms_excludes_windows(frontmatter) -> None:
+    # Upstream uses func_timeout (POSIX signals) and uv subprocess pipelines; the
+    # skill is gated [linux, macos]. If we ever port to Windows, update this test
+    # to assert ["linux", "macos", "windows"].
+    assert "windows" not in frontmatter["platforms"]
+    assert set(frontmatter["platforms"]) >= {"linux", "macos"}
+
+
+def test_author_credits_contributor(frontmatter) -> None:
+    author = frontmatter["author"]
+    assert "Bihruze" in author, f"author should credit the original contributor: {author!r}"
+
+
+def test_license_mit(frontmatter) -> None:
+    assert frontmatter["license"] == "MIT"
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "scripts/parrot_openrouter.py",
+        "scripts/show_snapshot.py",
+        "templates/custom_problem_template.py",
+    ],
+)
+def test_shipped_scripts_parse(path: str) -> None:
+    src = (SKILL_DIR / path).read_text()
+    ast.parse(src)  # raises SyntaxError on broken Python
+
+
+def test_parrot_script_uses_openrouter() -> None:
+    src = (SKILL_DIR / "scripts" / "parrot_openrouter.py").read_text()
+    assert "OPENROUTER_API_KEY" in src, "parrot driver should read OPENROUTER_API_KEY"
+    assert "openrouter.ai/api/v1" in src, "parrot driver should target OpenRouter"
+    assert "EVOLVER_MODEL" in src, "model should be overridable via EVOLVER_MODEL"
+
+
+def test_parrot_script_has_error_swallowing() -> None:
+    """Provider content-filter / rate-limit must not kill the run — see Pitfall 2."""
+    src = (SKILL_DIR / "scripts" / "parrot_openrouter.py").read_text()
+    assert "LLM_ERROR" in src, "_prompt_llm should swallow provider errors and tag them"
+
+
+def test_skill_calls_out_agpl(frontmatter) -> None:
+    """The upstream tool is AGPL-3.0. The skill MUST flag this so users don't
+    import it into MIT-licensed code by accident."""
+    src = (SKILL_DIR / "SKILL.md").read_text()
+    assert "AGPL" in src, "SKILL.md must mention upstream AGPL license"
+
+
+def test_skill_pitfalls_section_present() -> None:
+    src = (SKILL_DIR / "SKILL.md").read_text()
+    assert "## Pitfalls" in src
+    # Pitfalls we discovered during the spike — keep them in sync with reality.
+    assert "Initial organism must be viable" in src
+    assert "generator" in src  # loop.run() pitfall

tests/test_sanitize_tool_error.py

@@ -0,0 +1,137 @@
+"""Tests for `_sanitize_tool_error` in model_tools.
+
+Ported from ironclaw#1639 — defense-in-depth on tool exception strings before
+they enter the model's `tool` message content. Note that `json.dumps()` in
+`handle_function_call` already handles quote/backslash escaping at the wire
+layer; this helper exists to strip structural framing tokens the model
+itself might react to (XML role tags, CDATA, markdown code fences) and to
+cap pathological lengths.
+"""
+from __future__ import annotations
+
+from model_tools import _sanitize_tool_error, _TOOL_ERROR_MAX_LEN
+
+
+class TestRoleTagStripping:
+    def test_strips_tool_call_tags(self):
+        out = _sanitize_tool_error("bad <tool_call>injected</tool_call> happened")
+        assert "<tool_call>" not in out
+        assert "</tool_call>" not in out
+        assert "bad injected happened" in out
+
+    def test_strips_function_call_tags(self):
+        out = _sanitize_tool_error("<function_call>x</function_call>")
+        assert "<function_call>" not in out
+        assert "</function_call>" not in out
+
+    def test_strips_role_tags(self):
+        # Each of these should be stripped
+        for tag in ("system", "assistant", "user", "result", "response", "output", "input"):
+            raw = f"prefix <{tag}>hi</{tag}> suffix"
+            out = _sanitize_tool_error(raw)
+            assert f"<{tag}>" not in out, f"failed to strip <{tag}>"
+            assert f"</{tag}>" not in out, f"failed to strip </{tag}>"
+
+    def test_role_tag_strip_is_case_insensitive(self):
+        out = _sanitize_tool_error("<TOOL_CALL>x</Tool_Call>")
+        assert "<" not in out.replace("[TOOL_ERROR]", "")  # only the prefix bracket survives
+
+    def test_unrelated_xml_kept(self):
+        # We intentionally only strip the role-like tag whitelist, not all XML
+        out = _sanitize_tool_error("Error parsing <ParseError>line 5</ParseError>")
+        assert "<ParseError>" in out
+
+
+class TestCDATAStripping:
+    def test_strips_cdata(self):
+        out = _sanitize_tool_error("error: <![CDATA[malicious]]> here")
+        assert "<![CDATA[" not in out
+        assert "]]>" not in out
+
+    def test_strips_multiline_cdata(self):
+        out = _sanitize_tool_error("a\n<![CDATA[line1\nline2]]>\nb")
+        assert "CDATA" not in out
+        assert "a" in out and "b" in out
+
+
+class TestCodeFenceStripping:
+    def test_strips_leading_fence_with_lang(self):
+        out = _sanitize_tool_error("```json\n{\"x\": 1}")
+        assert not out.replace("[TOOL_ERROR] ", "").startswith("```")
+
+    def test_strips_trailing_fence(self):
+        out = _sanitize_tool_error("payload\n```")
+        assert not out.rstrip().endswith("```")
+
+    def test_strips_bare_fence(self):
+        out = _sanitize_tool_error("```\nstuff")
+        assert "```" not in out.split("\n")[0]
+
+
+class TestTruncation:
+    def test_caps_long_input(self):
+        long = "A" * (_TOOL_ERROR_MAX_LEN * 2)
+        out = _sanitize_tool_error(long)
+        # Total length is prefix + truncated body
+        body = out[len("[TOOL_ERROR] "):]
+        assert len(body) == _TOOL_ERROR_MAX_LEN
+        assert body.endswith("...")
+
+    def test_does_not_truncate_short_input(self):
+        msg = "short error"
+        out = _sanitize_tool_error(msg)
+        assert "..." not in out
+        assert msg in out
+
+
+class TestEnvelope:
+    def test_wraps_with_prefix(self):
+        out = _sanitize_tool_error("oh no")
+        assert out.startswith("[TOOL_ERROR] ")
+
+    def test_empty_input(self):
+        out = _sanitize_tool_error("")
+        assert out == "[TOOL_ERROR] "
+
+    def test_preserves_normal_error_text(self):
+        msg = "Error executing read_file: FileNotFoundError: /tmp/missing"
+        out = _sanitize_tool_error(msg)
+        assert msg in out
+
+
+class TestHandleFunctionCallIntegration:
+    """Verify handle_function_call routes exception-path errors through the sanitizer.
+
+    Note: the "Unknown tool: ..." early-return in tools/registry.py is a
+    *different* code path from `except Exception` in handle_function_call —
+    that one returns directly without sanitization (and there's nothing to
+    sanitize in a hardcoded format string anyway). This test exercises the
+    real exception path by passing args that make a known tool raise.
+    """
+
+    def test_exception_path_error_is_sanitized(self):
+        import json
+        from model_tools import handle_function_call
+        from tools.registry import registry as _registry
+
+        # Force a known tool to raise with a payload containing role tags.
+        def boom(_args, **_kwargs):
+            raise RuntimeError("<tool_call>injected</tool_call> boom")
+
+        all_tools = _registry.get_all_tool_names()
+        assert all_tools, "no tools registered — test environment broken"
+        target = all_tools[0]
+        original = _registry._tools[target].handler
+        _registry._tools[target].handler = boom
+        try:
+            result_str = handle_function_call(target, {})
+        finally:
+            _registry._tools[target].handler = original
+
+        payload = json.loads(result_str)
+        assert "error" in payload, payload
+        assert payload["error"].startswith("[TOOL_ERROR] "), payload["error"]
+        # Role-tag stripping carried through
+        assert "<tool_call>" not in payload["error"]
+        assert "</tool_call>" not in payload["error"]
+        assert "boom" in payload["error"]

tests/tools/test_approval.py

@@ -1102,3 +1102,206 @@ class TestDetectSudoStdin:
             "make 2>&1 | tee build.log"
         )
         assert is_dangerous is False
+
+
+class TestMacOSPrivateSystemPaths:
+    """Inspired by Claude Code 2.1.113 "dangerous path protection".
+
+    On macOS, /etc, /var, /tmp, /home are symlinks to
+    /private/{etc,var,tmp,home}. A command that writes to
+    /private/etc/sudoers works identically to /etc/sudoers but bypasses
+    a plain "/etc/" pattern check.  These tests guard the shared
+    _SYSTEM_CONFIG_PATH fragment used across redirect / tee / cp / mv /
+    install / sed -i patterns.
+    """
+
+    def test_private_etc_redirect(self):
+        dangerous, _, desc = detect_dangerous_command(
+            "echo 'root ALL=NOPASSWD: ALL' > /private/etc/sudoers"
+        )
+        assert dangerous is True
+        assert "system config" in desc.lower()
+
+    def test_private_var_redirect(self):
+        dangerous, _, _ = detect_dangerous_command(
+            "echo payload > /private/var/db/dslocal/nodes/x"
+        )
+        assert dangerous is True
+
+    def test_private_etc_via_tee(self):
+        dangerous, _, desc = detect_dangerous_command(
+            "echo malicious | tee /private/etc/hosts"
+        )
+        assert dangerous is True
+        assert "tee" in desc.lower() or "system" in desc.lower()
+
+    def test_private_etc_cp(self):
+        dangerous, _, desc = detect_dangerous_command(
+            "cp malicious.conf /private/etc/hosts"
+        )
+        assert dangerous is True
+        assert "copy" in desc.lower() or "system config" in desc.lower()
+
+    def test_private_etc_mv(self):
+        dangerous, _, _ = detect_dangerous_command(
+            "mv evil /private/etc/ssh/sshd_config"
+        )
+        assert dangerous is True
+
+    def test_private_etc_install(self):
+        dangerous, _, _ = detect_dangerous_command(
+            "install -m 600 key /private/etc/ssh/keys"
+        )
+        assert dangerous is True
+
+    def test_private_etc_sed_in_place(self):
+        dangerous, _, desc = detect_dangerous_command(
+            "sed -i 's/root/pwned/' /private/etc/passwd"
+        )
+        assert dangerous is True
+        assert "in-place" in desc.lower() or "system config" in desc.lower()
+
+    def test_private_var_sed_long_flag(self):
+        dangerous, _, _ = detect_dangerous_command(
+            "sed --in-place 's/x/y/' /private/var/log/wtmp"
+        )
+        assert dangerous is True
+
+    def test_private_tmp_cp(self):
+        dangerous, _, _ = detect_dangerous_command(
+            "cp rootkit /private/tmp/payload"
+        )
+        assert dangerous is True
+
+    def test_ls_private_is_safe(self):
+        """Reading under /private/ must not trigger approval."""
+        dangerous, _, _ = detect_dangerous_command("ls /private")
+        assert dangerous is False
+
+    def test_echo_mentioning_private_path_is_safe(self):
+        """Literal mention of /private/etc in an echo string must not fire."""
+        dangerous, _, _ = detect_dangerous_command(
+            "echo 'the macOS path is /private/etc on disk'"
+        )
+        assert dangerous is False
+
+
+class TestKillallKillSignals:
+    """Inspired by Claude Code 2.1.113 expanded deny rules.
+
+    The existing pattern caught `pkill -9` but not the equivalent
+    `killall -9` / `-KILL` / `-s KILL` / `-r <regex>` broad sweeps that
+    can wipe out unrelated processes.
+    """
+
+    def test_killall_dash_9(self):
+        dangerous, _, desc = detect_dangerous_command("killall -9 firefox")
+        assert dangerous is True
+        assert "kill" in desc.lower()
+
+    def test_killall_dash_kill(self):
+        dangerous, _, _ = detect_dangerous_command("killall -KILL firefox")
+        assert dangerous is True
+
+    def test_killall_dash_sigkill(self):
+        dangerous, _, _ = detect_dangerous_command("killall -SIGKILL firefox")
+        assert dangerous is True
+
+    def test_killall_dash_s_kill(self):
+        dangerous, _, _ = detect_dangerous_command("killall -s KILL firefox")
+        assert dangerous is True
+
+    def test_killall_dash_s_signum(self):
+        dangerous, _, _ = detect_dangerous_command("killall -s 9 firefox")
+        assert dangerous is True
+
+    def test_killall_regex(self):
+        """killall -r <regex> is a broad sweep; require approval."""
+        dangerous, _, desc = detect_dangerous_command("killall -r 'fire.*'")
+        assert dangerous is True
+        assert "regex" in desc.lower() or "kill" in desc.lower()
+
+    def test_killall_combined_flags(self):
+        dangerous, _, _ = detect_dangerous_command("killall -9 -r 'herm.*'")
+        assert dangerous is True
+
+    def test_killall_list_signals_is_safe(self):
+        """`killall -l` lists signals and is harmless — must not fire."""
+        dangerous, _, _ = detect_dangerous_command("killall -l")
+        assert dangerous is False
+
+    def test_killall_version_is_safe(self):
+        dangerous, _, _ = detect_dangerous_command("killall -V")
+        assert dangerous is False
+
+
+class TestFindExecdir:
+    """Inspired by Claude Code 2.1.113 tightening of find rules.
+
+    `find -execdir rm` has the same destructive effect as `find -exec rm`
+    but ran in each match's directory. Previously missed because the
+    pattern required a literal `-exec ` followed by a space.
+    """
+
+    def test_find_execdir_rm(self):
+        dangerous, _, desc = detect_dangerous_command(
+            "find . -execdir rm {} \\;"
+        )
+        assert dangerous is True
+        assert "find" in desc.lower() or "rm" in desc.lower()
+
+    def test_find_execdir_with_absolute_rm(self):
+        dangerous, _, _ = detect_dangerous_command(
+            "find /var -execdir /bin/rm -rf {} \\;"
+        )
+        assert dangerous is True
+
+    def test_find_exec_rm_still_caught(self):
+        """Original -exec pattern must still fire (regression guard)."""
+        dangerous, _, _ = detect_dangerous_command(
+            "find . -exec rm {} \\;"
+        )
+        assert dangerous is True
+
+    def test_find_execdir_ls_is_safe(self):
+        """-execdir with a read-only command is not dangerous."""
+        dangerous, _, _ = detect_dangerous_command(
+            "find . -execdir ls {} \\;"
+        )
+        assert dangerous is False
+
+
+class TestEtcPatternsUnaffectedByRefactor:
+    """Regression guard: the /etc/ patterns were refactored to share the
+    _SYSTEM_CONFIG_PATH fragment with the /private/ mirror. Make sure the
+    existing /etc/ coverage remains identical.
+    """
+
+    def test_etc_redirect(self):
+        dangerous, _, _ = detect_dangerous_command("echo x > /etc/hosts")
+        assert dangerous is True
+
+    def test_etc_cp(self):
+        dangerous, _, _ = detect_dangerous_command("cp evil /etc/hosts")
+        assert dangerous is True
+
+    def test_etc_sed_inline(self):
+        dangerous, _, _ = detect_dangerous_command(
+            "sed -i 's/a/b/' /etc/hosts"
+        )
+        assert dangerous is True
+
+    def test_etc_tee(self):
+        dangerous, _, _ = detect_dangerous_command(
+            "echo x | tee /etc/hosts"
+        )
+        assert dangerous is True
+
+    def test_cat_etc_hostname_is_safe(self):
+        """Reading /etc/ files is safe — only writes require approval."""
+        dangerous, _, _ = detect_dangerous_command("cat /etc/hostname")
+        assert dangerous is False
+
+    def test_grep_etc_passwd_is_safe(self):
+        dangerous, _, _ = detect_dangerous_command("grep root /etc/passwd")
+        assert dangerous is False

tests/tools/test_delegate.py

@@ -890,6 +890,63 @@ class TestDelegationCredentialResolution(unittest.TestCase):
         self.assertEqual(creds["api_key"], "local-key")
         self.assertEqual(creds["api_mode"], "chat_completions")
 
+    def test_direct_endpoint_auto_detects_anthropic_messages_suffix(self):
+        # Issue #10213: Azure AI Foundry exposes Anthropic-compatible models at
+        # a /anthropic URL suffix. Subagents must pick anthropic_messages
+        # automatically, matching the main agent's runtime resolver.
+        parent = _make_mock_parent(depth=0)
+        cfg = {
+            "model": "claude-opus-4-6",
+            "provider": "custom",
+            "base_url": "https://myfoundry.services.ai.azure.com/anthropic",
+            "api_key": "foundry-key",
+        }
+        creds = _resolve_delegation_credentials(cfg, parent)
+        self.assertEqual(creds["provider"], "custom")
+        self.assertEqual(creds["base_url"], "https://myfoundry.services.ai.azure.com/anthropic")
+        self.assertEqual(creds["api_key"], "foundry-key")
+        self.assertEqual(creds["api_mode"], "anthropic_messages")
+
+    def test_direct_endpoint_honors_explicit_api_mode(self):
+        # When delegation.api_mode is set explicitly, it overrides URL-based
+        # detection so users can force a transport on non-standard endpoints.
+        parent = _make_mock_parent(depth=0)
+        cfg = {
+            "model": "claude-opus-4-6",
+            "provider": "custom",
+            "base_url": "https://proxy.example.com/v1",
+            "api_key": "proxy-key",
+            "api_mode": "anthropic_messages",
+        }
+        creds = _resolve_delegation_credentials(cfg, parent)
+        self.assertEqual(creds["api_mode"], "anthropic_messages")
+
+    def test_direct_endpoint_explicit_api_mode_overrides_url_detection(self):
+        # Explicit api_mode in config always wins over auto-detection.
+        parent = _make_mock_parent(depth=0)
+        cfg = {
+            "model": "claude-opus-4-6",
+            "provider": "custom",
+            "base_url": "https://myfoundry.services.ai.azure.com/anthropic",
+            "api_key": "foundry-key",
+            "api_mode": "chat_completions",
+        }
+        creds = _resolve_delegation_credentials(cfg, parent)
+        self.assertEqual(creds["api_mode"], "chat_completions")
+
+    def test_direct_endpoint_invalid_api_mode_falls_back_to_detection(self):
+        # An invalid api_mode string must not break detection; fall back to URL heuristic.
+        parent = _make_mock_parent(depth=0)
+        cfg = {
+            "model": "claude-opus-4-6",
+            "provider": "custom",
+            "base_url": "https://myfoundry.services.ai.azure.com/anthropic",
+            "api_key": "foundry-key",
+            "api_mode": "garbage",
+        }
+        creds = _resolve_delegation_credentials(cfg, parent)
+        self.assertEqual(creds["api_mode"], "anthropic_messages")
+
     def test_direct_endpoint_returns_none_api_key_when_not_configured(self):
         # When base_url is set without api_key, api_key should be None so
         # _build_child_agent inherits the parent's key (effective_api_key = override or parent).

tests/tools/test_mcp_tool.py

@@ -3762,3 +3762,135 @@ class TestRegisterMcpServers:
                 )
 
         _servers.pop("srv", None)
+
+
+# ---------------------------------------------------------------------------
+# Tests for parallel tool call support (port from openai/codex#17667)
+# ---------------------------------------------------------------------------
+
+class TestMcpParallelToolCalls:
+    """Tests for the supports_parallel_tool_calls config option."""
+
+    def test_is_mcp_tool_parallel_safe_non_mcp_tool(self):
+        """Non-MCP tool names always return False."""
+        from tools.mcp_tool import is_mcp_tool_parallel_safe
+        assert is_mcp_tool_parallel_safe("web_search") is False
+        assert is_mcp_tool_parallel_safe("read_file") is False
+        assert is_mcp_tool_parallel_safe("terminal") is False
+        assert is_mcp_tool_parallel_safe("") is False
+
+    def test_is_mcp_tool_parallel_safe_no_servers(self):
+        """MCP tool from unknown server returns False."""
+        from tools.mcp_tool import is_mcp_tool_parallel_safe, _parallel_safe_servers, _lock
+        with _lock:
+            _parallel_safe_servers.clear()
+        assert is_mcp_tool_parallel_safe("mcp_docs_search") is False
+
+    def test_is_mcp_tool_parallel_safe_with_flag(self):
+        """MCP tool from a parallel-safe server returns True."""
+        from tools.mcp_tool import is_mcp_tool_parallel_safe, _parallel_safe_servers, _lock
+        with _lock:
+            _parallel_safe_servers.add("docs")
+        try:
+            assert is_mcp_tool_parallel_safe("mcp_docs_search") is True
+            assert is_mcp_tool_parallel_safe("mcp_docs_read_file") is True
+            # Different server should be False
+            assert is_mcp_tool_parallel_safe("mcp_github_list_repos") is False
+        finally:
+            with _lock:
+                _parallel_safe_servers.discard("docs")
+
+    def test_is_mcp_tool_parallel_safe_server_with_underscores(self):
+        """Server names containing underscores are correctly matched."""
+        from tools.mcp_tool import is_mcp_tool_parallel_safe, _parallel_safe_servers, _lock
+        with _lock:
+            _parallel_safe_servers.add("my_server")
+        try:
+            assert is_mcp_tool_parallel_safe("mcp_my_server_query") is True
+        finally:
+            with _lock:
+                _parallel_safe_servers.discard("my_server")
+
+    def test_is_mcp_tool_parallel_safe_no_tool_suffix(self):
+        """Tool name that is just 'mcp_{server}' without a tool part returns False."""
+        from tools.mcp_tool import is_mcp_tool_parallel_safe, _parallel_safe_servers, _lock
+        with _lock:
+            _parallel_safe_servers.add("docs")
+        try:
+            # "mcp_docs" has no tool part after the server name
+            assert is_mcp_tool_parallel_safe("mcp_docs") is False
+            # "mcp_docs_" has empty tool part
+            assert is_mcp_tool_parallel_safe("mcp_docs_") is False
+        finally:
+            with _lock:
+                _parallel_safe_servers.discard("docs")
+
+    def test_register_mcp_servers_tracks_parallel_flag(self):
+        """register_mcp_servers populates _parallel_safe_servers from config."""
+        from tools.mcp_tool import (
+            register_mcp_servers, _parallel_safe_servers, _lock,
+            sanitize_mcp_name_component,
+        )
+        fake_config = {
+            "parallel_srv": {
+                "command": "echo",
+                "supports_parallel_tool_calls": True,
+            },
+            "serial_srv": {
+                "command": "echo",
+                "supports_parallel_tool_calls": False,
+            },
+            "default_srv": {
+                "command": "echo",
+                # no supports_parallel_tool_calls key
+            },
+        }
+        with patch("tools.mcp_tool._MCP_AVAILABLE", True), \
+             patch("tools.mcp_tool._ensure_mcp_loop"), \
+             patch("tools.mcp_tool._run_on_mcp_loop"), \
+             patch("tools.mcp_tool._existing_tool_names", return_value=[]):
+            register_mcp_servers(fake_config)
+
+        with _lock:
+            assert sanitize_mcp_name_component("parallel_srv") in _parallel_safe_servers
+            assert sanitize_mcp_name_component("serial_srv") not in _parallel_safe_servers
+            assert sanitize_mcp_name_component("default_srv") not in _parallel_safe_servers
+            # Cleanup
+            _parallel_safe_servers.discard(sanitize_mcp_name_component("parallel_srv"))
+
+    def test_register_mcp_servers_removes_parallel_flag_on_toggle(self):
+        """Toggling supports_parallel_tool_calls to false removes server from the set."""
+        from tools.mcp_tool import (
+            register_mcp_servers, _parallel_safe_servers, _lock,
+            sanitize_mcp_name_component,
+        )
+
+        # First registration: parallel enabled
+        config_on = {
+            "toggle_srv": {
+                "command": "echo",
+                "supports_parallel_tool_calls": True,
+            },
+        }
+        with patch("tools.mcp_tool._MCP_AVAILABLE", True), \
+             patch("tools.mcp_tool._ensure_mcp_loop"), \
+             patch("tools.mcp_tool._run_on_mcp_loop"), \
+             patch("tools.mcp_tool._existing_tool_names", return_value=[]):
+            register_mcp_servers(config_on)
+        with _lock:
+            assert sanitize_mcp_name_component("toggle_srv") in _parallel_safe_servers
+
+        # Second registration: parallel disabled
+        config_off = {
+            "toggle_srv": {
+                "command": "echo",
+                "supports_parallel_tool_calls": False,
+            },
+        }
+        with patch("tools.mcp_tool._MCP_AVAILABLE", True), \
+             patch("tools.mcp_tool._ensure_mcp_loop"), \
+             patch("tools.mcp_tool._run_on_mcp_loop"), \
+             patch("tools.mcp_tool._existing_tool_names", return_value=[]):
+            register_mcp_servers(config_off)
+        with _lock:
+            assert sanitize_mcp_name_component("toggle_srv") not in _parallel_safe_servers

tests/tools/test_x_search_tool.py

@@ -0,0 +1,438 @@
+"""Tests for the X (Twitter) Search tool backed by xAI Responses API.
+
+Covers:
+- HTTP request shape (URL, headers, payload, model from config)
+- Handle filter validation (allowed vs excluded mutual exclusion)
+- Inline url_citation extraction from message annotations
+- Structured error handling (4xx with code, 5xx retry, ReadTimeout retry)
+- Credential resolution: API key path, OAuth path, both-set preference, none-set
+- check_x_search_requirements gating in registry
+"""
+
+import json
+
+import requests
+
+
+class _FakeResponse:
+    def __init__(self, payload, *, status_code=200, text=None):
+        self._payload = payload
+        self.status_code = status_code
+        self.text = text if text is not None else json.dumps(payload)
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            err = requests.HTTPError(f"{self.status_code} Client Error")
+            err.response = self
+            raise err
+
+    def json(self):
+        return self._payload
+
+
+# ---------------------------------------------------------------------------
+# Original PR #10786 test coverage (HTTP shape, handle validation, citations,
+# retry behavior) — preserved verbatim. Uses XAI_API_KEY env var via the
+# default resolver path.
+# ---------------------------------------------------------------------------
+
+def test_x_search_posts_responses_request(monkeypatch):
+    from tools.x_search_tool import x_search_tool
+    from hermes_cli import __version__
+
+    captured = {}
+
+    def _fake_post(url, headers=None, json=None, timeout=None):
+        captured["url"] = url
+        captured["headers"] = headers
+        captured["json"] = json
+        captured["timeout"] = timeout
+        return _FakeResponse(
+            {
+                "output_text": "People on X are discussing xAI's latest launch.",
+                "citations": [{"url": "https://x.com/example/status/1", "title": "Example post"}],
+            }
+        )
+
+    monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
+    monkeypatch.setattr("requests.post", _fake_post)
+
+    result = json.loads(
+        x_search_tool(
+            query="What are people saying about xAI on X?",
+            allowed_x_handles=["xai", "@grok"],
+            from_date="2026-04-01",
+            to_date="2026-04-10",
+            enable_image_understanding=True,
+        )
+    )
+
+    tool_def = captured["json"]["tools"][0]
+    assert captured["url"] == "https://api.x.ai/v1/responses"
+    assert captured["headers"]["User-Agent"] == f"Hermes-Agent/{__version__}"
+    assert captured["json"]["model"] == "grok-4.20-reasoning"
+    assert captured["json"]["store"] is False
+    assert tool_def["type"] == "x_search"
+    assert tool_def["allowed_x_handles"] == ["xai", "grok"]
+    assert tool_def["from_date"] == "2026-04-01"
+    assert tool_def["to_date"] == "2026-04-10"
+    assert tool_def["enable_image_understanding"] is True
+    assert result["success"] is True
+    assert result["answer"] == "People on X are discussing xAI's latest launch."
+
+
+def test_x_search_rejects_conflicting_handle_filters(monkeypatch):
+    from tools.x_search_tool import x_search_tool
+
+    monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
+
+    result = json.loads(
+        x_search_tool(
+            query="latest xAI discussion",
+            allowed_x_handles=["xai"],
+            excluded_x_handles=["grok"],
+        )
+    )
+
+    assert result["error"] == "allowed_x_handles and excluded_x_handles cannot be used together"
+
+
+def test_x_search_extracts_inline_url_citations(monkeypatch):
+    from tools.x_search_tool import x_search_tool
+
+    def _fake_post(url, headers=None, json=None, timeout=None):
+        return _FakeResponse(
+            {
+                "output": [
+                    {
+                        "type": "message",
+                        "content": [
+                            {
+                                "type": "output_text",
+                                "text": "xAI posted an update on X.",
+                                "annotations": [
+                                    {
+                                        "type": "url_citation",
+                                        "url": "https://x.com/xai/status/123",
+                                        "title": "xAI update",
+                                        "start_index": 0,
+                                        "end_index": 3,
+                                    }
+                                ],
+                            }
+                        ],
+                    }
+                ]
+            }
+        )
+
+    monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
+    monkeypatch.setattr("requests.post", _fake_post)
+
+    result = json.loads(x_search_tool(query="latest post from xai"))
+
+    assert result["success"] is True
+    assert result["answer"] == "xAI posted an update on X."
+    assert result["inline_citations"] == [
+        {
+            "url": "https://x.com/xai/status/123",
+            "title": "xAI update",
+            "start_index": 0,
+            "end_index": 3,
+        }
+    ]
+
+
+def test_x_search_returns_structured_http_error(monkeypatch):
+    from tools.x_search_tool import x_search_tool
+
+    class _FailingResponse:
+        status_code = 403
+        text = '{"code":"forbidden","error":"x_search is not enabled for this model"}'
+
+        def json(self):
+            return {
+                "code": "forbidden",
+                "error": "x_search is not enabled for this model",
+            }
+
+        def raise_for_status(self):
+            err = requests.HTTPError("403 Client Error: Forbidden")
+            err.response = self
+            raise err
+
+    monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
+    monkeypatch.setattr("requests.post", lambda *a, **k: _FailingResponse())
+
+    result = json.loads(x_search_tool(query="latest xai discussion"))
+
+    assert result["success"] is False
+    assert result["provider"] == "xai"
+    assert result["tool"] == "x_search"
+    assert result["error_type"] == "HTTPError"
+    assert result["error"] == "forbidden: x_search is not enabled for this model"
+
+
+def test_x_search_retries_read_timeout_then_succeeds(monkeypatch):
+    from tools.x_search_tool import x_search_tool
+
+    calls = {"count": 0}
+
+    def _fake_post(url, headers=None, json=None, timeout=None):
+        calls["count"] += 1
+        if calls["count"] == 1:
+            raise requests.ReadTimeout("timed out")
+        return _FakeResponse(
+            {
+                "output_text": "Recovered after retry.",
+                "citations": [],
+            }
+        )
+
+    monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
+    monkeypatch.setattr("requests.post", _fake_post)
+    monkeypatch.setattr("tools.x_search_tool.time.sleep", lambda *_: None)
+
+    result = json.loads(x_search_tool(query="grok xai"))
+
+    assert calls["count"] == 2
+    assert result["success"] is True
+    assert result["answer"] == "Recovered after retry."
+
+
+def test_x_search_retries_5xx_then_succeeds(monkeypatch):
+    from tools.x_search_tool import x_search_tool
+
+    calls = {"count": 0}
+
+    def _fake_post(url, headers=None, json=None, timeout=None):
+        calls["count"] += 1
+        if calls["count"] == 1:
+            return _FakeResponse(
+                {"code": "Internal error", "error": "Service temporarily unavailable."},
+                status_code=500,
+            )
+        return _FakeResponse({"output_text": "Recovered after 5xx retry."})
+
+    monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
+    monkeypatch.setattr("requests.post", _fake_post)
+    monkeypatch.setattr("tools.x_search_tool.time.sleep", lambda *_: None)
+
+    result = json.loads(x_search_tool(query="grok xai"))
+
+    assert calls["count"] == 2
+    assert result["success"] is True
+    assert result["answer"] == "Recovered after 5xx retry."
+
+
+# ---------------------------------------------------------------------------
+# Credential-resolution coverage — the OAuth-or-API-key gating contract.
+# ---------------------------------------------------------------------------
+
+def _no_xai_env(monkeypatch):
+    """Strip any XAI_* env vars so the resolver doesn't see a leaked dev key."""
+    for var in ("XAI_API_KEY", "XAI_BASE_URL", "HERMES_XAI_BASE_URL"):
+        monkeypatch.delenv(var, raising=False)
+
+
+def test_x_search_uses_xai_oauth_when_only_oauth_available(monkeypatch):
+    """OAuth-only user: credential_source should be ``xai-oauth``."""
+    from tools.registry import invalidate_check_fn_cache
+    from tools.x_search_tool import check_x_search_requirements, x_search_tool
+
+    _no_xai_env(monkeypatch)
+
+    def _fake_resolve():
+        return {
+            "provider": "xai-oauth",
+            "api_key": "oauth-bearer-token",
+            "base_url": "https://api.x.ai/v1",
+        }
+
+    monkeypatch.setattr(
+        "tools.x_search_tool.resolve_xai_http_credentials", _fake_resolve
+    )
+    invalidate_check_fn_cache()
+
+    assert check_x_search_requirements() is True
+
+    captured = {}
+
+    def _fake_post(url, headers=None, json=None, timeout=None):
+        captured["headers"] = headers
+        return _FakeResponse({"output_text": "Found posts via OAuth."})
+
+    monkeypatch.setattr("requests.post", _fake_post)
+
+    result = json.loads(x_search_tool(query="anything about xai"))
+
+    assert result["success"] is True
+    assert result["credential_source"] == "xai-oauth"
+    assert captured["headers"]["Authorization"] == "Bearer oauth-bearer-token"
+
+
+def test_x_search_uses_api_key_when_only_xai_api_key_set(monkeypatch):
+    """API-key-only user: credential_source should be ``xai``."""
+    from tools.registry import invalidate_check_fn_cache
+    from tools.x_search_tool import check_x_search_requirements, x_search_tool
+
+    _no_xai_env(monkeypatch)
+
+    def _fake_resolve():
+        # Real ``resolve_xai_http_credentials`` returns ``"xai"`` when it
+        # falls through to the XAI_API_KEY env var path.
+        return {
+            "provider": "xai",
+            "api_key": "raw-api-key",
+            "base_url": "https://api.x.ai/v1",
+        }
+
+    monkeypatch.setattr(
+        "tools.x_search_tool.resolve_xai_http_credentials", _fake_resolve
+    )
+    invalidate_check_fn_cache()
+
+    assert check_x_search_requirements() is True
+
+    captured = {}
+
+    def _fake_post(url, headers=None, json=None, timeout=None):
+        captured["headers"] = headers
+        return _FakeResponse({"output_text": "Found posts via API key."})
+
+    monkeypatch.setattr("requests.post", _fake_post)
+
+    result = json.loads(x_search_tool(query="anything"))
+
+    assert result["success"] is True
+    assert result["credential_source"] == "xai"
+    assert captured["headers"]["Authorization"] == "Bearer raw-api-key"
+
+
+def test_x_search_prefers_oauth_when_both_available(monkeypatch):
+    """Both credentials present: OAuth wins (matches Teknium's billing preference).
+
+    The real ordering is implemented in ``tools.xai_http.resolve_xai_http_credentials``
+    — OAuth runtime first, fallback OAuth resolver second, ``XAI_API_KEY`` third.
+    This test exercises the contract by having the resolver return the OAuth
+    bearer (the ``xai-oauth`` ``provider`` tag is the marker).
+    """
+    from tools.registry import invalidate_check_fn_cache
+    from tools.x_search_tool import x_search_tool
+
+    monkeypatch.setenv("XAI_API_KEY", "raw-api-key")
+
+    # Mimic xai_http's preference: OAuth wins, so we return the OAuth tuple
+    # even though XAI_API_KEY is also set.
+    def _fake_resolve():
+        return {
+            "provider": "xai-oauth",
+            "api_key": "oauth-bearer-token",
+            "base_url": "https://api.x.ai/v1",
+        }
+
+    monkeypatch.setattr(
+        "tools.x_search_tool.resolve_xai_http_credentials", _fake_resolve
+    )
+    invalidate_check_fn_cache()
+
+    captured = {}
+
+    def _fake_post(url, headers=None, json=None, timeout=None):
+        captured["headers"] = headers
+        return _FakeResponse({"output_text": "OAuth preferred."})
+
+    monkeypatch.setattr("requests.post", _fake_post)
+
+    result = json.loads(x_search_tool(query="anything"))
+
+    assert result["credential_source"] == "xai-oauth"
+    assert captured["headers"]["Authorization"] == "Bearer oauth-bearer-token"
+
+
+def test_x_search_returns_tool_error_when_no_credentials(monkeypatch):
+    """No credentials anywhere: tool returns a clear error, not a 401 from xAI."""
+    from tools.registry import invalidate_check_fn_cache
+    from tools.x_search_tool import check_x_search_requirements, x_search_tool
+
+    _no_xai_env(monkeypatch)
+
+    def _fake_resolve():
+        return {
+            "provider": "xai",
+            "api_key": "",
+            "base_url": "https://api.x.ai/v1",
+        }
+
+    monkeypatch.setattr(
+        "tools.x_search_tool.resolve_xai_http_credentials", _fake_resolve
+    )
+    invalidate_check_fn_cache()
+
+    assert check_x_search_requirements() is False
+
+    # If a model somehow invokes the tool despite a False check_fn, the call
+    # surfaces a friendly error rather than an HTTP exception.
+    result = x_search_tool(query="anything")
+    assert "No xAI credentials available" in result
+    assert "hermes auth add xai-oauth" in result
+
+
+def test_x_search_check_fn_false_when_resolver_raises(monkeypatch):
+    """Resolver exceptions (e.g. expired token + failed refresh) gate the tool out."""
+    from tools.registry import invalidate_check_fn_cache
+    from tools.x_search_tool import check_x_search_requirements
+
+    _no_xai_env(monkeypatch)
+
+    def _boom():
+        raise RuntimeError("token revoked and refresh failed")
+
+    monkeypatch.setattr(
+        "tools.x_search_tool.resolve_xai_http_credentials", _boom
+    )
+    invalidate_check_fn_cache()
+
+    assert check_x_search_requirements() is False
+
+
+def test_x_search_honors_config_model_and_timeout(monkeypatch, tmp_path):
+    """``x_search.model`` and ``x_search.timeout_seconds`` override the defaults."""
+    from tools.x_search_tool import x_search_tool
+
+    monkeypatch.setenv("XAI_API_KEY", "xai-test-key")
+
+    # Patch the in-module config loader so tests don't touch ~/.hermes/config.yaml.
+    monkeypatch.setattr(
+        "tools.x_search_tool._load_x_search_config",
+        lambda: {"model": "grok-custom-test", "timeout_seconds": 45, "retries": 0},
+    )
+
+    captured = {}
+
+    def _fake_post(url, headers=None, json=None, timeout=None):
+        captured["model"] = json["model"]
+        captured["timeout"] = timeout
+        return _FakeResponse({"output_text": "Custom model OK."})
+
+    monkeypatch.setattr("requests.post", _fake_post)
+
+    result = json.loads(x_search_tool(query="anything"))
+
+    assert result["success"] is True
+    assert captured["model"] == "grok-custom-test"
+    assert captured["timeout"] == 45
+
+
+def test_x_search_registered_in_registry_with_check_fn():
+    """The tool is registered under the x_search toolset with the gating check_fn."""
+    import tools.x_search_tool  # noqa: F401 — ensures registration runs
+    from tools.registry import registry
+
+    entry = registry.get_entry("x_search")
+    assert entry is not None
+    assert entry.toolset == "x_search"
+    assert entry.check_fn is not None
+    assert entry.check_fn.__name__ == "check_x_search_requirements"
+    assert "XAI_API_KEY" in entry.requires_env
+    assert entry.emoji == "🐦"