ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(gateway): unify adapter handler wiring, detach-wait, background-task tracking; split reconnect watcher

Browse Source
This commit is contained in:
Teknium
2026-09-02 18:29:04 -07:00
parent 113f04616b
commit 4368d6df52
1 changed files with 330 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/run_adapters.py
+330 -364
View File
@@ -10,6 +10,7 @@ from __future__ import annotations
import logging
from typing import TYPE_CHECKING
import asyncio
import contextlib
import os
import time
import weakref as _weakref
@@ -33,20 +34,15 @@ logger = logging.getLogger("gateway.run")
class GatewayAdapterLifecycleMixin:
"""Adapter connect/disconnect, fatal-error recovery, reconnect watcher and multiplex profile adapter methods for GatewayRunner."""
async def _await_adapter_cleanup_with_timeout(
self, awaitable: Awaitable[Any], timeout: float
) -> bool:
"""Wait for adapter cleanup without letting cancellation swallowing hang us.
@staticmethod
async def _wait_or_detach(task: "asyncio.Future", timeout: float) -> bool:
"""Wait up to ``timeout`` for ``task``; on deadline (or our own cancellation) detach it.
``asyncio.wait_for`` cancels an overdue child but then waits for it to exit. An adapter
close path that catches ``CancelledError`` can therefore block recovery forever. Keep
ownership of the old task through its done callback, but release the runner at the deadline.
Deliberately not ``asyncio.wait_for``: that cancels the overdue child and then WAITS for it
to exit, so a connect()/close() that swallows ``CancelledError`` blocks recovery forever.
Ownership stays with the task's done callback; the runner is released at the deadline.
Returns True when the task finished in time.
"""
if timeout <= 0:
await awaitable
return True
task = asyncio.ensure_future(awaitable)
try:
done, _pending = await asyncio.wait({task}, timeout=timeout)
except asyncio.CancelledError:
@@ -54,13 +50,24 @@ class GatewayAdapterLifecycleMixin:
task.add_done_callback(consume_detached_task_result)
raise
if task in done:
await task
return True
task.cancel()
task.add_done_callback(consume_detached_task_result)
return False
async def _await_adapter_cleanup_with_timeout(
self, awaitable: Awaitable[Any], timeout: float
) -> bool:
"""Await adapter cleanup with a detach-on-deadline bound; True when it completed."""
if timeout <= 0:
await awaitable
return True
task = asyncio.ensure_future(awaitable)
if await self._wait_or_detach(task, timeout):
await task
return True
return False
async def _safe_adapter_disconnect(self, adapter, platform) -> None:
"""Call adapter.disconnect() defensively, swallowing any error.
@@ -130,21 +137,23 @@ class GatewayAdapterLifecycleMixin:
platform.value, time.monotonic() - started_at, suffix, e,
)
@staticmethod
def _env_timeout_override(name: str) -> Optional[float]:
"""Non-negative float from env var ``name``; None when unset or unparseable (warned)."""
raw = os.getenv(name, "").strip()
if not raw:
return None
try:
return max(0.0, float(raw))
except ValueError:
logger.warning("Ignoring invalid %s=%r", name, raw)
return None
def _adapter_disconnect_timeout_secs(self) -> float:
"""Return the per-adapter disconnect timeout used during shutdown."""
from gateway.run import _ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT
raw = os.getenv("HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT", "").strip()
if raw:
try:
timeout = float(raw)
except ValueError:
logger.warning(
"Ignoring invalid HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT=%r",
raw,
)
else:
return max(0.0, timeout)
return _ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT
override = self._env_timeout_override("HERMES_GATEWAY_ADAPTER_DISCONNECT_TIMEOUT")
return _ADAPTER_DISCONNECT_TIMEOUT_SECS_DEFAULT if override is None else override
def _platform_connect_timeout_secs(self, platform=None, *, initial: bool = False) -> float:
"""Return the per-platform connect timeout used during startup/retry.
@@ -159,17 +168,9 @@ class GatewayAdapterLifecycleMixin:
_TELEGRAM_CONNECT_TIMEOUT_SECS_DEFAULT,
_TELEGRAM_INITIAL_CONNECT_TIMEOUT_SECS_DEFAULT,
)
raw = os.getenv("HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT", "").strip()
if raw:
try:
timeout = float(raw)
except ValueError:
logger.warning(
"Ignoring invalid HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT=%r",
raw,
)
else:
return max(0.0, timeout)
override = self._env_timeout_override("HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT")
if override is not None:
return override
if platform == Platform.TELEGRAM:
if initial:
return _TELEGRAM_INITIAL_CONNECT_TIMEOUT_SECS_DEFAULT
@@ -189,26 +190,10 @@ class GatewayAdapterLifecycleMixin:
timeout = self._platform_connect_timeout_secs(platform, initial=initial)
if timeout <= 0:
return await adapter.connect(is_reconnect=is_reconnect)
# Detach-on-timeout rather than plain asyncio.wait_for: wait_for cancels the overdue task but
# then waits for it to exit, so a connect() that catches CancelledError blocks recovery
# forever (watcher never retries). Keep ownership via its done callback; release at deadline.
task = asyncio.ensure_future(
adapter.connect(is_reconnect=is_reconnect)
)
try:
done, _pending = await asyncio.wait({task}, timeout=timeout)
except asyncio.CancelledError:
task.cancel()
task.add_done_callback(consume_detached_task_result)
raise
if task in done:
result = await task
return bool(result)
task.cancel()
task.add_done_callback(consume_detached_task_result)
raise TimeoutError(
f"{platform.value} connect timed out after {timeout:g}s"
)
task = asyncio.ensure_future(adapter.connect(is_reconnect=is_reconnect))
if await self._wait_or_detach(task, timeout):
return bool(await task)
raise TimeoutError(f"{platform.value} connect timed out after {timeout:g}s")
async def _connect_initial_adapter_with_timeout(self, adapter, platform) -> bool:
"""Connect one cold-start adapter with tightly scoped replace intent.
@@ -330,32 +315,15 @@ class GatewayAdapterLifecycleMixin:
)
self._queue_retryable_fatal_platform(adapter)
except asyncio.CancelledError:
# Best-effort queue before re-raising: a cancelled fatal handler
# must not strand a retryable platform (#80598).
try:
self._queue_retryable_fatal_platform(adapter)
except Exception:
logger.debug(
"Failed to queue %s after fatal-handler cancellation",
adapter.platform.value,
exc_info=True,
)
# A cancelled or raising fatal handler must not strand a retryable platform.
self._queue_retryable_best_effort(adapter, "cancellation")
raise
except Exception:
logger.exception(
"Fatal-error handling for %s raised unexpectedly",
adapter.platform.value,
)
# Best-effort queue so an unexpected raise mid-handler cannot
# leave a retryable platform permanently deaf (#80598).
try:
self._queue_retryable_fatal_platform(adapter)
except Exception:
logger.debug(
"Failed to queue %s after fatal-handler exception",
adapter.platform.value,
exc_info=True,
)
self._queue_retryable_best_effort(adapter, "exception")
finally:
platform = adapter.platform
shutdown_event = getattr(self, "_shutdown_event", None)
@@ -377,6 +345,15 @@ class GatewayAdapterLifecycleMixin:
self._exit_with_failure = True
await self.stop()
def _queue_retryable_best_effort(self, adapter: BasePlatformAdapter, why: str) -> None:
try:
self._queue_retryable_fatal_platform(adapter)
except Exception:
logger.debug(
"Failed to queue %s after fatal-handler %s",
adapter.platform.value, why, exc_info=True,
)
async def _handle_adapter_fatal_error_impl(self, adapter: BasePlatformAdapter) -> None:
# Snapshot this platform slot's current owner first: acting on a stale notification would
# overwrite a healthy platform's runtime status and wrongly re-queue it for reconnection.
@@ -447,6 +424,15 @@ class GatewayAdapterLifecycleMixin:
len(self._failed_platforms),
)
def _retain_background_task(self, task: "asyncio.Task") -> "asyncio.Task":
"""Register ``task`` in ``_background_tasks`` (created lazily for bare test runners)."""
tasks = getattr(self, "_background_tasks", None)
if not isinstance(tasks, set):
tasks = self._background_tasks = set()
tasks.add(task)
task.add_done_callback(tasks.discard)
return task
def _request_clean_exit(self, reason: str) -> None:
self._exit_cleanly = True
self._exit_reason = reason
@@ -475,11 +461,7 @@ class GatewayAdapterLifecycleMixin:
(e.g. ``_reconnect_watcher_task``) MUST pass it or a respawn leaves a stale handle and a
SECOND watcher. ``on_give_up(name)`` fires when the restart budget is spent.
"""
if getattr(self, "_background_tasks", None) is None:
self._background_tasks = set()
# Monotonic spawn timestamp captured per spawn: the ``_done`` callback
# uses it to distinguish a rapid crash-loop from a healthy-run-then-crash.
# Spawn timestamp lets ``_done`` tell a rapid crash-loop from a healthy-run-then-crash.
_started = time.monotonic()
# Deliberately no kwargs to create_task (some test doubles mock a narrow signature); calling
@@ -489,7 +471,7 @@ class GatewayAdapterLifecycleMixin:
# must ignore process-lifetime watchers or the gateway counts itself busy forever. Transient
# tasks added to _background_tasks elsewhere (startup-resume events etc.) stay counted.
task._hermes_supervised_watcher = True # type: ignore[attr-defined]
self._background_tasks.add(task)
self._retain_background_task(task)
if on_spawn is not None:
# Record the live handle NOW so an external tracker (e.g. _reconnect_watcher_task)
# points at the current task, not a dead one left by a prior supervised respawn.
@@ -551,9 +533,7 @@ class GatewayAdapterLifecycleMixin:
# The done callback retains its registration context, so isolate the backoff task
# too; otherwise a restart could reintroduce the original caller's turn scope.
respawn_task = Context().run(lambda: asyncio.create_task(_respawn()))
self._background_tasks.add(respawn_task)
respawn_task.add_done_callback(self._background_tasks.discard)
self._retain_background_task(Context().run(lambda: asyncio.create_task(_respawn())))
task.add_done_callback(_done)
return task
@@ -641,13 +621,15 @@ class GatewayAdapterLifecycleMixin:
# A row still 'running' at startup belongs to a gateway that died mid-dispatch: it can never
# reach a terminal state, and request_handoff refuses new requests while it sits there.
def _scope(profile_home):
if profile_home is None:
return contextlib.nullcontext()
return _async_profile_runtime_scope(profile_home)
for _pname, _phome in _handoff_watch_scopes(self):
try:
if _phome is None:
async with _scope(_phome):
await _reclaim_stale(self)
else:
async with _async_profile_runtime_scope(_phome):
await _reclaim_stale(self)
except Exception:
logger.debug("Stale-handoff reclaim failed", exc_info=True)
@@ -655,11 +637,8 @@ class GatewayAdapterLifecycleMixin:
while self._running:
try:
for profile_name, profile_home in _handoff_watch_scopes(self):
if profile_home is None:
async with _scope(profile_home):
await _tick(profile_name)
else:
async with _async_profile_runtime_scope(profile_home):
await _tick(profile_name)
except asyncio.CancelledError:
raise
except Exception as exc:
@@ -736,11 +715,7 @@ class GatewayAdapterLifecycleMixin:
)
)
respawn_task = asyncio.create_task(_slow_respawn())
if getattr(self, "_background_tasks", None) is None:
self._background_tasks = set()
self._background_tasks.add(respawn_task)
respawn_task.add_done_callback(self._background_tasks.discard)
self._retain_background_task(asyncio.create_task(_slow_respawn()))
def _spawn_reconnect_watcher(self, *, on_give_up=None):
"""Single place that knows how to launch the reconnect watcher.
@@ -781,12 +756,6 @@ class GatewayAdapterLifecycleMixin:
indefinitely so transient outages self-heal, non-retryable (bad auth) drop out immediately.
The circuit breaker (``/platform pause``) is manual only — auto-pausing left bots dead.
"""
from gateway.run import (
_dispose_unused_adapter,
_platform_has_bot_credential,
_reconnect_backoff,
_reconnect_needs_attention,
)
await asyncio.sleep(10) # initial delay — let startup finish
while self._running:
if not self._failed_platforms:
@@ -803,195 +772,7 @@ class GatewayAdapterLifecycleMixin:
for platform in list(self._failed_platforms.keys()):
if not self._running:
return
info = self._failed_platforms.get(platform)
if info is None:
# Removed concurrently (/platform resume, reconnect via another path) between
# the snapshot above and this lookup — not an error, nothing to do this pass.
continue
# Skip paused platforms entirely — they need explicit
# /platform resume to come back.
if info.get("paused"):
continue
# Long-lived retry escalation: past the attention threshold flag the platform
# NEEDS_ATTENTION in runtime status so a dead token/revoked intent doesn't look
# like ordinary "retrying" forever. A signal, NOT a circuit breaker — retries continue.
if not info.get("attention_flagged") and _reconnect_needs_attention(info, now):
info["attention_flagged"] = True
queued_for = now - info.get("queued_at", now)
retrying_since_iso = (
datetime.now(timezone.utc) - timedelta(seconds=queued_for)
).isoformat()
logger.warning(
"%s has been failing/reconnecting continuously for "
"%.1f hours (%d attempts) — flagging NEEDS_ATTENTION. "
"Retries continue, but this usually means a permanent "
"problem (revoked credentials, missing intents, broken "
"sidecar). Check `hermes status` / `/platform list`.",
platform.value,
queued_for / 3600.0,
info.get("attempts", 0),
)
self._update_platform_runtime_status(
platform.value,
platform_state="retrying",
needs_attention=True,
retrying_since=retrying_since_iso,
)
if now < info["next_retry"]:
continue # not time yet
platform_config = info["config"]
attempt = info["attempts"] + 1
# Empty-token primary configs can never reconnect; drop them so multiplex setups
# where a secondary profile owns the bot do not spin forever.
if not _platform_has_bot_credential(platform, platform_config):
logger.warning(
"Reconnect %s: no bot credential on queued config, "
"removing from retry queue",
platform.value,
)
del self._failed_platforms[platform]
continue
logger.info(
"Reconnecting %s (attempt %d)...",
platform.value, attempt,
)
adapter = None
try:
adapter = self._create_adapter(platform, platform_config)
if not adapter:
logger.warning(
"Reconnect %s: adapter creation returned None, removing from retry queue",
platform.value,
)
del self._failed_platforms[platform]
continue
adapter.set_message_handler(self._primary_message_handler())
adapter.set_fatal_error_handler(self._handle_adapter_fatal_error)
adapter.set_session_store(self.session_store)
adapter.set_busy_session_handler(self._handle_active_session_busy_message)
_set_reaction = getattr(adapter, "set_reaction_handler", None)
if callable(_set_reaction):
_set_reaction(self._handle_reaction_event)
adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform))
adapter.set_platform_event_handler(self._primary_platform_event_handler())
adapter._busy_text_mode = self._busy_text_mode
# Reconnect after outage: keep the platform's server-side update queue so
# messages sent while the bot was offline are delivered rather than dropped.
success = await self._connect_adapter_with_timeout(
adapter, platform, is_reconnect=True
)
if success:
self.adapters[platform] = adapter
self._sync_voice_mode_state_to_adapter(adapter)
# Wire voice input callback on reconnect as well (#60623).
self._bind_voice_input_callback(adapter)
self.delivery_router.adapters = self.adapters
del self._failed_platforms[platform]
self._update_platform_runtime_status(
platform.value,
platform_state="connected",
error_code=None,
error_message=None,
needs_attention=False,
retrying_since=None,
)
logger.info("✓ %s reconnected successfully", platform.value)
# Final responses rejected while this adapter was down are still owned by
# this live process, so startup recovery cannot claim them. Replay the
# explicitly transient subset now that the platform is usable.
try:
await self._redeliver_failed_obligations_for_platform(
platform
)
except Exception:
logger.debug(
"failed-obligation redelivery after %s reconnect failed",
platform.value,
exc_info=True,
)
# Rebuild channel directory with the new adapter
try:
from gateway.channel_directory import build_channel_directory
await build_channel_directory(self.adapters)
except Exception:
pass
# A platform that was offline at gateway startup never got its restart-
# interrupted sessions auto-resumed — the startup pass skips sessions whose
# adapter isn't connected yet.
try:
self._schedule_resume_pending_sessions(platform=platform)
except Exception:
logger.debug(
"resume-pending reschedule after %s reconnect failed",
platform.value,
exc_info=True,
)
# Check if the failure is non-retryable
elif adapter.has_fatal_error and not adapter.fatal_error_retryable:
self._update_platform_runtime_status(
platform.value,
platform_state="fatal",
error_code=adapter.fatal_error_code,
error_message=adapter.fatal_error_message,
)
logger.warning(
"Reconnect %s: non-retryable error (%s), removing from retry queue",
platform.value, adapter.fatal_error_message,
)
# The adapter is about to be dropped from the queue without ever being
# installed on self.adapters, so nothing else will call disconnect() on it.
# Dispose here or the resource owners built in __init__ (ResponseStore etc.)
# leak ~2 fds each; at the 300s cap the gateway hits the fd limit in ~12h.
await _dispose_unused_adapter(adapter)
del self._failed_platforms[platform]
else:
self._update_platform_runtime_status(
platform.value,
platform_state="retrying",
error_code=adapter.fatal_error_code,
error_message=adapter.fatal_error_message or "failed to reconnect",
)
backoff = _reconnect_backoff(attempt)
info["attempts"] = attempt
info["next_retry"] = time.monotonic() + backoff
logger.info(
"Reconnect %s failed, next retry in %ds",
platform.value, backoff,
)
# Same fd-leak concern as the non-retryable branch above: the adapter failed
# to connect and is being thrown away.
await _dispose_unused_adapter(adapter)
# Retryable failures (network/DNS blips) retry at the backoff cap forever,
# self-healing when connectivity returns. Never auto-pause them: a transient
# outage must not need `/platform resume`. Everything here is retryable.
except Exception as e:
if adapter is not None:
# An exception escaping connect (DNS timeout, aiohttp server.start() crash,
# etc.) leaves the adapter in the same unowned state as the branches above.
await _dispose_unused_adapter(adapter)
self._update_platform_runtime_status(
platform.value,
platform_state="retrying",
error_code=None,
error_message=str(e),
)
backoff = _reconnect_backoff(attempt)
info["attempts"] = attempt
info["next_retry"] = time.monotonic() + backoff
logger.warning(
"Reconnect %s error: %s, next retry in %ds",
platform.value, e, backoff,
)
# A reconnect exception (connect timeout, DNS failure, ...) is transient; keep
# retrying at the backoff cap rather than auto-pausing.
await self._reconnect_failed_platform(platform, now)
# Check every 10 seconds for platforms that need reconnection
for _ in range(10):
@@ -999,6 +780,190 @@ class GatewayAdapterLifecycleMixin:
return
await asyncio.sleep(1)
def _flag_reconnect_needs_attention(self, platform, info: dict, now: float) -> None:
"""Past the attention threshold flag NEEDS_ATTENTION in runtime status (once).
A signal, NOT a circuit breaker — retries continue; a dead token/revoked intent must not
look like ordinary "retrying" forever.
"""
from gateway.run import _reconnect_needs_attention
if info.get("attention_flagged") or not _reconnect_needs_attention(info, now):
return
info["attention_flagged"] = True
queued_for = now - info.get("queued_at", now)
retrying_since_iso = (
datetime.now(timezone.utc) - timedelta(seconds=queued_for)
).isoformat()
logger.warning(
"%s has been failing/reconnecting continuously for "
"%.1f hours (%d attempts) — flagging NEEDS_ATTENTION. "
"Retries continue, but this usually means a permanent "
"problem (revoked credentials, missing intents, broken "
"sidecar). Check `hermes status` / `/platform list`.",
platform.value,
queued_for / 3600.0,
info.get("attempts", 0),
)
self._update_platform_runtime_status(
platform.value,
platform_state="retrying",
needs_attention=True,
retrying_since=retrying_since_iso,
)
@staticmethod
def _bump_reconnect_backoff(info: dict, attempt: int) -> int:
"""Record a failed attempt in the queue entry; returns the backoff applied."""
from gateway.run import _reconnect_backoff
backoff = _reconnect_backoff(attempt)
info["attempts"] = attempt
info["next_retry"] = time.monotonic() + backoff
return backoff
async def _reconnect_failed_platform(self, platform, now: float) -> None:
"""One watcher pass for a queued platform: gate, attempt, and record the outcome."""
from gateway.run import _dispose_unused_adapter, _platform_has_bot_credential
info = self._failed_platforms.get(platform)
if info is None:
# Removed concurrently (/platform resume, reconnect via another path) between the
# caller's snapshot and this lookup — not an error, nothing to do this pass.
return
# Paused platforms need an explicit /platform resume to come back.
if info.get("paused"):
return
self._flag_reconnect_needs_attention(platform, info, now)
if now < info["next_retry"]:
return # not time yet
platform_config = info["config"]
attempt = info["attempts"] + 1
# Empty-token primary configs can never reconnect; drop them so multiplex setups
# where a secondary profile owns the bot do not spin forever.
if not _platform_has_bot_credential(platform, platform_config):
logger.warning(
"Reconnect %s: no bot credential on queued config, "
"removing from retry queue",
platform.value,
)
del self._failed_platforms[platform]
return
logger.info("Reconnecting %s (attempt %d)...", platform.value, attempt)
adapter = None
try:
adapter = self._create_adapter(platform, platform_config)
if not adapter:
logger.warning(
"Reconnect %s: adapter creation returned None, removing from retry queue",
platform.value,
)
del self._failed_platforms[platform]
return
self._wire_adapter_handlers(adapter)
# Reconnect after outage: keep the platform's server-side update queue so
# messages sent while the bot was offline are delivered rather than dropped.
success = await self._connect_adapter_with_timeout(
adapter, platform, is_reconnect=True
)
if success:
await self._install_reconnected_adapter(platform, adapter)
elif adapter.has_fatal_error and not adapter.fatal_error_retryable:
self._update_platform_runtime_status(
platform.value,
platform_state="fatal",
error_code=adapter.fatal_error_code,
error_message=adapter.fatal_error_message,
)
logger.warning(
"Reconnect %s: non-retryable error (%s), removing from retry queue",
platform.value, adapter.fatal_error_message,
)
# Dropped from the queue without ever being installed on self.adapters, so nothing
# else will call disconnect() on it. Dispose here or the resource owners built in
# __init__ (ResponseStore etc.) leak ~2 fds each — fd limit in ~12h at the 300s cap.
await _dispose_unused_adapter(adapter)
del self._failed_platforms[platform]
else:
self._update_platform_runtime_status(
platform.value,
platform_state="retrying",
error_code=adapter.fatal_error_code,
error_message=adapter.fatal_error_message or "failed to reconnect",
)
backoff = self._bump_reconnect_backoff(info, attempt)
logger.info(
"Reconnect %s failed, next retry in %ds",
platform.value, backoff,
)
# Same fd-leak concern as the non-retryable branch: the adapter is thrown away.
# Retryable failures (network/DNS blips) retry at the backoff cap forever; never
# auto-pause them — a transient outage must not need `/platform resume`.
await _dispose_unused_adapter(adapter)
except Exception as e:
if adapter is not None:
# An exception escaping connect (DNS timeout, aiohttp server.start() crash, ...)
# leaves the adapter in the same unowned state as the branches above.
await _dispose_unused_adapter(adapter)
self._update_platform_runtime_status(
platform.value,
platform_state="retrying",
error_code=None,
error_message=str(e),
)
backoff = self._bump_reconnect_backoff(info, attempt)
# A reconnect exception is transient; keep retrying at the cap rather than auto-pausing.
logger.warning(
"Reconnect %s error: %s, next retry in %ds",
platform.value, e, backoff,
)
async def _install_reconnected_adapter(self, platform, adapter) -> None:
"""Publish a freshly reconnected primary adapter and replay what it missed while down."""
self.adapters[platform] = adapter
self._sync_voice_mode_state_to_adapter(adapter)
self._bind_voice_input_callback(adapter)
self.delivery_router.adapters = self.adapters
del self._failed_platforms[platform]
self._update_platform_runtime_status(
platform.value,
platform_state="connected",
error_code=None,
error_message=None,
needs_attention=False,
retrying_since=None,
)
logger.info("✓ %s reconnected successfully", platform.value)
# Final responses rejected while this adapter was down are still owned by this live
# process, so startup recovery cannot claim them. Replay the transient subset now.
try:
await self._redeliver_failed_obligations_for_platform(platform)
except Exception:
logger.debug(
"failed-obligation redelivery after %s reconnect failed",
platform.value,
exc_info=True,
)
# Rebuild channel directory with the new adapter
try:
from gateway.channel_directory import build_channel_directory
await build_channel_directory(self.adapters)
except Exception:
pass
# A platform offline at gateway startup never got its restart-interrupted sessions
# auto-resumed — the startup pass skips sessions whose adapter isn't connected yet.
try:
self._schedule_resume_pending_sessions(platform=platform)
except Exception:
logger.debug(
"resume-pending reschedule after %s reconnect failed",
platform.value,
exc_info=True,
)
async def _cancel_secondary_profile_reconnect_tasks(self) -> None:
"""Cancel profile-scoped reconnects before tearing down their registry.
@@ -1335,6 +1300,42 @@ class GatewayAdapterLifecycleMixin:
)
return connected
def _wire_adapter_handlers(
self,
adapter: BasePlatformAdapter,
*,
message_handler=None,
fatal_error_handler=None,
busy_session_handler=None,
authorization_check=None,
platform_event_handler=None,
busy_text_mode: Optional[str] = None,
) -> None:
"""Install the runner callbacks every adapter needs, in the shared order.
Defaults are the primary-adapter handlers; secondary-profile wiring passes its
profile-scoped variants. ``set_reaction_handler`` is optional on plugin adapters.
"""
adapter.set_message_handler(message_handler or self._primary_message_handler())
adapter.set_fatal_error_handler(fatal_error_handler or self._handle_adapter_fatal_error)
adapter.set_session_store(self.session_store)
adapter.set_busy_session_handler(
busy_session_handler or self._handle_active_session_busy_message
)
_set_reaction = getattr(adapter, "set_reaction_handler", None)
if callable(_set_reaction):
_set_reaction(self._handle_reaction_event)
adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
adapter.set_authorization_check(
authorization_check or self._make_adapter_auth_check(adapter.platform)
)
adapter.set_platform_event_handler(
platform_event_handler or self._primary_platform_event_handler()
)
adapter._busy_text_mode = (
self._busy_text_mode if busy_text_mode is None else busy_text_mode
)
def _configure_profile_adapter(
self,
adapter: BasePlatformAdapter,
@@ -1345,11 +1346,6 @@ class GatewayAdapterLifecycleMixin:
# Runtime status is process-scoped while message/config work is profile-scoped. Keep both
# dimensions in the key so dashboard/NAS health aggregation sees which secondary failed.
adapter._runtime_status_platform_key = f"{profile_name}:{platform.value}"
adapter.set_message_handler(self._make_profile_message_handler(profile_name))
adapter.set_fatal_error_handler(
self._make_profile_fatal_error_handler(profile_name, platform)
)
adapter.set_session_store(self.session_store)
# Declare credential ownership BEFORE any inbound event can be handled: adapter-level
# session keys (batching, _active_sessions, busy guard) are derived at ingress, before the
# handler stamps source.profile — without this every secondary bot would key into the
@@ -1357,30 +1353,25 @@ class GatewayAdapterLifecycleMixin:
_set_owner = getattr(adapter, "set_owner_profile", None)
if callable(_set_owner):
_set_owner(profile_name)
adapter.set_busy_session_handler(
self._make_profile_busy_session_handler(profile_name)
)
_set_reaction = getattr(adapter, "set_reaction_handler", None)
if callable(_set_reaction):
_set_reaction(self._handle_reaction_event)
adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
adapter.set_authorization_check(
self._make_adapter_auth_check(platform, profile_name=profile_name)
)
adapter.set_platform_event_handler(
self._make_profile_platform_event_handler(profile_name)
)
# Voice transcripts from this bot's channels dispatch through THIS
# adapter (primary wiring lives at connect time; see #75198).
self._bind_voice_input_callback(adapter)
text_modes = getattr(self, "_busy_text_modes_by_profile", None)
adapter._busy_text_mode = (
text_modes.get(profile_name, self._busy_text_mode)
if isinstance(text_modes, dict)
else self._busy_text_mode
self._wire_adapter_handlers(
adapter,
message_handler=self._make_profile_message_handler(profile_name),
fatal_error_handler=self._make_profile_fatal_error_handler(profile_name, platform),
busy_session_handler=self._make_profile_busy_session_handler(profile_name),
authorization_check=self._make_adapter_auth_check(platform, profile_name=profile_name),
platform_event_handler=self._make_profile_platform_event_handler(profile_name),
busy_text_mode=(
text_modes.get(profile_name, self._busy_text_mode)
if isinstance(text_modes, dict)
else self._busy_text_mode
),
)
# Secondary adapters always carry the profile they serve so prune
# paths namespace topic bindings correctly under multiplex (#76423).
# Voice transcripts from this bot's channels dispatch through THIS adapter (primary
# wiring lives at connect time).
self._bind_voice_input_callback(adapter)
# Secondary adapters always carry the profile they serve so prune paths namespace topic
# bindings correctly under multiplex.
adapter._hermes_profile_name = profile_name
async def _run_secondary_profile_reconnect(
@@ -1532,51 +1523,33 @@ class GatewayAdapterLifecycleMixin:
)
return
def _handoff() -> None:
try:
self._schedule_secondary_profile_reconnect(profile_name, platform, adapter)
except Exception:
# The handoff touches live registries; if it raises, the parked task dies as an
# unretrieved-task exception logged only at GC. Surface it where operators look.
logger.exception(
"secondary-startup-reconnect handoff failed (profile=%s platform=%s)",
profile_name,
platform.value,
)
async def _await_running_then_schedule() -> None:
if self._running:
try:
self._schedule_secondary_profile_reconnect(
profile_name, platform, adapter
)
except Exception:
# Same GC-time-exception hazard as the post-poll handoff
# below; surface it in gateway.log instead.
logger.exception(
"secondary-startup-reconnect handoff failed "
"(profile=%s platform=%s)",
profile_name,
platform.value,
)
_handoff()
return
# Modest poll: startup completion has no dedicated event, and the reconnect runner's own
# backoff makes sub-100ms precision irrelevant. Bounded so a wedged startup cannot spin.
while not self._running and not self._shutdown_event.is_set():
await asyncio.sleep(0.1)
if self._running and not self._shutdown_event.is_set():
try:
self._schedule_secondary_profile_reconnect(
profile_name, platform, adapter
)
except Exception:
# The handoff touches live registries; if it raises, the parked task dies as an
# unretrieved-task exception logged only at GC. Surface it where operators look.
logger.exception(
"secondary-startup-reconnect handoff failed "
"(profile=%s platform=%s)",
profile_name,
platform.value,
)
_handoff()
task = asyncio.create_task(
self._retain_background_task(asyncio.create_task(
_await_running_then_schedule(),
name=f"secondary-startup-reconnect:{profile_name}:{platform.value}",
)
background_tasks = getattr(self, "_background_tasks", None)
if not isinstance(background_tasks, set):
background_tasks = set()
self._background_tasks = background_tasks
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)
))
def _schedule_secondary_profile_reconnect(
self, profile_name: str, platform: Platform, adapter: BasePlatformAdapter
@@ -1591,17 +1564,10 @@ class GatewayAdapterLifecycleMixin:
profile_pending = pending.setdefault(profile_name, {})
if platform in profile_pending:
return
task = asyncio.create_task(
profile_pending[platform] = self._retain_background_task(asyncio.create_task(
self._run_secondary_profile_reconnect(profile_name, platform),
name=f"secondary-reconnect:{profile_name}:{platform.value}",
)
profile_pending[platform] = task
background_tasks = getattr(self, "_background_tasks", None)
if not isinstance(background_tasks, set):
background_tasks = set()
self._background_tasks = background_tasks
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)
))
def _make_profile_fatal_error_handler(
self, profile_name: str, platform: Platform