Agent tool-call loop detection and guardrails

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #53565 fix(gateway,feishu): owned executors survive default-executor shutdown (#10849)

  merged to main

  files: gateway/run.py, plugins/platforms/feishu/adapter.py, tests/gateway/test_feishu_sdk_executor.py, tests/gateway/test_session_env.py, tests/gateway/test_stream_consumer_thread_routing.py

  Show the diff

  • gateway/run.py

    @@ -25,6 +25,7 @@
         pass
     
     import asyncio
    +import concurrent.futures
     import dataclasses
     import inspect
     import json
    @@ -2641,7 +2642,11 @@ def __init__(self, config: Optional[GatewayConfig] = None):
             self._restart_command_source: Optional[SessionSource] = None
             self._stop_task: Optional[asyncio.Task] = None
             self._restart_task: Optional[asyncio.Task] = None
    -        
    +        self._executor_lock = threading.Lock()
    +        self._executor: Optional[concurrent.futures.ThreadPoolExecutor] = None
    +        # Set on gateway stop so the recreate-on-shutdown path can't resurrect
    +        # the pool during a real shutdown.
    +        self._executor_closing = False
             # Track running agents per session for interrupt support
             # Key: session_key, Value: AIAgent instance
             self._running_agents: Dict[str, Any] = {}
    @@ -7303,6 +7308,7 @@ def _phase_elapsed() -> float:
                         _db.close()
                     except Exception as _e:
                         logger.debug("SessionDB close error: %s", _e)
    +            GatewayRunner._shutdown_executor(self)
                 logger.info(
                     "Shutdown phase: SessionDB close done at +%.2fs",
                     _phase_elapsed(),
    @@ -13396,7 +13402,50 @@ async def _run_in_executor_with_context(self, func, *args):
             """Run blocking work in the thread pool while preserving session contextvars."""
             loop = asyncio.get_running_loop()
             ctx = copy_context()
    -        return await loop.run_in_executor(None, ctx.run, func, *args)
    +        return await loop.run_in_executor(
    +            self._get_executor(),
    +            ctx.run,
    +            func,
    +            *args,
    +        )
    +
    +    def _get_executor(self) -> concurrent.futures.ThreadPoolExecutor:
    +        """Return the gateway-owned executor for blocking agent work."""
    +        lock = getattr(self, "_executor_lock", None)
    +        if lock is None:
    +            lock = threading.Lock()
    +            self._executor_lock = lock
    +
    +        with lock:
    +            if getattr(self, "_executor_closing", False):
    +                raise RuntimeError("Gateway is shutting down; executor unavailable")
    +            executor = getattr(self, "_executor", None)
    +            if executor is None or getattr(executor, "_shutdown", False):
    +                executor = concurrent.futures.ThreadPoolExecutor(
    +                    max_workers=10,
    +                    thread_name_prefix="hermes-gateway",
    +                )
    +                self._executor = executor
    +            return executor
    +
    +    def _shutdown_executor(self) -> None:
    +        """Stop the gateway-owned executor without touching the loop default."""
    +        lock = getattr(self, "_executor_lock", None)
    +        if lock is None:
    +            return
    +
    +        with lock:
    +            self._executor_closing = True
    +            executor = getattr(self, "_executor", None)
    +            self._executor = None
    +
    +        if executor is None:
    +            return
    +
    +        try:
    +            executor.shutdown(wait=False, cancel_futures=True)
    +        except TypeError:
    +            executor.shutdown(wait=False)
     
         def _decide_image_input_mode(self) -> str:
             """Resolve the image-input routing for the currently active model.

  • plugins/platforms/feishu/adapter.py

    @@ -49,6 +49,7 @@
     
     import asyncio
     import collections
    +import concurrent.futures
     import hashlib
     import hmac
     import itertools
    @@ -1430,6 +1431,16 @@ def __init__(self, config: PlatformConfig):
             self._settings = self._load_settings(config.extra or {})
             self._apply_settings(self._settings)
             self._client: Optional[Any] = None
    +        # Adapter-owned thread pool for blocking Feishu SDK calls. Routing SDK
    +        # work through this pool (instead of asyncio's shared default executor)
    +        # means a torn-down default executor can no longer wedge sends with
    +        # "Executor shutdown has been called" — the pool is recreated on demand
    +        # if it has been shut down. See issue #10849.
    +        self._sdk_executor_lock = threading.Lock()
    +        self._sdk_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None
    +        # Set on disconnect/shutdown so a real teardown can't be resurrected
    +        # by the recreate-on-shutdown path; cleared on connect for reconnects.
    +        self._sdk_executor_closing = False
             self._ws_client: Optional[Any] = None
             self._ws_future: Optional[asyncio.Future] = None
             self._ws_thread_loop: Optional[asyncio.AbstractEventLoop] = None
    @@ -1641,8 +1652,56 @@ def _build_event_handler(self) -> Any:
                 .build()
             )
     
    +    def _get_sdk_executor(self) -> concurrent.futures.ThreadPoolExecutor:
    +        """Return the adapter-owned executor for blocking Feishu SDK calls.
    +
    +        Recreates the pool if it was never built or was shut down by an
    +        *external* teardown of the loop's default executor, so that can no
    +        longer permanently wedge sends (#10849). Refuses to resurrect once
    +        the adapter itself is closing — a real disconnect/shutdown stays shut.
    +        """
    +        lock = getattr(self, "_sdk_executor_lock", None)
    +        if lock is None:
    +            lock = threading.Lock()
    +            self._sdk_executor_lock = lock
    +        with lock:
    +            if getattr(self, "_sdk_executor_closing", False):
    +                raise RuntimeError("Feishu adapter is shutting down; SDK executor unavailable")
    +            executor = getattr(self, "_sdk_executor", None)
    +            if executor is None or getattr(executor, "_shutdown", False):
    +                executor = concurrent.futures.ThreadPoolExecutor(
    +                    max_workers=10,
    +                    thread_name_prefix="hermes-feishu-sdk",
    +                )
    +                self._sdk_executor = executor
    +            return executor
    +
    +    async def _run_blocking(self, func, *args):
    +        """Run a blocking Feishu SDK call on the adapter-owned thread pool."""
    +        loop = asyncio.get_running_loop()
    +        return await loop.run_in_executor(self._get_sdk_executor(), func, *args)
    +
    +    def _shutdown_sdk_executor(self) -> None:
    +        """Stop the adapter-owned SDK executor without touching the loop default."""
    +        lock = getattr(self, "_sdk_executor_lock", None)
    +        if lock is None:
    +            return
    +        with lock:
    +            self._sdk_executor_closing = True
    +            executor = getattr(self, "_sdk_executor", None)
    +            self._sdk_executor = None
    +        if executor is None:
    +            return
    +        try:
    +            executor.shutdown(wait=False, cancel_futures=True)
    +        except TypeError:
    +            executor.shutdown(wait=False)
    +
         async def connect(self, *, is_reconnect: bool = False) -> bool:
             """Connect to Feishu/Lark."""
    +        # A fresh connect (or reconnect) re-arms the SDK executor after a prior
    +        # disconnect set the closing flag.
    +        self._sdk_executor_closing = False
             if not FEISHU_AVAILABLE:
                 logger.error("[Feishu] lark-oapi not installed")
                 return False
    @@ -1730,6 +1789,7 @@ def cancel_all_tasks() -> None:
             self._ws_thread_loop = None
             self._loop = None
             self._event_handler = None
    +        self._shutdown_sdk_executor()
             self._persist_seen_message_ids()
             await self._release_app_lock()
     
    @@ -1846,7 +1906,7 @@ async def edit_message(
                 msg_type, payload = self._build_outbound_payload(content)
                 body = self._build_update_message_body(msg_type=msg_type, content=payload)
                 request = self._build_update_message_request(message_id=message_id, request_body=body)
    -            response = await asyncio.to_thread(self._client.im.v1.message.update, request)
    +            response = await self._run_blocking(self._client.im.v1.message.update, request)
                 result = self._finalize_send_result(response, "update failed")
                 if not result.success and msg_type == "post" and _POST_CONTENT_INVALID_RE.search(result.error or ""):
                     logger.warning("[Feishu] Invalid post update payload rejected by API; falling back to plain text")
    @@ -1855,7 +1915,7 @@ async def edit_message(
                         content=json.dumps({"text": _strip_markdown_to_plain_text(content)}, ensure_ascii=False),
                     )
                     fallback_request = self._build_update_message_request(message_id=message_id, request_body=fallback_body)
    -                fallback_response = await asyncio.to_thread(self._client.im.v1.message.update, fallback_request)
    +                fallback_response = await self._run_blocking(self._client.im.v1.message.update, fallback_request)
                     result = self._finalize_send_result(fallback_response, "update failed")
                 if result.success:
                     result.message_id = message_id
    @@ -2128,7 +2188,7 @@ async def send_image_file(
                     image=image_file,
                 )
                 request = self._build_image_upload_request(body)
    -            upload_response = await asyncio.to_thread(self._client.im.v1.image.create, request)
    +            upload_response = await self._run_blocking(self._client.im.v1.image.create, request)
                 image_key = self._extract_response_field(upload_response, "image_key")
                 if not image_key:
                     return self._response_error_result(
    @@ -2244,7 +2304,7 @@ async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
     
             try:
                 request = self._build_get_chat_request(chat_id)
    -            response = await asyncio.to_thread(self._client.im.v1.chat.get, request)
    +            response = await self._run_blocking(self._client.im.v1.chat.get, request)
                 if not response or getattr(response, "success", lambda: False)() is False:
                     code = getattr(response, "code", "unknown")
                     msg = getattr(response, "msg", "chat lookup failed")
    @@ -2789,7 +2849,7 @@ async def _handle_reaction_event(self, event_type: str, data: Any) -> None:
             # Fetch the target message to verify it was sent by us and to obtain chat context.
             try:
                 request = self._build_get_message_request(message_id)
    -            response = await asyncio.to_thread(self._client.im.v1.message.get, request)
    +            response = await self._run_blocking(self._client.im.v1.message.get, request)
                 if not response or not getattr(response, "success", lambda: False)():
                     return
                 items = getattr(getattr(response, "data", None), "items", None) or []
    @@ -2967,7 +3027,7 @@ async def _add_reaction(self, message_id: str, emoji_type: str) -> Optional[str]
                     .request_body(body)
                     .build()
                 )
    -            response = await asyncio.to_thread(self._client.im.v1.message_reaction.create, request)
    +            response = await self._run_blocking(self._client.im.v1.message_reaction.create, request)
                 if response and getattr(response, "success", lambda: False)():
                     data = getattr(response, "data", None)
                     return getattr(data, "reaction_id", None)
    @@ -2998,7 +3058,7 @@ async def _remove_reaction(self, message_id: str, reaction_id: str) -> bool:
                     .reaction_id(reaction_id)
                     .build()
                 )
    -            response = await asyncio.to_thread(self._client.im.v1.message_reaction.delete, request)
    +            response = await self._run_blocking(self._client.im.v1.message_reaction.delete, request)
                 if response and getattr(response, "success", lambda: False)():
                     return True
                 logger.debug(
    @@ -3713,7 +3773,7 @@ async def _download_feishu_image(self, *, message_id: str, image_key: str) -> tu
                     file_key=image_key,
                     resource_type="image",
                 )
    -            response = await asyncio.to_thread(self._client.im.v1.message_resource.get, request)
    +            response = await self._run_blocking(self._client.im.v1.message_resource.get, request)
                 if not response or not response.success():
                     logger.warning(
                         "[Feishu] Failed to download image %s: %s %s",
    @@ -3757,7 +3817,7 @@ async def _download_feishu_message_resource(
                         file_key=file_key,
                         resource_type=request_type,
                     )
    -                response = await asyncio.to_thread(self._client.im.v1.message_resource.get, request)
    +                response = await self._run_blocking(self._client.im.v1.message_resource.get, request)
                     if not response or not response.success():
                         logger.debug(
                             "[Feishu] Resource download failed for %s/%s via type=%s: %s %s",
    @@ -3975,7 +4035,7 @@ async def _resolve_sender_name_from_api(
                 else:
                     id_type = "user_id"
                 request = GetUserRequest.builder().user_id(trimmed).user_id_type(id_type).build()
    -            response = await asyncio.to_thread(self._client.contact.v3.user.get, request)
    +            response = await self._run_blocking(self._client.contact.v3.user.get, request)
                 if not response or not response.success():
                     return None
                 user = getattr(getattr(response, "data", None), "user", None)
    @@ -4006,7 +4066,7 @@ async def _fetch_bot_names(self, bot_ids: List[str]) -> Optional[Dict[str, str]]
                     .token_types({AccessTokenType.TENANT})
                     .build()
                 )
    -            resp = await asyncio.to_thread(self._client.request, req)
    +            resp = await self._run_blocking(self._client.request, req)
                 content = getattr(getattr(resp, "raw", None), "content", None)
                 if not content:
                     return None
    @@ -4031,7 +4091,7 @@ async def _fetch_message_text(self, message_id: str) -> Optional[str]:
                 return self._message_text_cache[message_id]
             try:
                 request = self._build_get_message_request(message_id)
    -            response = await asyncio.to_thread(self._client.im.v1.message.get, request)
    +            response = await self._run_blocking(self._client.im.v1.message.get, request)
                 if not response or getattr(response, "success", lambda: False)() is False:
                     code = getattr(response, "code", "unknown")
                     msg = getattr(response, "msg", "message lookup failed")
    @@ -4255,7 +4315,7 @@ async def _hydrate_bot_identity(self) -> None:
                     .token_types({AccessTokenType.TENANT})
                     .build()
                 )
    -            resp = await asyncio.to_thread(self._client.request, req)
    +            resp = await self._run_blocking(self._client.request, req)
                 content = getattr(getattr(resp, "raw", None), "content", None)
                 if content:
                     payload = json.loads(content)
    @@ -4287,7 +4347,7 @@ async def _hydrate_bot_identity(self) -> None:
                 return
             try:
                 request = self._build_get_application_request(app_id=self._app_id, lang="en_us")
    -            response = await asyncio.to_thread(self._client.application.v6.application.get, request)
    +            response = await self._run_blocking(self._client.application.v6.application.get, request)
                 if not response or not response.success():
                     code = getattr(response, "code", None)
                     if code == 99991672:
    @@ -4415,7 +4475,7 @@ async def _send_uploaded_file_message(
                         file=file_obj,
                     )
                     request = self._build_file_upload_request(body)
    -                upload_response = await asyncio.to_thread(self._client.im.v1.file.create, request)
    +                upload_response = await self._run_blocking(self._client.im.v1.file.create, request)
                 file_key = self._extract_response_field(upload_response, "file_key")
                 if not file_key:
                     return self._response_error_result(
    @@ -4471,7 +4531,7 @@ async def _send_raw_message(
                     uuid_value=str(uuid.uuid4()),
                 )
                 request = self._build_reply_message_request(effective_reply_to, body)
    -            return await asyncio.to_thread(self._client.im.v1.message.reply, request)
    +            return await self._run_blocking(self._client.im.v1.message.reply, request)
     
             # For topic/thread messages that fell back from reply→create, use
             # thread_id as receive_id so the message lands in the topic instead of
    @@ -4501,7 +4561,7 @@ async def _send_raw_message(
                     uuid_value=str(uuid.uuid4()),
                 )
                 request = self._build_create_message_request(receive_id_type, body)
    -        return await asyncio.to_thread(self._client.im.v1.message.create, request)
    +        return await self._run_blocking(self._client.im.v1.message.create, request)
     
         @staticmethod
         def _response_succeeded(response: Any) -> bool:

  • tests/gateway/test_feishu_sdk_executor.py

    @@ -0,0 +1,123 @@
    +"""Regression tests for the Feishu adapter's owned SDK executor.
    +
    +Blocking Feishu SDK calls used to run on asyncio's shared default executor.
    +When that executor was torn down (agent thread exit / loop cleanup), every
    +subsequent send failed permanently with "Executor shutdown has been called"
    +and the gateway became a zombie. The adapter now owns its own
    +ThreadPoolExecutor and recreates it on demand if it has been shut down.
    +
    +Covers: #10849
    +"""
    +import concurrent.futures
    +
    +import pytest
    +
    +from plugins.platforms.feishu.adapter import FeishuAdapter
    +
    +
    +def _bare_adapter() -> FeishuAdapter:
    +    """A FeishuAdapter with only the executor fields wired (no __init__)."""
    +    adapter = object.__new__(FeishuAdapter)
    +    import threading
    +
    +    adapter._sdk_executor_lock = threading.Lock()
    +    adapter._sdk_executor = None
    +    adapter._sdk_executor_closing = False
    +    return adapter
    +
    +
    +def test_get_executor_creates_pool():
    +    adapter = _bare_adapter()
    +    executor = adapter._get_sdk_executor()
    +    assert isinstance(executor, concurrent.futures.ThreadPoolExecutor)
    +    # Same instance returned while alive.
    +    assert adapter._get_sdk_executor() is executor
    +    adapter._shutdown_sdk_executor()
    +
    +
    +def test_get_executor_recreates_after_shutdown():
    +    """A shut-down pool must be transparently replaced — the #10849 recovery."""
    +    adapter = _bare_adapter()
    +    first = adapter._get_sdk_executor()
    +    first.shutdown(wait=True)
    +    assert getattr(first, "_shutdown", False) is True
    +
    +    second = adapter._get_sdk_executor()
    +    assert second is not first
    +    assert getattr(second, "_shutdown", False) is False
    +    adapter._shutdown_sdk_executor()
    +
    +
    +def test_shutdown_clears_reference():
    +    adapter = _bare_adapter()
    +    adapter._get_sdk_executor()
    +    adapter._shutdown_sdk_executor()
    +    assert adapter._sdk_executor is None
    +    # Idempotent.
    +    adapter._shutdown_sdk_executor()
    +
    +
    [redacted:email]
    +async def test_run_blocking_executes_on_owned_pool():
    +    adapter = _bare_adapter()
    +    captured = {}
    +
    +    def _work(value):
    +        import threading
    +
    +        captured["thread"] = threading.current_thread().name
    +        return value * 2
    +
    +    result = await adapter._run_blocking(_work, 21)
    +    assert result == 42
    +    # Ran on the adapter-owned pool, not the default executor.
    +    assert captured["thread"].startswith("hermes-feishu-sdk")
    +    adapter._shutdown_sdk_executor()
    +
    +
    [redacted:email]
    +async def test_run_blocking_survives_pool_shutdown():
    +    """After the pool is shut down, _run_blocking transparently recovers."""
    +    adapter = _bare_adapter()
    +    assert await adapter._run_blocking(lambda: "first") == "first"
    +
    +    adapter._shutdown_sdk_executor()
    +
    +    # _shutdown set the closing flag, so this would now refuse — re-arm first
    +    # the way a reconnect does, then the next call rebuilds the pool.
    +    adapter._sdk_executor_closing = False
    +    assert await adapter._run_blocking(lambda: "second") == "second"
    +    adapter._shutdown_sdk_executor()
    +
    +
    +def test_closing_flag_refuses_resurrection():
    +    """A real disconnect/shutdown must NOT be resurrected by the recreate path."""
    +    adapter = _bare_adapter()
    +    adapter._get_sdk_executor()  # build a live pool
    +    adapter._shutdown_sdk_executor()  # real teardown sets _closing
    +
    +    assert adapter._sdk_executor_closing is True
    +    with pytest.raises(RuntimeError, match="shutting down"):
    +        adapter._get_sdk_executor()
    +
    +
    [redacted:email]
    +async def test_reconnect_rearms_executor():
    +    """connect() clears the closing flag so a reconnect can use the pool again."""
    +    import threading
    +
    +    adapter = object.__new__(FeishuAdapter)
    +    adapter._sdk_executor_lock = threading.Lock()
    +    adapter._sdk_executor = None
    +    adapter._sdk_executor_closing = True  # as if a prior disconnect ran
    +
    +    # connect() bails early (no creds) but must still re-arm the executor.
    +    adapter._app_id = ""
    +    adapter._app_secret = ""
    +    ok = await adapter.connect()
    +    assert ok is False  # bailed on missing creds
    +    assert adapter._sdk_executor_closing is False
    +    # And now the executor is usable again.
    +    assert await adapter._run_blocking(lambda: "rearmed") == "rearmed"
    +    adapter._shutdown_sdk_executor()
    +

  • tests/gateway/test_session_env.py

    @@ -317,6 +317,7 @@ async def test_run_in_executor_with_context_preserves_session_env(monkeypatch):
             )
         finally:
             runner._clear_session_env(tokens)
    +        runner._shutdown_executor()
     
         assert result == {
             "platform": "telegram",
    @@ -334,7 +335,10 @@ async def test_run_in_executor_with_context_forwards_args():
         def add(a, b):
             return a + b
     
    -    result = await runner._run_in_executor_with_context(add, 3, 7)
    +    try:
    +        result = await runner._run_in_executor_with_context(add, 3, 7)
    +    finally:
    +        runner._shutdown_executor()
         assert result == 10
     
     
    @@ -346,5 +350,47 @@ async def test_run_in_executor_with_context_propagates_exceptions():
         def blow_up():
             raise ValueError("boom")
     
    -    with pytest.raises(ValueError, match="boom"):
    -        await runner._run_in_executor_with_context(blow_up)
    +    try:
    +        with pytest.raises(ValueError, match="boom"):
    +            await runner._run_in_executor_with_context(blow_up)
    +    finally:
    +        runner._shutdown_executor()
    +
    +
    [redacted:email]
    +async def test_run_in_executor_with_context_survives_default_executor_shutdown():
    +    """Gateway agent work should not depend on asyncio's default executor."""
    +    runner = object.__new__(GatewayRunner)
    +    loop = asyncio.get_running_loop()
    +
    +    await loop.run_in_executor(None, lambda: None)
    +    await loop.shutdown_default_executor()
    +
    +    try:
    +        result = await runner._run_in_executor_with_context(lambda: "ok")
    +    finally:
    +        runner._shutdown_executor()
    +
    +    assert result == "ok"
    +
    +
    [redacted:email]
    +async def test_gateway_executor_refuses_resurrection_after_shutdown():
    +    """A real gateway shutdown must NOT be resurrected by the recreate path.
    +
    +    _shutdown_executor() means "we're stopping" — the recreate-on-shutdown
    +    logic exists to survive an *external* teardown of the loop default
    +    (test_..._survives_default_executor_shutdown), not to undo our own stop.
    +    """
    +    runner = object.__new__(GatewayRunner)
    +
    +    try:
    +        first = await runner._run_in_executor_with_context(lambda: "first")
    +        assert first == "first"
    +        runner._shutdown_executor()
    +
    +        with pytest.raises(RuntimeError, match="shutting down"):
    +            await runner._run_in_executor_with_context(lambda: "second")
    +    finally:
    +        runner._shutdown_executor()
    +

  • tests/gateway/test_stream_consumer_thread_routing.py

    @@ -197,6 +197,12 @@ async def test_create_uses_thread_id_when_available(self):
             adapter._client = mock_client
             adapter._build_create_message_body = FeishuAdapter._build_create_message_body
             adapter._build_create_message_request = FeishuAdapter._build_create_message_request
    +        # _send_raw_message routes blocking SDK calls through _run_blocking
    +        # (adapter-owned executor). On a MagicMock(spec=...) that method is
    +        # auto-mocked and would swallow the real call, so wire a passthrough.
    +        async def _run_blocking_passthrough(func, *args):
    +            return func(*args)
    +        adapter._run_blocking = _run_blocking_passthrough
     
             # Call _send_raw_message with reply_to=None and thread_id in metadata
             import json
    @@ -250,6 +256,9 @@ async def test_create_uses_chat_id_when_no_thread(self):
             adapter._client = mock_client
             adapter._build_create_message_body = FeishuAdapter._build_create_message_body
             adapter._build_create_message_request = FeishuAdapter._build_create_message_request
    +        async def _run_blocking_passthrough(func, *args):
    +            return func(*args)
    +        adapter._run_blocking = _run_blocking_passthrough
     
             import json
             result = await FeishuAdapter._send_raw_message(

• #100967 fix(guardrails): tool-call loops hard-stop on unattended platforms, identical replays halt for any tool (#49075, #89069; salvage #49189, #47231)

  merged to main

  files: agent/agent_init.py, agent/tool_guardrails.py, cli-config.yaml.example, hermes_cli/config_defaults.py, run_agent.py, tests/agent/test_tool_guardrails.py, tests/run_agent/test_tool_call_guardrail_runtime.py, website/docs/user-guide/configuration.md, website/docs/user-guide/docker.md