diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py
index 01ea45d7be24..df270cb426fe 100644
--- a/agent/auxiliary_client.py
+++ b/agent/auxiliary_client.py
@@ -2600,22 +2600,13 @@ def _is_model_not_found_error(exc: Exception) -> bool:
def _evict_cached_clients(provider: str) -> None:
"""Drop cached auxiliary clients for a provider so fresh creds are used."""
normalized = _normalize_aux_provider(provider)
- with _client_cache_lock:
- stale_keys = [
- key for key in _client_cache
- if _normalize_aux_provider(str(key[0])) == normalized
- ]
- for key in stale_keys:
- client = _client_cache.get(key, (None, None, None))[0]
- if client is not None:
- _force_close_async_httpx(client)
- try:
- close_fn = getattr(client, "close", None)
- if callable(close_fn):
- close_fn()
- except Exception:
- pass
- _client_cache.pop(key, None)
+ removed_clients, remaining_owner_ids = _pop_cached_clients(
+ lambda key, _entry: _normalize_aux_provider(str(key[0])) == normalized
+ )
+ _dispose_removed_cached_clients(
+ removed_clients,
+ remaining_owner_ids=remaining_owner_ids,
+ )
def _evict_cached_client_instance(target: Any) -> bool:
@@ -2637,20 +2628,127 @@ def _evict_cached_client_instance(target: Any) -> bool:
"""
if target is None:
return False
- evicted = False
+ owner = _cached_client_owner(target)
+ removed_clients, remaining_owner_ids = _pop_cached_clients(
+ lambda _key, entry: (
+ bool(entry)
+ and (
+ entry[0] is target
+ or _cached_client_owner(entry[0]) is owner
+ )
+ )
+ )
+ _dispose_removed_cached_clients(
+ removed_clients,
+ remaining_owner_ids=remaining_owner_ids,
+ )
+ return bool(removed_clients)
+
+
+def _explicit_real_client(client: Any) -> Any:
+ """Return ``client._real_client`` only when it is explicitly stored.
+
+ ``MagicMock`` auto-creates missing attributes on access, so plain
+ ``getattr(client, "_real_client", None)`` misclassifies direct mock clients
+ as wrappers. We only follow ``_real_client`` when it is explicitly present
+ on the instance (or declared in slots).
+ """
+ if client is None:
+ return None
+ try:
+ data = vars(client)
+ except TypeError:
+ data = None
+ if isinstance(data, dict) and "_real_client" in data:
+ return data.get("_real_client")
+ slots = getattr(type(client), "__slots__", ())
+ if isinstance(slots, str):
+ slots = (slots,)
+ if "_real_client" in slots:
+ try:
+ return object.__getattribute__(client, "_real_client")
+ except AttributeError:
+ return None
+ return None
+
+
+def _cached_client_owner(client: Any) -> Any:
+ """Return the underlying owner responsible for transport lifetime."""
+ cur = client
+ seen: set[int] = set()
+ while cur is not None:
+ ident = id(cur)
+ if ident in seen:
+ break
+ seen.add(ident)
+ real = _explicit_real_client(cur)
+ if real is None:
+ return cur
+ cur = real
+ return client
+
+
+def _cached_owner_ids_unlocked() -> set[int]:
+ owner_ids: set[int] = set()
+ for entry in _client_cache.values():
+ if not entry:
+ continue
+ owner = _cached_client_owner(entry[0])
+ if owner is not None:
+ owner_ids.add(id(owner))
+ return owner_ids
+
+
+def _dispose_cached_client_owner(owner: Any) -> None:
+ """Release transports for a removed cached client owner."""
+ if owner is None:
+ return
+ _force_close_async_httpx(owner)
+ try:
+ import inspect
+
+ close_fn = getattr(owner, "close", None)
+ if callable(close_fn) and not inspect.iscoroutinefunction(close_fn):
+ close_fn()
+ except Exception:
+ pass
+
+
+def _dispose_removed_cached_clients(
+ removed_clients: list[Any],
+ *,
+ remaining_owner_ids: set[int],
+) -> None:
+ """Dispose removed cached clients once no surviving cache entry owns them."""
+ disposed_owner_ids: set[int] = set()
+ for client in removed_clients:
+ owner = _cached_client_owner(client)
+ if owner is None:
+ continue
+ owner_id = id(owner)
+ if owner_id in disposed_owner_ids:
+ continue
+ if owner_id in remaining_owner_ids:
+ continue
+ disposed_owner_ids.add(owner_id)
+ _dispose_cached_client_owner(owner)
+
+
+def _pop_cached_clients(match) -> tuple[list[Any], set[int]]:
+ """Remove matching cached clients and return removed owners + survivors."""
+ removed_clients: list[Any] = []
with _client_cache_lock:
- for key in list(_client_cache.keys()):
- entry = _client_cache.get(key)
- if entry is None:
- continue
- cached = entry[0]
- if cached is None:
- continue
- real = getattr(cached, "_real_client", None)
- if cached is target or real is target:
- del _client_cache[key]
- evicted = True
- return evicted
+ removed_keys: list[tuple] = []
+ for key, entry in list(_client_cache.items()):
+ if match(key, entry):
+ removed_keys.append(key)
+ client = entry[0] if entry else None
+ if client is not None:
+ removed_clients.append(client)
+ for key in removed_keys:
+ _client_cache.pop(key, None)
+ remaining_owner_ids = _cached_owner_ids_unlocked()
+ return removed_clients, remaining_owner_ids
def _pool_cache_hint(
@@ -4365,17 +4463,17 @@ def _client_cache_key(
def _store_cached_client(cache_key: tuple, client: Any, default_model: Optional[str], *, bound_loop: Any = None) -> None:
+ removed_clients: list[Any] = []
with _client_cache_lock:
old_entry = _client_cache.get(cache_key)
if old_entry is not None and old_entry[0] is not client:
- _force_close_async_httpx(old_entry[0])
- try:
- close_fn = getattr(old_entry[0], "close", None)
- if callable(close_fn):
- close_fn()
- except Exception:
- pass
+ removed_clients.append(old_entry[0])
_client_cache[cache_key] = (client, default_model, bound_loop)
+ remaining_owner_ids = _cached_owner_ids_unlocked()
+ _dispose_removed_cached_clients(
+ removed_clients,
+ remaining_owner_ids=remaining_owner_ids,
+ )
def _refresh_nous_auxiliary_client(
@@ -4481,25 +4579,13 @@ def shutdown_cached_clients() -> None:
Call this during CLI shutdown, *before* the event loop is closed, to
avoid ``AsyncHttpxClientWrapper.__del__`` raising on a dead loop.
"""
- import inspect
-
- with _client_cache_lock:
- for key, entry in list(_client_cache.items()):
- client = entry[0]
- if client is None:
- continue
- # Mark any async httpx transport as closed first (prevents __del__
- # from scheduling aclose() on a dead event loop).
- _force_close_async_httpx(client)
- # Sync clients: close the httpx connection pool cleanly.
- # Async clients: skip — we already neutered __del__ above.
- try:
- close_fn = getattr(client, "close", None)
- if close_fn and not inspect.iscoroutinefunction(close_fn):
- close_fn()
- except Exception:
- pass
- _client_cache.clear()
+ removed_clients, remaining_owner_ids = _pop_cached_clients(
+ lambda _key, _client: True
+ )
+ _dispose_removed_cached_clients(
+ removed_clients,
+ remaining_owner_ids=remaining_owner_ids,
+ )
def cleanup_stale_async_clients() -> None:
@@ -4510,15 +4596,17 @@ def cleanup_stale_async_clients() -> None:
This is defense-in-depth — the primary fix is ``neuter_async_httpx_del``
which disables ``__del__`` entirely.
"""
- with _client_cache_lock:
- stale_keys = []
- for key, entry in _client_cache.items():
- client, _default, cached_loop = entry
- if cached_loop is not None and cached_loop.is_closed():
- _force_close_async_httpx(client)
- stale_keys.append(key)
- for key in stale_keys:
- del _client_cache[key]
+ removed_clients, remaining_owner_ids = _pop_cached_clients(
+ lambda _key, entry: (
+ bool(entry)
+ and entry[2] is not None
+ and entry[2].is_closed()
+ )
+ )
+ _dispose_removed_cached_clients(
+ removed_clients,
+ remaining_owner_ids=remaining_owner_ids,
+ )
def _is_openrouter_client(client: Any) -> bool:
@@ -4592,6 +4680,8 @@ def _get_cached_client(
main_runtime=main_runtime,
is_vision=is_vision,
)
+ removed_clients: list[Any] = []
+ remaining_owner_ids: set[int] = set()
with _client_cache_lock:
if cache_key in _client_cache:
cached_client, cached_default, cached_loop = _client_cache[cache_key]
@@ -4608,11 +4698,18 @@ def _get_cached_client(
effective = _compat_model(cached_client, model, cached_default)
return cached_client, effective
# Stale — evict and fall through to create a new client.
- _force_close_async_httpx(cached_client)
- del _client_cache[cache_key]
+ removed_entry = _client_cache.pop(cache_key, None)
+ if removed_entry and removed_entry[0] is not None:
+ removed_clients.append(removed_entry[0])
+ remaining_owner_ids = _cached_owner_ids_unlocked()
else:
effective = _compat_model(cached_client, model, cached_default)
return cached_client, effective
+ if removed_clients:
+ _dispose_removed_cached_clients(
+ removed_clients,
+ remaining_owner_ids=remaining_owner_ids,
+ )
# Build outside the lock.
# For pool-backed api_key providers, derive the active API key from the
# pool entry rather than from env vars. resolve_api_key_provider_credentials
@@ -4640,17 +4737,28 @@ def _get_cached_client(
# For async clients, remember which loop they were created on so we
# can detect stale entries later.
bound_loop = current_loop
+ removed_clients = []
+ remaining_owner_ids = set()
with _client_cache_lock:
if cache_key not in _client_cache:
# Safety belt: if the cache has grown beyond the max, evict
# the oldest entries (FIFO — dict preserves insertion order).
while len(_client_cache) >= _CLIENT_CACHE_MAX_SIZE:
evict_key, evict_entry = next(iter(_client_cache.items()))
- _force_close_async_httpx(evict_entry[0])
- del _client_cache[evict_key]
+ _client_cache.pop(evict_key, None)
+ if evict_entry and evict_entry[0] is not None:
+ removed_clients.append(evict_entry[0])
_client_cache[cache_key] = (client, default_model, bound_loop)
+ remaining_owner_ids = _cached_owner_ids_unlocked()
else:
+ removed_clients.append(client)
client, default_model, _ = _client_cache[cache_key]
+ remaining_owner_ids = _cached_owner_ids_unlocked()
+ if removed_clients:
+ _dispose_removed_cached_clients(
+ removed_clients,
+ remaining_owner_ids=remaining_owner_ids,
+ )
return client, model or default_model
diff --git a/gateway/platforms/feishu.py b/gateway/platforms/feishu.py
index 4814107bacd2..43457603a612 100644
--- a/gateway/platforms/feishu.py
+++ b/gateway/platforms/feishu.py
@@ -1696,10 +1696,11 @@ async def disconnect(self) -> None:
await self._cancel_pending_tasks(self._pending_text_batch_tasks)
await self._cancel_pending_tasks(self._pending_media_batch_tasks)
self._reset_batch_buffers()
- self._disable_websocket_auto_reconnect()
+ ws_client = self._disable_websocket_auto_reconnect()
await self._stop_webhook_server()
ws_thread_loop = self._ws_thread_loop
+ await self._close_websocket_client(ws_client, ws_thread_loop)
if ws_thread_loop is not None and not ws_thread_loop.is_closed():
logger.debug("[Feishu] Cancelling websocket thread tasks and stopping loop")
@@ -1749,14 +1750,44 @@ def _reset_batch_buffers(self) -> None:
self._pending_media_batches.clear()
def _disable_websocket_auto_reconnect(self) -> None:
- if self._ws_client is None:
- return
+ ws_client = self._ws_client
+ if ws_client is None:
+ return None
try:
- setattr(self._ws_client, "_auto_reconnect", False)
+ setattr(ws_client, "_auto_reconnect", False)
except Exception:
pass
finally:
self._ws_client = None
+ return ws_client
+
+ async def _close_websocket_client(
+ self,
+ ws_client: Optional[Any],
+ ws_thread_loop: Optional[asyncio.AbstractEventLoop],
+ ) -> None:
+ if ws_client is None:
+ return
+ disconnect = getattr(ws_client, "_disconnect", None)
+ if not callable(disconnect):
+ return
+ if ws_thread_loop is not None and not ws_thread_loop.is_closed():
+ coro = disconnect()
+ try:
+ future = asyncio.run_coroutine_threadsafe(coro, ws_thread_loop)
+ except Exception:
+ coro.close()
+ logger.debug("[Feishu] Failed to submit websocket close to thread loop", exc_info=True)
+ return
+ try:
+ await asyncio.wait_for(asyncio.wrap_future(future), timeout=5.0)
+ except Exception:
+ logger.debug("[Feishu] Failed to close websocket client cleanly", exc_info=True)
+ return
+ try:
+ await disconnect()
+ except Exception:
+ logger.debug("[Feishu] Failed to close websocket client without thread loop", exc_info=True)
async def _stop_webhook_server(self) -> None:
if self._webhook_runner is None:
diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py
index 6516c165401b..ae23d0ec574f 100644
--- a/gateway/platforms/telegram.py
+++ b/gateway/platforms/telegram.py
@@ -2206,6 +2206,41 @@ def _should_thread_reply(self, reply_to: Optional[str], chunk_index: int) -> boo
else: # "first" (default)
return chunk_index == 0
+ async def _send_via_standalone_fallback(
+ self,
+ chat_id: str,
+ content: str,
+ metadata: Optional[Dict[str, Any]] = None,
+ ) -> SendResult:
+ """Bypass PTB's live Bot client when its send path is gated unhealthy."""
+ try:
+ from tools.send_message_tool import _send_telegram
+
+ result = await _send_telegram(
+ self.config.token,
+ chat_id,
+ content,
+ thread_id=self._metadata_thread_id(metadata),
+ disable_link_previews=self._disable_link_previews,
+ )
+ except Exception as exc:
+ logger.warning("[%s] Standalone Telegram fallback raised: %s", self.name, exc)
+ return SendResult(success=False, error="send_path_degraded", retryable=True)
+
+ if isinstance(result, dict) and result.get("success"):
+ return SendResult(
+ success=True,
+ message_id=str(result.get("message_id")) if result.get("message_id") else None,
+ raw_response=result,
+ )
+ error = result.get("error") if isinstance(result, dict) else result
+ return SendResult(
+ success=False,
+ error=f"send_path_degraded: standalone fallback failed: {error}",
+ retryable=True,
+ raw_response=result,
+ )
+
async def send(
self,
chat_id: str,
@@ -2219,7 +2254,7 @@ async def send(
# getattr() — tests build adapters via object.__new__() (no __init__).
if getattr(self, "_send_path_degraded", False):
- return SendResult(success=False, error="send_path_degraded", retryable=True)
+ return await self._send_via_standalone_fallback(chat_id, content, metadata)
# Skip whitespace-only text to prevent Telegram 400 empty-text errors.
if not content or not content.strip():
diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py
index d5b76574a120..7684522dc463 100644
--- a/hermes_cli/gateway.py
+++ b/hermes_cli/gateway.py
@@ -48,6 +48,9 @@
logger = logging.getLogger(__name__)
+_LAUNCHD_MAXFILES_SOFT = 4096
+_LAUNCHD_MAXFILES_HARD = 8192
+
# =============================================================================
# Process Management (for manual gateway runs)
# =============================================================================
@@ -3426,6 +3429,18 @@ def generate_launchd_plist() -> str:
KeepAlive
+
+ SoftResourceLimits
+
+ NumberOfFiles
+ {_LAUNCHD_MAXFILES_SOFT}
+
+
+ HardResourceLimits
+
+ NumberOfFiles
+ {_LAUNCHD_MAXFILES_HARD}
+
StandardOutPath
{log_dir}/gateway.log
@@ -3449,7 +3464,6 @@ def launchd_plist_is_current() -> bool:
installed
) == _normalize_launchd_plist_for_comparison(expected)
-
def refresh_launchd_plist_if_needed() -> bool:
"""Rewrite the installed launchd plist when the generated definition has changed.
diff --git a/tests/agent/test_auxiliary_client.py b/tests/agent/test_auxiliary_client.py
index 7770b2e8c887..b92187d98cbc 100644
--- a/tests/agent/test_auxiliary_client.py
+++ b/tests/agent/test_auxiliary_client.py
@@ -3250,6 +3250,23 @@ def test_evict_cached_client_instance_drops_direct_match(self):
with _client_cache_lock:
_client_cache.clear()
+ def test_evict_cached_client_instance_closes_direct_match(self):
+ """Eviction must close the stored client, not just drop the dict entry."""
+ from agent.auxiliary_client import (
+ _client_cache, _client_cache_lock, _evict_cached_client_instance,
+ )
+
+ target = MagicMock(name="target_client")
+ with _client_cache_lock:
+ _client_cache.clear()
+ _client_cache[("openrouter", False, None, None, None)] = (target, "x", None)
+ try:
+ assert _evict_cached_client_instance(target) is True
+ target.close.assert_called_once()
+ finally:
+ with _client_cache_lock:
+ _client_cache.clear()
+
def test_evict_cached_client_instance_walks_codex_wrapper(self):
"""Closing the underlying OpenAI client must evict the Codex shim."""
from agent.auxiliary_client import (
@@ -3314,6 +3331,50 @@ def test_evict_cached_client_instance_walks_async_wrapper(self):
with _client_cache_lock:
_client_cache.clear()
+ def test_evict_cached_client_instance_closes_async_wrapper_real_client(self):
+ """Evicting an async shim must close its underlying sync real client.
+
+ Async auxiliary wrappers do not expose ``close()`` themselves, but they
+ retain ``_real_client`` specifically so cache-eviction code can reach
+ the actual OpenAI/httpx owner. Dropping only the wrapper leaks the real
+ client's pool FDs until process exit.
+ """
+ from agent.auxiliary_client import (
+ _client_cache, _client_cache_lock, _evict_cached_client_instance,
+ CodexAuxiliaryClient, AsyncCodexAuxiliaryClient,
+ )
+
+ close_calls = {"count": 0}
+
+ def _close():
+ close_calls["count"] += 1
+
+ real = SimpleNamespace(
+ api_key="k",
+ base_url="https://chatgpt.com/backend-api/codex",
+ responses=SimpleNamespace(stream=lambda **k: None),
+ close=_close,
+ )
+ sync_wrapper = CodexAuxiliaryClient(real, "gpt-5.5")
+ async_wrapper = AsyncCodexAuxiliaryClient(sync_wrapper)
+ with _client_cache_lock:
+ _client_cache.clear()
+ _client_cache[("openai-codex", True, None, None, None)] = (
+ async_wrapper,
+ "gpt-5.5",
+ None,
+ )
+ try:
+ assert _evict_cached_client_instance(async_wrapper) is True
+ assert close_calls["count"] == 1, (
+ "evicting an async wrapper must close its underlying real client; "
+ "otherwise the wrapper disappears but the real httpx/OpenAI pool "
+ "keeps its FDs open"
+ )
+ finally:
+ with _client_cache_lock:
+ _client_cache.clear()
+
def test_codex_timeout_evicts_cached_wrapper(self):
"""The timeout closer evicts the cache entry that wraps the closed client."""
from agent.auxiliary_client import (
diff --git a/tests/gateway/test_feishu.py b/tests/gateway/test_feishu.py
index 4d78b454b0ca..6acc6ab482c6 100644
--- a/tests/gateway/test_feishu.py
+++ b/tests/gateway/test_feishu.py
@@ -1,6 +1,7 @@
"""Tests for the Feishu gateway integration."""
import asyncio
+import concurrent.futures
import json
import os
import tempfile
@@ -252,6 +253,56 @@ def is_closed(self):
)
release_lock.assert_called_once_with("feishu-app-id", "cli_app")
+ @patch.dict(os.environ, {
+ "FEISHU_APP_ID": "cli_app",
+ "FEISHU_APP_SECRET": "secret_app",
+ }, clear=True)
+ def test_disconnect_closes_live_websocket_client_before_stopping_thread_loop(self):
+ from gateway.config import PlatformConfig
+ from gateway.platforms.feishu import FeishuAdapter
+
+ adapter = FeishuAdapter(PlatformConfig())
+ ws_client = SimpleNamespace(_disconnect=AsyncMock(), _auto_reconnect=True)
+
+ class _ThreadLoop:
+ def __init__(self):
+ self.stopped = False
+
+ def is_closed(self):
+ return False
+
+ def call_soon_threadsafe(self, callback):
+ callback()
+
+ def call_later(self, _delay, callback):
+ callback()
+
+ def stop(self):
+ self.stopped = True
+
+ def _submit(coro, _loop):
+ coro.close()
+ future = concurrent.futures.Future()
+ future.set_result(None)
+ return future
+
+ adapter._ws_client = ws_client
+ adapter._ws_thread_loop = _ThreadLoop()
+ adapter._persist_seen_message_ids = Mock()
+ adapter._release_app_lock = AsyncMock()
+
+ with (
+ patch("gateway.platforms.feishu.asyncio.all_tasks", return_value=[]),
+ patch(
+ "gateway.platforms.feishu.asyncio.run_coroutine_threadsafe",
+ side_effect=_submit,
+ ) as submit,
+ ):
+ asyncio.run(adapter.disconnect())
+
+ ws_client._disconnect.assert_called_once_with()
+ submit.assert_called_once()
+
@patch.dict(os.environ, {
"FEISHU_APP_ID": "cli_app",
"FEISHU_APP_SECRET": "secret_app",
diff --git a/tests/gateway/test_telegram_send_path_health.py b/tests/gateway/test_telegram_send_path_health.py
index 05972bdba437..728db137b1c6 100644
--- a/tests/gateway/test_telegram_send_path_health.py
+++ b/tests/gateway/test_telegram_send_path_health.py
@@ -50,17 +50,49 @@ async def test_send_succeeds_when_path_healthy():
@pytest.mark.asyncio
-async def test_send_short_circuits_when_path_degraded():
- """Degraded adapter returns failure WITHOUT calling send_message,
- so cron's live-adapter branch falls through to standalone HTTP."""
+async def test_send_reports_retryable_failure_when_standalone_fallback_fails(monkeypatch):
+ """Degraded adapter never uses PTB; failed fallback remains retryable."""
adapter = _make_adapter()
adapter._send_path_degraded = True
+ standalone = AsyncMock(return_value={"error": "fallback failed"})
+ monkeypatch.setattr("tools.send_message_tool._send_telegram", standalone)
result = await adapter.send("123", "hello")
assert result.success is False
- assert result.error == "send_path_degraded"
+ assert result.error == "send_path_degraded: standalone fallback failed: fallback failed"
assert result.retryable is True
+ standalone.assert_awaited_once()
+ adapter._bot.send_message.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_degraded_send_path_uses_standalone_fallback(monkeypatch):
+ """Gateway replies must still deliver while the PTB send path is gated."""
+ adapter = _make_adapter()
+ adapter.config.token = "tok"
+ adapter._send_path_degraded = True
+ standalone = AsyncMock(
+ return_value={
+ "success": True,
+ "platform": "telegram",
+ "chat_id": "123",
+ "message_id": "99",
+ }
+ )
+ monkeypatch.setattr("tools.send_message_tool._send_telegram", standalone)
+
+ result = await adapter.send("123", "hello")
+
+ assert result.success is True
+ assert result.message_id == "99"
+ standalone.assert_awaited_once_with(
+ "tok",
+ "123",
+ "hello",
+ thread_id=None,
+ disable_link_previews=False,
+ )
adapter._bot.send_message.assert_not_awaited()
diff --git a/tests/hermes_cli/test_gateway_service.py b/tests/hermes_cli/test_gateway_service.py
index f9cdcc1f3134..852d15fe391a 100644
--- a/tests/hermes_cli/test_gateway_service.py
+++ b/tests/hermes_cli/test_gateway_service.py
@@ -604,7 +604,8 @@ def fake_run(cmd, check=False, **kwargs):
label = gateway_cli.get_launchd_label()
domain = gateway_cli._launchd_domain()
- assert "--replace" in plist_path.read_text(encoding="utf-8")
+ plist_text = plist_path.read_text(encoding="utf-8")
+ assert "--replace" in plist_text
# The calls list includes launchctl print probes from _launchd_domain()
# before the bootout/bootstrap calls. Filter to only bootout/bootstrap.
service_calls = [c for c in calls if "bootout" in c or "bootstrap" in c]
@@ -3045,3 +3046,23 @@ def test_launchd_plist_keepalive_unconditional(self, tmp_path, monkeypatch):
# The old conditional dict form must NOT appear
assert "SuccessfulExit" not in plist
assert "KeepAlive\n " not in plist
+
+ def test_launchd_plist_includes_number_of_files_resource_limits(self, tmp_path, monkeypatch):
+ """launchd plist must raise RLIMIT_NOFILE above the macOS default 256.
+
+ Regression: the installed launchd plist used the wrong resource-limit
+ key, so launchd ignored the requested fd ceiling and the gateway
+ inherits launchd's default soft limit of 256. Under any residual fd
+ growth, the process collapses into global ``EMFILE`` much earlier than
+ intended.
+ """
+ home = tmp_path / ".hermes"
+ home.mkdir()
+ monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: home)
+ plist = gateway_cli.generate_launchd_plist()
+
+ assert "SoftResourceLimits" in plist
+ assert "HardResourceLimits" in plist
+ assert "NumberOfFiles" in plist
+ assert "4096" in plist
+ assert "8192" in plist
diff --git a/tests/run_agent/test_async_httpx_del_neuter.py b/tests/run_agent/test_async_httpx_del_neuter.py
index 946d73dbdf11..af9f852eeb01 100644
--- a/tests/run_agent/test_async_httpx_del_neuter.py
+++ b/tests/run_agent/test_async_httpx_del_neuter.py
@@ -13,6 +13,7 @@
"""
import asyncio
+import os
from unittest.mock import MagicMock, patch
import pytest
@@ -159,6 +160,57 @@ def test_keeps_entries_without_loop(self):
with _client_cache_lock:
_client_cache.pop(key, None)
+ def test_removing_stale_async_wrapper_preserves_shared_sync_owner(self):
+ """Cleaning a stale async wrapper must not close the shared sync owner."""
+ from agent.auxiliary_client import (
+ _client_cache,
+ _client_cache_lock,
+ cleanup_stale_async_clients,
+ CodexAuxiliaryClient,
+ AsyncCodexAuxiliaryClient,
+ )
+
+ close_calls = {"count": 0}
+
+ def _close():
+ close_calls["count"] += 1
+
+ real = type(
+ "RealClient",
+ (),
+ {
+ "api_key": "k",
+ "base_url": "https://chatgpt.com/backend-api/codex",
+ "responses": MagicMock(stream=lambda **kwargs: None),
+ "close": staticmethod(_close),
+ },
+ )()
+ sync_wrapper = CodexAuxiliaryClient(real, "gpt-5.5")
+ async_wrapper = AsyncCodexAuxiliaryClient(sync_wrapper)
+
+ stale_loop = asyncio.new_event_loop()
+ stale_loop.close()
+
+ sync_key = ("openai-codex", False, "", "", "", (), False, "")
+ async_key = ("openai-codex", True, "", "", "", (), False, "")
+ with _client_cache_lock:
+ _client_cache[sync_key] = (sync_wrapper, "gpt-5.5", None)
+ _client_cache[async_key] = (async_wrapper, "gpt-5.5", stale_loop)
+
+ try:
+ cleanup_stale_async_clients()
+ with _client_cache_lock:
+ assert sync_key in _client_cache
+ assert async_key not in _client_cache
+ assert close_calls["count"] == 0, (
+ "stale async wrapper cleanup must not close the shared real client "
+ "while the sync cache entry still survives"
+ )
+ finally:
+ with _client_cache_lock:
+ _client_cache.pop(sync_key, None)
+ _client_cache.pop(async_key, None)
+
# ---------------------------------------------------------------------------
# Cache bounded growth (#10200)
@@ -286,3 +338,180 @@ def test_max_cache_size_eviction(self):
with _client_cache_lock:
_client_cache.clear()
_client_cache.update(saved)
+
+ def test_fifo_cache_eviction_closes_sync_client(self, monkeypatch):
+ """Real FIFO eviction must close the oldest sync client.
+
+ The cache-size guard in ``_get_cached_client()`` used to delete the
+ oldest entry without calling ``close()``. For sync OpenAI clients that
+ leaks the underlying httpx pool until process exit, which is exactly
+ the slow-burn FD growth pattern we are chasing in the gateway.
+ """
+ import agent.auxiliary_client as aux
+ from agent.auxiliary_client import _client_cache, _client_cache_lock
+
+ class FakeClient:
+ def __init__(self, name: str):
+ self.name = name
+ self.close_calls = 0
+
+ def close(self):
+ self.close_calls += 1
+
+ created: list[FakeClient] = []
+
+ def fake_resolve(provider, model, async_mode, **kwargs):
+ client = FakeClient(provider)
+ created.append(client)
+ return client, model or f"{provider}-model"
+
+ with _client_cache_lock:
+ saved = dict(_client_cache)
+ _client_cache.clear()
+
+ monkeypatch.setattr(aux, "_CLIENT_CACHE_MAX_SIZE", 2)
+ monkeypatch.setattr(aux, "resolve_provider_client", fake_resolve)
+
+ try:
+ aux._get_cached_client("p1", async_mode=False)
+ aux._get_cached_client("p2", async_mode=False)
+ aux._get_cached_client("p3", async_mode=False)
+
+ assert [client.name for client in created] == ["p1", "p2", "p3"]
+ assert created[0].close_calls == 1, (
+ "oldest sync client must be closed when FIFO cache eviction runs; "
+ "otherwise its httpx pool keeps holding FDs after the cache drops "
+ "the only reference"
+ )
+ assert created[1].close_calls == 0
+ assert created[2].close_calls == 0
+ finally:
+ with _client_cache_lock:
+ _client_cache.clear()
+ _client_cache.update(saved)
+
+ def test_same_key_race_loser_sync_client_is_closed(self, monkeypatch):
+ """If another builder wins the same cache key, the losing sync client must close."""
+ import agent.auxiliary_client as aux
+ from agent.auxiliary_client import _client_cache, _client_cache_lock
+
+ class FakeClient:
+ def __init__(self, name: str):
+ self.name = name
+ self.close_calls = 0
+
+ def close(self):
+ self.close_calls += 1
+
+ created: list[FakeClient] = []
+
+ def fake_resolve(provider, model, async_mode, **kwargs):
+ client = FakeClient(f"{provider}-{len(created)}")
+ created.append(client)
+ if len(created) == 2:
+ with _client_cache_lock:
+ _client_cache[cache_key] = (created[0], "winner-model", None)
+ return client, model or f"{provider}-model"
+
+ cache_key = ("p-race", False, "", "", "", (), False, "")
+ with _client_cache_lock:
+ saved = dict(_client_cache)
+ _client_cache.clear()
+
+ monkeypatch.setattr(aux, "resolve_provider_client", fake_resolve)
+
+ try:
+ first, _ = aux._get_cached_client("p-race", async_mode=False)
+ with _client_cache_lock:
+ _client_cache.clear()
+ second, _ = aux._get_cached_client("p-race", async_mode=False)
+
+ assert first is created[0]
+ assert second is created[0], "cache winner should be reused"
+ assert created[0].close_calls == 0
+ assert created[1].close_calls == 1, (
+ "same-key race loser must be closed when another client already "
+ "occupied the cache entry"
+ )
+ finally:
+ with _client_cache_lock:
+ _client_cache.clear()
+ _client_cache.update(saved)
+
+ def test_repeated_fifo_eviction_does_not_accumulate_os_fds(self, monkeypatch):
+ """Repeated auxiliary cache churn must not linearly leak real OS file descriptors."""
+ import agent.auxiliary_client as aux
+ from agent.auxiliary_client import _client_cache, _client_cache_lock
+
+ def _get_open_fd_count():
+ for fd_dir in ("/dev/fd", "/proc/self/fd"):
+ try:
+ return sum(
+ 1 for entry in os.listdir(fd_dir) if str(entry).isdigit()
+ )
+ except OSError:
+ continue
+ return None
+
+ class PipeClient:
+ def __init__(self, name: str):
+ self.name = name
+ self._read_fd, self._write_fd = os.pipe()
+ self.closed = False
+
+ def close(self):
+ if self.closed:
+ return
+ os.close(self._read_fd)
+ os.close(self._write_fd)
+ self.closed = True
+
+ created: list[PipeClient] = []
+
+ def fake_resolve(provider, model, async_mode, **kwargs):
+ client = PipeClient(provider)
+ created.append(client)
+ return client, model or f"{provider}-model"
+
+ with _client_cache_lock:
+ saved = dict(_client_cache)
+ _client_cache.clear()
+
+ baseline = _get_open_fd_count()
+ if baseline is None:
+ pytest.skip("open fd count unavailable on this platform")
+
+ monkeypatch.setattr(aux, "_CLIENT_CACHE_MAX_SIZE", 2)
+ monkeypatch.setattr(aux, "resolve_provider_client", fake_resolve)
+
+ try:
+ for idx in range(24):
+ aux._get_cached_client(f"p{idx}", async_mode=False)
+
+ mid = _get_open_fd_count()
+ assert mid is not None
+ # Two live cached clients each hold a pipe pair (4 fds total).
+ # Allow a little slack for interpreter/test harness noise, but
+ # reject the old linear-growth shape where every churned client
+ # stayed open.
+ assert mid - baseline <= 8, (
+ f"fd count grew from {baseline} to {mid} after repeated FIFO churn; "
+ "expected only the live cached clients to remain open"
+ )
+
+ aux.shutdown_cached_clients()
+ after_shutdown = _get_open_fd_count()
+ assert after_shutdown is not None
+ assert after_shutdown - baseline <= 4, (
+ f"fd count stayed elevated after cache shutdown: baseline={baseline}, "
+ f"after_shutdown={after_shutdown}"
+ )
+ finally:
+ for client in created:
+ try:
+ client.close()
+ except OSError:
+ pass
+ with _client_cache_lock:
+ _client_cache.clear()
+ _client_cache.update(saved)