common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

chore(gateway): replace deprecated asyncio.get_event_loop() with get_running_loop() (#11005)

Browse Source 
All 10 call sites in gateway/run.py and gateway/platforms/api_server.py
are inside async functions where a loop is guaranteed to be running.

get_event_loop() is deprecated since Python 3.10 — it can silently
create a new loop when none is running, masking bugs.
get_running_loop() raises RuntimeError instead, which is safer.

Surfaced during review of PRs #10533 and #10647.

Co-authored-by: kshitijk4poor <kshitijk4poor@users.noreply.github.com>
This commit is contained in:
kshitij
2026-04-16 05:13:39 -07:00
committed by GitHub
parent 0de6340a73
commit 92a78ffeee
2 changed files with 10 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
gateway/platforms/api_server.py
View File

@@ -902,7 +902,7 @@ class APIServerAdapter(BasePlatformAdapter):
return time.monotonic() return time.monotonic()
# Stream content chunks as they arrive from the agent # Stream content chunks as they arrive from the agent
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
while True: while True:
try: try:
delta = await loop.run_in_executor(None, lambda: stream_q.get(timeout=0.5)) delta = await loop.run_in_executor(None, lambda: stream_q.get(timeout=0.5))
@@ -1241,7 +1241,7 @@ class APIServerAdapter(BasePlatformAdapter):
await _emit_text_delta(it) await _emit_text_delta(it)
# Other types (non-string, non-tuple) are silently dropped. # Other types (non-string, non-tuple) are silently dropped.
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
while True: while True:
try: try:
item = await loop.run_in_executor(None, lambda: stream_q.get(timeout=0.5)) item = await loop.run_in_executor(None, lambda: stream_q.get(timeout=0.5))
@@ -2004,7 +2004,7 @@ class APIServerAdapter(BasePlatformAdapter):
callers (e.g. the SSE writer) to call ``agent.interrupt()`` from callers (e.g. the SSE writer) to call ``agent.interrupt()`` from
another thread to stop in-progress LLM calls. another thread to stop in-progress LLM calls.
""" """
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
def _run(): def _run():
agent = self._create_agent( agent = self._create_agent(

14
gateway/run.py
View File

@@ -835,7 +835,7 @@ class GatewayRunner:
session_key: Optional[str] = None, session_key: Optional[str] = None,
): ):
"""Run the sync memory flush in a thread pool so it won't block the event loop.""" """Run the sync memory flush in a thread pool so it won't block the event loop."""
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
await loop.run_in_executor( await loop.run_in_executor(
None, None,
self._flush_memories_for_session, self._flush_memories_for_session,
@@ -3777,7 +3777,7 @@ class GatewayRunner:
) )
_hyg_agent._print_fn = lambda *a, **kw: None _hyg_agent._print_fn = lambda *a, **kw: None
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
_compressed, _ = await loop.run_in_executor( _compressed, _ = await loop.run_in_executor(
None, None,
lambda: _hyg_agent._compress_context( lambda: _hyg_agent._compress_context(
@@ -6265,7 +6265,7 @@ class GatewayRunner:
if compress_start >= compress_end: if compress_start >= compress_end:
return "Nothing to compress yet (the transcript is still all protected context)." return "Nothing to compress yet (the transcript is still all protected context)."
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
compressed, _ = await loop.run_in_executor( compressed, _ = await loop.run_in_executor(
None, None,
lambda: tmp_agent._compress_context(msgs, "", approx_tokens=approx_tokens, focus_topic=focus_topic) lambda: tmp_agent._compress_context(msgs, "", approx_tokens=approx_tokens, focus_topic=focus_topic)
@@ -6650,7 +6650,7 @@ class GatewayRunner:
from hermes_state import SessionDB from hermes_state import SessionDB
from agent.insights import InsightsEngine from agent.insights import InsightsEngine
loop = _asyncio.get_event_loop() loop = _asyncio.get_running_loop()
def _run_insights(): def _run_insights():
db = SessionDB() db = SessionDB()
@@ -6667,7 +6667,7 @@ class GatewayRunner:
async def _handle_reload_mcp_command(self, event: MessageEvent) -> str: async def _handle_reload_mcp_command(self, event: MessageEvent) -> str:
"""Handle /reload-mcp command -- disconnect and reconnect all MCP servers.""" """Handle /reload-mcp command -- disconnect and reconnect all MCP servers."""
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
try: try:
from tools.mcp_tool import shutdown_mcp_servers, discover_mcp_tools, _servers, _lock from tools.mcp_tool import shutdown_mcp_servers, discover_mcp_tools, _servers, _lock
@@ -8388,7 +8388,7 @@ class GatewayRunner:
stream_consumer_holder = [None] # Mutable container for stream consumer stream_consumer_holder = [None] # Mutable container for stream consumer
# Bridge sync step_callback → async hooks.emit for agent:step events # Bridge sync step_callback → async hooks.emit for agent:step events
_loop_for_step = asyncio.get_event_loop() _loop_for_step = asyncio.get_running_loop()
_hooks_ref = self.hooks _hooks_ref = self.hooks
def _step_callback_sync(iteration: int, prev_tools: list) -> None: def _step_callback_sync(iteration: int, prev_tools: list) -> None:
@@ -9740,7 +9740,7 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
def restart_signal_handler(): def restart_signal_handler():
runner.request_restart(detached=False, via_service=True) runner.request_restart(detached=False, via_service=True)
loop = asyncio.get_event_loop() loop = asyncio.get_running_loop()
if threading.current_thread() is threading.main_thread(): if threading.current_thread() is threading.main_thread():
for sig in (signal.SIGINT, signal.SIGTERM): for sig in (signal.SIGINT, signal.SIGTERM):
try: try: