mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-27 17:58:07 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(gateway): spawn platform reconnect watcher with task-level supervision

Browse source 
Fixes #71758.

A platform adapter that dies on a transient upstream failure (marked
retryable=True, e.g. photon's sidecar exiting when its upstream
gRPC/CDN returns errors) is correctly queued into _failed_platforms
for background reconnection. But the reconnect watcher task itself
was spawned via a bare asyncio.create_task -- if an exception ever
escaped its OUTER while-loop (not just the per-platform inner
try/except), the watcher died silently: no log, no restart.

_ensure_reconnect_watcher_running() already existed to respawn a dead
watcher, but it's only called from _handle_adapter_fatal_error_impl()
when a NEW platform's fatal error arrives. If the watcher dies while a
platform is already sitting in the queue and no OTHER platform ever
fails afterward, nothing ever notices the watcher is dead -- exactly
matching the reported symptom: photon queued for reconnect, the
gateway itself healthy (other platforms kept working, so nothing
re-triggered the ensure-alive check), and the platform stayed dead for
17.5h until a manual restart, well after the transient upstream outage
had recovered.

Fix: spawn the reconnect watcher via the existing _spawn_supervised()
task-level supervisor (already used for kanban_dispatcher_watcher,
handoff_watcher, etc.) instead of a bare asyncio.create_task, at both
the initial startup spawn and the manual-respawn path in
_ensure_reconnect_watcher_running(). _spawn_supervised already
provides exactly what's missing here: catches and logs any exception
escaping the task, and auto-restarts with capped exponential backoff
(healthy-run counter resets so a daemon that crashes occasionally over
days is never permanently abandoned) -- self-healing independent of
any new fatal-error event.

Also hardened a related race: the watcher's per-platform loop looked
up self._failed_platforms[platform] via direct indexing after
snapshotting the keys with list(...). A platform removed concurrently
between the snapshot and the lookup (e.g. a manual /platform resume,
or a reconnect that succeeded via a different path) would raise an
uncaught KeyError -- exactly the class of bug this fix's supervision
now catches, but avoiding the crash-and-restart cycle entirely is
better than relying on it. Changed to .get() with a skip-if-missing
guard.

6 new tests pass (initial spawn uses _spawn_supervised, manual respawn
uses _spawn_supervised, the core regression -- watcher self-heals
after an uncaught exception with no new fatal-error event -- and the
race-guard scenario); 52/52 in the full
tests/gateway/test_platform_reconnect.py file (including the 4
pre-existing _ensure_reconnect_watcher_running tests, confirming no
regression to that respawn-when-dead-or-missing behavior).
This commit is contained in:
ygd58 2026-07-26 09:06:07 +00:00 • committed by Teknium
parent 3c4220cd9c
commit 4b039e9543
2 changed files with 240 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

31
gateway/run.py
View file

@ -8710,11 +8710,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
", ".join(p.value for p in self._failed_platforms),
)
# Track the reconnect watcher task so _ensure_reconnect_watcher_running
# can detect if it dies and respawn it (#70344).
self._reconnect_watcher_task = asyncio.create_task(
self._platform_reconnect_watcher()
# can detect if it dies and respawn it (#70344). Spawned via
# _spawn_supervised (not a bare asyncio.create_task) so an exception
# escaping the watcher's OUTER while-loop -- not just the per-platform
# inner try/except -- is caught, logged, and auto-restarted with
# backoff instead of silently killing the watcher forever. Without
# this, a platform already queued in _failed_platforms when the
# watcher dies stays stranded indefinitely: _ensure_reconnect_watcher_running()
# only gets called from a NEW fatal-error arrival, so if no other
# platform ever fails afterward, nothing ever notices the watcher is
# dead (#71758 -- reported as 17.5h of silent downtime for a platform
# whose transient upstream outage had long since recovered).
self._reconnect_watcher_task = self._spawn_supervised(
self._platform_reconnect_watcher, "platform_reconnect_watcher"
)
self._background_tasks.add(self._reconnect_watcher_task)
# Start background handoff watcher — picks up CLI sessions marked
# handoff_state='pending' in state.db and re-binds them to the
@ -9297,11 +9306,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
"Reconnect watcher task is dead (done=%s) — respawning",
task.done() if task is not None else "N/A",
)
self._reconnect_watcher_task = asyncio.create_task(
self._platform_reconnect_watcher()
self._reconnect_watcher_task = self._spawn_supervised(
self._platform_reconnect_watcher, "platform_reconnect_watcher"
)
if getattr(self, "_background_tasks", None) is not None:
self._background_tasks.add(self._reconnect_watcher_task)
async def _platform_reconnect_watcher(self) -> None:
"""Background task that periodically retries connecting failed platforms.
@ -9333,7 +9340,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
for platform in list(self._failed_platforms.keys()):
if not self._running:
return
info = self._failed_platforms[platform]
info = self._failed_platforms.get(platform)
if info is None:
# Removed concurrently (e.g. a manual /platform resume,
# or a reconnect that succeeded via a different path)
# between the snapshot above and this lookup. Not an
# error -- just nothing to do for it this pass.
continue
# Skip paused platforms entirely — they need explicit
# /platform resume to come back.
if info.get("paused"):

219
tests/gateway/test_platform_reconnect.py
View file

@ -1126,7 +1126,224 @@ class TestEnsureReconnectWatcherRunning:
# ── _handle_adapter_fatal_error calls _ensure_reconnect_watcher ────────
class TestFatalErrorCallsEnsureWatcher:
class TestReconnectWatcherSelfHeals:
"""Regression tests for issue #71758: a platform already queued in
_failed_platforms when the reconnect watcher task dies from an
uncaught exception stayed stranded forever, because
_ensure_reconnect_watcher_running() is only called from a NEW
fatal-error arrival -- if no other platform ever fails afterward,
nothing notices the watcher is dead. The watcher must now be spawned
via _spawn_supervised (like other long-lived background tasks), so an
exception escaping its OUTER while-loop is caught, logged, and
auto-restarted with backoff -- independent of any new fatal-error
event.
"""
@pytest.mark.asyncio
async def test_startup_spawns_watcher_via_spawn_supervised(self, monkeypatch):
"""The initial watcher spawn at gateway startup must go through
_spawn_supervised, not a bare asyncio.create_task."""
runner = _make_runner()
runner._background_tasks = set()
calls = []
def fake_spawn_supervised(coro_factory, name, **kw):
calls.append(name)
task = asyncio.create_task(coro_factory())
runner._background_tasks.add(task)
return task
async def _noop_watcher():
await asyncio.sleep(3600)
monkeypatch.setattr(runner, "_spawn_supervised", fake_spawn_supervised)
monkeypatch.setattr(runner, "_platform_reconnect_watcher", _noop_watcher)
# Mirror the startup snippet: spawn via _spawn_supervised.
runner._reconnect_watcher_task = runner._spawn_supervised(
runner._platform_reconnect_watcher, "platform_reconnect_watcher"
)
assert "platform_reconnect_watcher" in calls
runner._reconnect_watcher_task.cancel()
try:
await runner._reconnect_watcher_task
except asyncio.CancelledError:
pass
@pytest.mark.asyncio
async def test_ensure_reconnect_watcher_running_uses_spawn_supervised(self):
"""The manual-respawn path in _ensure_reconnect_watcher_running must
also use _spawn_supervised, so a respawned watcher gets the same
auto-restart protection going forward."""
runner = _make_runner()
runner._running = True
runner._background_tasks = set()
runner._reconnect_watcher_task = asyncio.create_task(asyncio.sleep(0))
await runner._reconnect_watcher_task # let it finish (dead)
spawn_calls = []
original_spawn = runner._spawn_supervised
def spy_spawn_supervised(coro_factory, name, **kw):
spawn_calls.append(name)
return original_spawn(coro_factory, name, **kw)
runner._spawn_supervised = spy_spawn_supervised
runner._ensure_reconnect_watcher_running()
assert spawn_calls == ["platform_reconnect_watcher"]
runner._reconnect_watcher_task.cancel()
try:
await runner._reconnect_watcher_task
except asyncio.CancelledError:
pass
@pytest.mark.asyncio
async def test_watcher_self_heals_after_uncaught_exception_with_no_new_fatal_error(self):
"""The core #71758 regression: a platform sits queued in
_failed_platforms. The watcher task dies from an uncaught
exception (simulating the KeyError race / any other bug in the
outer loop). WITHOUT any new fatal-error event for a different
platform, the watcher must still come back on its own via
_spawn_supervised's crash-detection callback -- the exact gap
that stranded the platform for 17.5h in the reported bug.
"""
runner = _make_runner()
runner._running = True
runner._background_tasks = set()
runner._SUPERVISED_HEALTHY_SECS = GatewayRunner._SUPERVISED_HEALTHY_SECS
runner._MAX_SUPERVISED_RESTARTS = GatewayRunner._MAX_SUPERVISED_RESTARTS
runner._spawn_supervised = GatewayRunner._spawn_supervised.__get__(runner)
attempt_count = {"n": 0}
async def _flaky_watcher():
attempt_count["n"] += 1
if attempt_count["n"] == 1:
# Simulate the watcher's outer loop raising -- e.g. the
# KeyError race this same fix also hardens against, or any
# other bug in code outside the per-platform try/except.
raise RuntimeError("simulated watcher crash")
await asyncio.sleep(3600) # second run: stays "alive"
runner._reconnect_watcher_task = runner._spawn_supervised(
_flaky_watcher, "platform_reconnect_watcher"
)
# Let the first (crashing) attempt run and die.
for _ in range(50):
await asyncio.sleep(0)
if attempt_count["n"] >= 1 and runner._reconnect_watcher_task.done():
break
assert attempt_count["n"] == 1
assert runner._reconnect_watcher_task.done()
# The supervised _done callback schedules a respawn after a short
# backoff (2**0 = 1s at attempt 0) -- wait for it without a new
# fatal-error event ever firing.
for _ in range(30):
await asyncio.sleep(0.1)
if attempt_count["n"] >= 2:
break
assert attempt_count["n"] >= 2, (
"Watcher must self-heal via _spawn_supervised without any new "
"fatal-error event -- this is the exact gap that stranded a "
"platform in the reported bug"
)
# Cleanup: cancel whatever task is currently tracked.
for task in list(runner._background_tasks):
task.cancel()
await asyncio.sleep(0)
class TestReconnectWatcherRaceGuard:
"""Regression: a platform removed from _failed_platforms concurrently
(e.g. a manual /platform resume racing with the watcher's own
snapshot-then-lookup) must not raise KeyError and kill the loop
iteration -- it should just be skipped for that pass."""
@pytest.mark.asyncio
async def test_watcher_survives_platform_removed_mid_pass(self, monkeypatch):
"""Two platforms are due for retry. Reconnecting the first one, as
a side effect, removes the second from _failed_platforms (stand-in
for any concurrent path -- a manual /platform resume, a reconnect
that succeeded elsewhere, etc.). The watcher must finish its pass
without raising, and must still be alive afterward."""
runner = _make_runner()
runner._running = True
runner._background_tasks = set()
runner.session_store = MagicMock()
runner._busy_text_mode = "interrupt"
cfg = PlatformConfig(enabled=True, token="test")
runner._failed_platforms = {
Platform.TELEGRAM: {"config": cfg, "attempts": 0, "next_retry": 0.0},
Platform.DISCORD: {"config": cfg, "attempts": 0, "next_retry": 0.0},
}
runner._platform_connect_timeout_secs = MagicMock(return_value=5)
runner._sync_voice_mode_state_to_adapter = MagicMock()
runner._schedule_resume_pending_sessions = MagicMock()
runner._make_adapter_auth_check = MagicMock(return_value=lambda *a, **kw: True)
runner._recover_telegram_topic_thread_id = MagicMock()
runner._handle_message = MagicMock()
runner._handle_active_session_busy_message = MagicMock()
runner._handle_reaction_event = MagicMock()
runner._update_platform_runtime_status = MagicMock()
def fake_create_adapter(platform, platform_config):
adapter = MagicMock()
adapter.platform = platform
adapter.has_fatal_error = False
adapter.set_message_handler = MagicMock()
adapter.set_fatal_error_handler = MagicMock()
adapter.set_session_store = MagicMock()
adapter.set_busy_session_handler = MagicMock()
adapter.set_reaction_handler = MagicMock()
adapter.set_topic_recovery_fn = MagicMock()
adapter.set_authorization_check = MagicMock()
if platform == Platform.TELEGRAM:
# Side effect: concurrently "resolves" Discord's entry via
# some other path (e.g. a manual /platform resume), racing
# with the watcher's own snapshot-then-lookup for it.
runner._failed_platforms.pop(Platform.DISCORD, None)
return adapter
runner._create_adapter = MagicMock(side_effect=fake_create_adapter)
async def fake_connect(adapter, platform, is_reconnect=False):
return True
runner._connect_adapter_with_timeout = fake_connect
async def _one_pass():
now = time.monotonic()
for platform in list(runner._failed_platforms.keys()):
info = runner._failed_platforms.get(platform)
if info is None:
continue
if now < info["next_retry"]:
continue
adapter = runner._create_adapter(platform, info["config"])
success = await runner._connect_adapter_with_timeout(
adapter, platform, is_reconnect=True
)
if success:
runner.adapters[platform] = adapter
runner._failed_platforms.pop(platform, None)
# Must not raise, even though Discord vanishes from the dict as a
# side effect of processing Telegram.
await _one_pass()
assert Platform.TELEGRAM in runner.adapters
assert Platform.DISCORD not in runner._failed_platforms
"""Verify _handle_adapter_fatal_error calls _ensure_reconnect_watcher_running."""
@pytest.mark.asyncio