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

Merge branch 'simp/r3-11-2d' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-03 00:11:10 -07:00
parent e88ebcb0c7 792960cba1
commit 4ad49247e1
4 changed files with 573 additions and 1079 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/run_adapters.py
+130 -238
View File
@@ -11,6 +11,7 @@ import logging
from typing import TYPE_CHECKING
import asyncio
import contextlib
from contextlib import suppress
import functools
import os
import time
@@ -38,22 +39,17 @@ class GatewayAdapterLifecycleMixin:
@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.
Not ``asyncio.wait_for``: that WAITS for the cancelled child to exit, so a connect()/close()
that swallows ``CancelledError`` blocks recovery forever. Returns True if it finished in time.
"""
"""Wait up to ``timeout`` for ``task``; on deadline (or our own cancellation) detach it. Not
``asyncio.wait_for``: that WAITS for the cancelled child, so a connect()/close() swallowing
``CancelledError`` blocks recovery forever. True if it finished in time."""
done: set = set()
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:
return True
task.cancel()
task.add_done_callback(consume_detached_task_result)
return False
finally:
if task not in done: # timed out, or our own cancellation
task.cancel()
task.add_done_callback(consume_detached_task_result)
return task in done
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."""
@@ -67,11 +63,8 @@ class GatewayAdapterLifecycleMixin:
return True
async def _safe_adapter_disconnect(self, adapter, platform) -> None:
"""Call adapter.disconnect() defensively (bounded, never raises).
After a failed connect() partial resources (ClientSession, poll tasks, subprocesses) would
otherwise leak; must tolerate partial-init state.
"""
"""Call adapter.disconnect() defensively (bounded, never raises, tolerates partial-init state):
after a failed connect() partial resources (ClientSession, poll tasks, subprocesses) leak."""
timeout = self._adapter_disconnect_timeout_secs()
label = platform.value if platform is not None else "adapter"
with _log_suppressed(logging.DEBUG, "Defensive %s disconnect after failed connect raised: %s", label):
@@ -82,25 +75,24 @@ class GatewayAdapterLifecycleMixin:
)
async def _bounded_adapter_teardown(self, adapter, platform, *, profile: Optional[str] = None) -> None:
"""Tear down one adapter on the shutdown path with bounded awaits (never raises).
Unbounded, a half-dead transport stalls shutdown past systemd's ``TimeoutStopSec``; the
SIGKILL skips ``atexit`` PID-file cleanup and the next start dies with "PID file race lost".
"""
"""Tear down one adapter on the shutdown path with bounded awaits (never raises). Unbounded,
a half-dead transport stalls past systemd's ``TimeoutStopSec``; the SIGKILL skips ``atexit``
PID-file cleanup and the next start dies with "PID file race lost"."""
timeout = self._adapter_disconnect_timeout_secs()
suffix = f" (profile: {profile})" if profile else ""
started_at = time.monotonic()
try:
if not await self._await_adapter_cleanup_with_timeout(
adapter.cancel_background_tasks(), timeout
):
if not await self._await_adapter_cleanup_with_timeout(adapter.cancel_background_tasks(), timeout):
logger.warning(
"✗ %s background-task cancel timed out after %.1fs - forcing continue%s",
platform.value, timeout, suffix,
)
except Exception as e:
logger.debug("✗ %s background-task cancel error%s: %s", platform.value, suffix, e)
try:
with _log_suppressed(
logging.ERROR, "✗ %s disconnect error after %.2fs%s: %s",
platform.value, time.monotonic() - started_at, suffix,
):
if await self._await_adapter_cleanup_with_timeout(adapter.disconnect(), timeout):
logger.info(
"✓ %s disconnected (%.2fs)%s", platform.value, time.monotonic() - started_at, suffix,
@@ -110,11 +102,6 @@ class GatewayAdapterLifecycleMixin:
"✗ %s disconnect timed out after %.1fs - forcing continue%s",
platform.value, timeout, suffix,
)
except Exception as e:
logger.error(
"✗ %s disconnect error after %.2fs%s: %s",
platform.value, time.monotonic() - started_at, suffix, e,
)
@staticmethod
def _env_timeout_override(name: str) -> Optional[float]:
@@ -177,18 +164,17 @@ class GatewayAdapterLifecycleMixin:
await self.hooks.emit(event_name, ctx)
async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> None:
"""React to an adapter failure after startup (retryable → background reconnect queue).
Detached task: the notification arrives on the failing adapter's own polling task, which
the handler's disconnect can cancel mid-flight, stranding the platform half-handled.
"""
"""React to an adapter failure after startup (retryable → background reconnect queue). Runs
detached: the notification arrives on the failing adapter's own polling task, which the
handler's disconnect can cancel mid-flight, stranding the platform half-handled."""
tasks = getattr(self, "_fatal_handler_tasks", None)
if tasks is None:
tasks = self._fatal_handler_tasks = set()
task = self._track_task_in(tasks, asyncio.create_task(self._handle_adapter_fatal_error_detached(adapter)))
# shield(): a plain `await task` would tunnel the caller's cancellation into the detached
# task; with shield the caller sees CancelledError and the handler runs to completion.
await asyncio.shield(task)
await asyncio.shield(
self._track_task_in(tasks, asyncio.create_task(self._handle_adapter_fatal_error_detached(adapter)))
)
def _reconnect_queue_entry(
self, platform, adapter, platform_config, *, attempts: int, delay: float, queued: bool = True
@@ -213,9 +199,8 @@ class GatewayAdapterLifecycleMixin:
if not platform_config:
return False
if adapter.platform in self._failed_platforms:
# Already queued is exactly when the watcher may have died (_spawn_supervised gives up
# after _MAX_SUPERVISED_RESTARTS); without this backstop the platform is a silent
# permanent outage — nothing retries and the stranded check treats "queued" as safe.
# Already queued is exactly when the watcher may have died (supervision gave up); without
# this backstop nothing retries and the stranded check treats "queued" as safe.
self._ensure_reconnect_watcher_running()
return False
self._failed_platforms[adapter.platform] = self._reconnect_queue_entry(
@@ -229,8 +214,7 @@ class GatewayAdapterLifecycleMixin:
"""Run the fatal handler; a platform left stranded (not reconnected, not queued, not
intentionally disabled) exits the gateway with failure so the service manager restarts it."""
try:
# Outer hard deadline: a hang anywhere in the impl must not wedge this task — the
# stranded check in ``finally`` only runs when we return.
# Outer hard deadline: the stranded check in ``finally`` only runs when we return.
timeout = self._adapter_disconnect_timeout_secs()
if timeout <= 0:
await self._handle_adapter_fatal_error_impl(adapter)
@@ -270,16 +254,14 @@ class GatewayAdapterLifecycleMixin:
await self.stop()
def _queue_retryable_best_effort(self, adapter: BasePlatformAdapter, why: str) -> None:
try:
with _log_suppressed(
logging.DEBUG, "Failed to queue %s after fatal-handler %s",
adapter.platform.value, why, exc_info=True,
):
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 the slot owner first: a stale notification must not overwrite a healthy
# platform's runtime status or re-queue it.
# Snapshot the slot owner first: a stale notification must not touch a healthy platform.
existing = self.adapters.get(adapter.platform)
if existing is not None and existing is not adapter:
logger.debug(
@@ -287,7 +269,6 @@ class GatewayAdapterLifecycleMixin:
adapter.platform.value, adapter.fatal_error_code or "unknown",
)
return
logger.error(
"Fatal %s adapter error (%s): %s", adapter.platform.value,
adapter.fatal_error_code or "unknown", adapter.fatal_error_message or "unknown error",
@@ -302,21 +283,15 @@ class GatewayAdapterLifecycleMixin:
error_code=adapter.fatal_error_code,
error_message=adapter.fatal_error_message,
)
if existing is adapter:
# Claim for teardown BEFORE awaiting disconnect(), else a concurrent second fatal
# notification for the same adapter would disconnect() it twice.
# Claim for teardown BEFORE awaiting disconnect(), else a second fatal disconnects it twice.
self.adapters.pop(adapter.platform, None)
self.delivery_router.adapters = self.adapters
# Queue BEFORE any disconnect await: a half-dead transport can wedge native close() (or
# swallow CancelledError), and "disconnect then queue" left platforms permanently deaf.
# Queue BEFORE any disconnect await: a wedged close() once left platforms permanently deaf.
self._queue_retryable_fatal_platform(adapter)
if existing is adapter:
# Bounded by the shutdown-path timeout so this always returns to the stranded check.
await self._safe_adapter_disconnect(adapter, adapter.platform)
if not self.adapters and not self._failed_platforms:
self._exit_reason = adapter.fatal_error_message or "All messaging adapters disconnected"
if adapter.fatal_error_retryable:
@@ -326,8 +301,7 @@ class GatewayAdapterLifecycleMixin:
logger.error("No connected messaging platforms remain. Shutting down gateway cleanly.")
await self.stop()
elif not self.adapters and self._failed_platforms:
# Everything is down but queued: stay alive (cron still runs, watcher recovers) rather
# than turn a transient outage into a state-killing systemd restart loop.
# All down but queued: stay alive (cron runs, watcher recovers) rather than restart-loop.
logger.warning(
"No connected messaging platforms remain, but %d platform(s) "
"queued for reconnection — gateway staying alive, watcher will "
@@ -363,24 +337,17 @@ class GatewayAdapterLifecycleMixin:
def _spawn_supervised(
self, coro_factory, name, *, restart=True, _attempt=0, on_spawn=None, on_give_up=None
):
"""Launch a long-lived background task with task-level supervision.
Catches outer-loop/pre-try exceptions a bare ``create_task`` drops silently. Restarts with
capped backoff up to ``_MAX_SUPERVISED_RESTARTS`` rapid failures; the counter resets after
``_SUPERVISED_HEALTHY_SECS`` of healthy running. Each spawn uses a fresh ``Context`` (an
inherited delegated-child marker would make the Kanban dispatcher reject its own writes).
``on_spawn`` fires on EVERY spawn incl. respawns — callers tracking the handle elsewhere
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.
"""
"""Launch a long-lived supervised background task: exceptions a bare ``create_task`` drops
are logged, and it respawns with capped backoff up to ``_MAX_SUPERVISED_RESTARTS`` rapid
failures (counter resets after ``_SUPERVISED_HEALTHY_SECS`` healthy). Fresh ``Context`` per
spawn (an inherited delegated-child marker would make the Kanban dispatcher reject its own
writes). ``on_spawn`` fires on EVERY spawn incl. respawns — handle trackers MUST pass it or a
respawn leaves a stale handle and a SECOND watcher; ``on_give_up(name)`` fires at budget end."""
# Spawn timestamp lets ``_done`` tell a rapid crash-loop from a healthy-run-then-crash.
_started = time.monotonic()
# No create_task kwargs (test doubles mock a narrow signature); running it from a fresh
# Context gives the same isolation as create_task(..., context=Context()).
# No create_task kwargs (test doubles mock a narrow signature); Context().run isolates instead.
task = Context().run(lambda: asyncio.create_task(coro_factory()))
# Marks a PERMANENT watcher so the scale-to-zero idle check ignores it (else the gateway
# counts itself busy forever); transient _background_tasks stay counted.
# PERMANENT watcher: the scale-to-zero idle check ignores it (else busy forever).
task._hermes_supervised_watcher = True # type: ignore[attr-defined]
self._retain_background_task(task)
if on_spawn is not None:
@@ -401,8 +368,7 @@ class GatewayAdapterLifecycleMixin:
logger.error("Supervised task %s died: %r", name, exc, exc_info=exc)
if not (restart and self._running):
return
# A healthy run before the crash is a FRESH failure, not a crash-loop: reset so a
# daemon crashing a few times over days is never abandoned.
# A healthy run before the crash is a FRESH failure, not a crash-loop: reset the counter.
healthy = time.monotonic() - _started >= self._SUPERVISED_HEALTHY_SECS
effective_attempt = 0 if healthy else _attempt
if effective_attempt >= self._MAX_SUPERVISED_RESTARTS:
@@ -434,23 +400,18 @@ class GatewayAdapterLifecycleMixin:
return task
async def _handoff_watcher(self, interval: float = 2.0, drain_timeout: float = 30.0) -> None:
"""Process pending CLI→gateway session handoffs from ``state.db``.
Claim atomically (pending → running), re-bind the home channel to the CLI session_id,
dispatch a synthetic ``MessageEvent``, mark ``completed``/``failed``.
"""
"""Process pending CLI→gateway session handoffs from ``state.db``: claim atomically (pending
→ running), re-bind the home channel to the CLI session_id, dispatch a synthetic event, mark
``completed``/``failed``."""
from gateway.run import _async_profile_runtime_scope, _handoff_watch_scopes, _reclaim_stale
await asyncio.sleep(5) # let platforms connect before dispatching through them
# Does _process_handoff accept the profile argument? Test stand-ins bind a one-arg callable.
try:
import inspect as _inspect
_process_takes_profile = len(_inspect.signature(self._process_handoff).parameters) >= 2
except Exception:
_process_takes_profile = False
# In-flight dispatches keyed by session id: a handoff runs a FULL agent turn, so inline
# processing would let one slow handoff time out every other profile's poll.
# In-flight dispatches by session id: a handoff is a FULL agent turn, so never process inline.
inflight: Dict[str, "asyncio.Task"] = {}
async def _dispatch(row, session_id, session_db, profile_name) -> None:
@@ -459,8 +420,7 @@ class GatewayAdapterLifecycleMixin:
await self._process_handoff(*((row, profile_name) if _process_takes_profile else (row,)))
await session_db.complete_handoff(session_id)
except asyncio.CancelledError:
# Gateway shutting down: leave the row 'running' so the next
# start's reclaim marks it failed with a clear reason.
# Leave the row 'running' so the next start's reclaim marks it failed with a clear reason.
raise
except Exception as exc:
logger.warning("Handoff for session %s failed: %s", session_id, exc, exc_info=True)
@@ -470,12 +430,9 @@ class GatewayAdapterLifecycleMixin:
inflight.pop(session_id, None)
async def _tick(profile_name: Optional[str] = None) -> None:
"""One poll of the CURRENTLY-SCOPED store; ``profile_name`` (None = root) routes
delivery to that profile's OWN adapter.
A closure, not a method: tests bind ``_handoff_watcher`` onto a ``SimpleNamespace``
with only ``_session_db``/``_running``/``_process_handoff``.
"""
"""One poll of the CURRENTLY-SCOPED store; ``profile_name`` (None = root) routes delivery
to that profile's OWN adapter. A closure, not a method: tests bind ``_handoff_watcher`` onto
a ``SimpleNamespace`` with only ``_session_db``/``_running``/``_process_handoff``."""
session_db = getattr(self, "_session_db", None)
if session_db is None:
return
@@ -487,15 +444,13 @@ class GatewayAdapterLifecycleMixin:
if not await session_db.claim_handoff(session_id):
# Another tick or another gateway already claimed it.
continue
# INVARIANT (do not weaken): this task is created inside _profile_runtime_scope but
# typically RUNS after it exits; it sees the profile's home/secret scope only because
# those seams are ContextVar-based and ensure_future copies the Context.
# INVARIANT (do not weaken): created inside _profile_runtime_scope but RUNS after it
# exits; it sees the profile scope only because ensure_future copies the Context.
inflight[session_id] = asyncio.ensure_future(
_dispatch(row, session_id, session_db, profile_name)
)
# A row still 'running' at startup belongs to a gateway that died mid-dispatch; it blocks
# request_handoff until reclaimed.
# A row still 'running' at startup died mid-dispatch and blocks request_handoff until reclaimed.
def _scope(profile_home): # local: tests bind this watcher onto bare SimpleNamespace runners
return GatewayAdapterLifecycleMixin._scope_or_null(_async_profile_runtime_scope, profile_home)
@@ -503,7 +458,6 @@ class GatewayAdapterLifecycleMixin:
with _log_suppressed(logging.DEBUG, "Stale-handoff reclaim failed", exc_info=True):
async with _scope(_phome):
await _reclaim_stale(self)
try:
while self._running:
try:
@@ -544,10 +498,9 @@ class GatewayAdapterLifecycleMixin:
"""Bounded slow-tier respawn of the reconnect watcher."""
if attempt >= self._MAX_SLOW_WATCHER_RESPAWNS:
logger.error(
"Reconnect watcher could not be kept alive after %d slow "
"respawns; %d platform(s) remain queued and unattended: %s. "
"Manual intervention or a gateway restart is required.", attempt,
len(self._failed_platforms), ", ".join(str(p) for p in self._failed_platforms),
"Reconnect watcher could not be kept alive after %d slow respawns; %d platform(s) remain "
"queued and unattended: %s. Manual intervention or a gateway restart is required.",
attempt, len(self._failed_platforms), ", ".join(str(p) for p in self._failed_platforms),
)
return
@@ -561,9 +514,8 @@ class GatewayAdapterLifecycleMixin:
):
return
logger.warning(
"Reconnect watcher still down with %d platform(s) queued — "
"slow respawn %d/%d", len(self._failed_platforms), attempt + 1,
self._MAX_SLOW_WATCHER_RESPAWNS,
"Reconnect watcher still down with %d platform(s) queued — slow respawn %d/%d",
len(self._failed_platforms), attempt + 1, self._MAX_SLOW_WATCHER_RESPAWNS,
)
self._spawn_reconnect_watcher(
on_give_up=lambda _name: self._schedule_slow_reconnect_watcher_respawn(attempt=attempt + 1)
@@ -627,18 +579,15 @@ class GatewayAdapterLifecycleMixin:
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),
"%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,
retrying_since=(datetime.now(timezone.utc) - timedelta(seconds=queued_for)).isoformat(),
)
def _mark_platform_fatal(self, status_key: str, adapter) -> None:
@@ -671,7 +620,6 @@ class GatewayAdapterLifecycleMixin:
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
@@ -680,17 +628,14 @@ class GatewayAdapterLifecycleMixin:
self._drop_from_reconnect_queue(platform, "no bot credential on queued config")
return
logger.info("Reconnecting %s (attempt %d)...", platform.value, attempt)
adapter = None
try:
adapter = self._create_adapter(platform, platform_config)
if not adapter:
self._drop_from_reconnect_queue(platform, "adapter creation returned None")
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.
# is_reconnect keeps the server-side update queue so offline-period messages are delivered.
success = await self._connect_adapter_with_timeout(adapter, platform, is_reconnect=True)
if success:
await self._install_reconnected_adapter(platform, adapter)
@@ -700,13 +645,11 @@ class GatewayAdapterLifecycleMixin:
"Reconnect %s: non-retryable error (%s), removing from retry queue",
platform.value, adapter.fatal_error_message,
)
# Never installed on self.adapters, so nothing else will disconnect() it: dispose or
# the resource owners built in __init__ leak ~2 fds each (fd limit in ~12h at the cap).
# Never installed on self.adapters: dispose here or its __init__ resources leak ~2 fds each.
await _dispose_unused_adapter(adapter)
del self._failed_platforms[platform]
else:
# Retryable failures retry at the backoff cap forever; never auto-pause them — a
# transient outage must not need `/platform resume`. Same fd-leak dispose as above.
# Retryable failures retry at the cap forever (never auto-pause). Same fd-leak dispose.
backoff = self._bump_reconnect_backoff(
platform, info, attempt, adapter.fatal_error_code,
adapter.fatal_error_message or "failed to reconnect",
@@ -741,31 +684,21 @@ class GatewayAdapterLifecycleMixin:
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:
# Responses rejected while down are owned by this live process (startup recovery cannot claim them).
with _log_suppressed(
logging.DEBUG, "failed-obligation redelivery after %s reconnect failed",
platform.value, exc_info=True,
):
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:
with suppress(Exception):
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.
# A platform offline at startup skipped its restart-interrupted sessions; resume them now.
try:
self._schedule_resume_pending_sessions(platform=platform)
except Exception:
logger.debug(
"resume-pending reschedule after %s reconnect failed", platform.value, exc_info=True
)
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, so a reconnect
@@ -794,21 +727,17 @@ class GatewayAdapterLifecycleMixin:
async def _start_secondary_profile_adapters(self) -> int:
"""Bring up adapters for every non-active profile (multiplex only); returns connected count.
Each profile connects under its own HERMES_HOME + secret scope; credential/listener
collisions are refused here — the only point seeing every profile's credentials together.
"""
Each profile connects under its own HERMES_HOME + secret scope; credential/listener collisions
are refused here — the only point seeing every profile's credentials together."""
from gateway.run import (
MultiplexConfigError, SecondaryPortBindingConfigError, _multiplex_profile_homes
)
if not getattr(self.config, "multiplex_profiles", False):
if not self._multiplex_on():
return 0
try:
from hermes_cli.profiles import get_active_profile_name
except Exception:
return 0
active = get_active_profile_name() or "default"
connected = 0
claimed = self._primary_resource_claims(active)
@@ -825,10 +754,7 @@ class GatewayAdapterLifecycleMixin:
except MultiplexConfigError:
raise
except Exception as e:
logger.error(
"Failed to start adapters for profile '%s': %s", profile_name, e, exc_info=True
)
logger.error("Failed to start adapters for profile '%s': %s", profile_name, e, exc_info=True)
self._record_served_profiles(active, profile_homes)
return connected
@@ -863,33 +789,25 @@ class GatewayAdapterLifecycleMixin:
write_runtime_status(served_profiles=served)
async def _load_secondary_profile_config(self, profile_name: str, profile_home: "Path"):
"""Hydrate + enter ``profile_home``'s scope once; return its gateway config.
Raises ``MultiplexConfigError`` (open dm/group policy) or ``SecondaryPortBindingConfigError``
(the default profile owns the single shared HTTP listener).
"""
"""Hydrate + enter ``profile_home``'s scope once; return its gateway config. Raises
``MultiplexConfigError`` (open dm/group policy) or ``SecondaryPortBindingConfigError`` (the
default profile owns the single shared HTTP listener)."""
from gateway.run import (
MultiplexConfigError, SecondaryPortBindingConfigError, _load_gateway_runtime_config,
_own_policy_open_startup_violation, _profile_runtime_scope,
)
from gateway.config import load_gateway_config
from hermes_cli.env_loader import hydrate_profile_secret_sources
# Hydrate external secret sources (1Password/vault/...) off-loop ONCE: the sync hydration is
# network-bound and would stall every other profile's heartbeat while this one boots.
# Hydrate external secret sources off-loop ONCE: sync hydration would stall every heartbeat.
await asyncio.to_thread(hydrate_profile_secret_sources, profile_home)
with _profile_runtime_scope(profile_home, hydrate_secrets=False):
profile_runtime_cfg = _load_gateway_runtime_config()
from hermes_cli.plugins import discover_plugins
discover_plugins()
# Register this profile's shell hooks / outbound webhooks: start() registers before any
# profile scope exists, so a secondary profile's `hooks:` block would be silently inert.
# This profile's `hooks:` block: start() registered before any profile scope existed.
self._register_config_hooks(
"shell-hook/webhook registration failed for profile '%s'", profile_name, level=logging.WARNING,
)
profile_cfg = load_gateway_config()
violation = _own_policy_open_startup_violation(profile_cfg)
self._snapshot_profile_busy_modes(profile_name, profile_runtime_cfg)
@@ -897,10 +815,8 @@ class GatewayAdapterLifecycleMixin:
raise MultiplexConfigError(
f"Profile '{profile_name}' enables {violation}. "
"Enable GATEWAY_ALLOW_ALL_USERS or the platform allow-all flag "
"for that profile, or change dm_policy/group_policy away from "
"'open'."
"for that profile, or change dm_policy/group_policy away from 'open'."
)
port_binding_platforms = sorted(
platform.value
for platform, platform_config in profile_cfg.platforms.items()
@@ -955,38 +871,35 @@ class GatewayAdapterLifecycleMixin:
"""Create+connect one profile's adapters under its runtime scope."""
from gateway.run import _platform_has_bot_credential, _profile_runtime_scope
profile_cfg = await self._load_secondary_profile_config(profile_name, profile_home)
multiplex = getattr(self.config, "multiplex_profiles", False)
multiplex = self._multiplex_on()
profile_map = self._profile_adapters.setdefault(profile_name, {})
connected = 0
for platform, platform_config in profile_cfg.platforms.items():
if not platform_config.enabled:
continue
# Enabled in the shared YAML but no credential in THIS profile's scope: building an
# adapter anyway would fan one inbound message out across every credential-less profile.
# No credential in THIS profile's scope: an adapter would fan inbound across every such profile.
if multiplex and not _platform_has_bot_credential(platform, platform_config):
logger.info(
"[MULTIPLEX] Profile '%s': skipping %s - no bot credential "
"in this profile's secrets", profile_name, platform.value,
)
continue
# Relay/WhatsApp are shared process-level ingress under multiplex (the active profile's
# one connection fans out by route); a secondary would only retry-loop and stall startup.
# Relay/WhatsApp are shared process-level ingress under multiplex; a secondary would retry-loop.
if multiplex and platform in (Platform.RELAY, Platform.WHATSAPP):
continue
try:
adapter = None
with _log_suppressed(
logging.ERROR, "[MULTIPLEX] Profile '%s': _create_adapter('%s') raised %s", profile_name,
platform.value, exc_info=True,
):
with _profile_runtime_scope(profile_home, hydrate_secrets=False):
adapter = self._create_adapter(platform, platform_config)
except Exception as e:
logger.error(
"[MULTIPLEX] Profile '%s': _create_adapter('%s') raised %s", profile_name,
platform.value, e, exc_info=True,
)
continue
if not adapter:
logger.warning(
"[MULTIPLEX] Profile '%s': skipping platform '%s' - adapter creation returned None",
profile_name, platform.value,
)
if not adapter:
logger.warning(
"[MULTIPLEX] Profile '%s': skipping platform '%s' - adapter creation returned None",
profile_name, platform.value,
)
continue
# Same-token / same-listener conflict detection — refuse a duplicate poll or bind.
credential_claim = self._adapter_credential_claim(platform, adapter)
@@ -1043,12 +956,10 @@ class GatewayAdapterLifecycleMixin:
self, adapter: BasePlatformAdapter, profile_name: str, platform: Platform
) -> None:
"""Install the profile-scoped handlers shared by startup and reconnect."""
# 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.
# Runtime status is process-scoped: key on profile:platform so health shows WHICH secondary failed.
adapter._runtime_status_platform_key = f"{profile_name}:{platform.value}"
# Declare ownership BEFORE any inbound event: adapter-level session keys are derived at
# ingress, before the handler stamps source.profile — without this every secondary bot
# would key into the default profile's `agent:main:` lane.
# Declare ownership BEFORE any inbound event: adapter-level session keys are derived at ingress,
# before the handler stamps source.profile (else every secondary keys into `agent:main:`).
_set_owner = getattr(adapter, "set_owner_profile", None)
if callable(_set_owner):
_set_owner(profile_name)
@@ -1066,19 +977,15 @@ class GatewayAdapterLifecycleMixin:
else self._busy_text_mode
),
)
# Voice transcripts from this bot's channels dispatch through THIS adapter (primary
# wiring lives at connect time).
# Voice transcripts from this bot's channels dispatch through THIS adapter.
self._bind_voice_input_callback(adapter)
# Secondary adapters always carry the profile they serve so prune paths namespace topic
# bindings correctly under multiplex.
# Secondary adapters carry their profile so prune paths namespace topic bindings correctly.
adapter._hermes_profile_name = profile_name
async def _secondary_reconnect_attempt(self, profile_name: str, platform: Platform):
"""One scoped attempt to rebuild+connect a secondary adapter → ``(adapter, success)``.
``(None, None)`` = give up for good (disabled, credential removed, adapter unavailable).
Caller tears down a RETURNED adapter; one whose configure/connect raised is torn down here.
"""
"""One scoped attempt to rebuild+connect a secondary adapter → ``(adapter, success)``;
``(None, None)`` = give up for good (disabled, credential removed, adapter unavailable). Caller
tears down a RETURNED adapter; one whose configure/connect raised is torn down here."""
from gateway.run import _platform_has_bot_credential, _profile_runtime_scope
# Lazy + per-attempt: keeps test monkeypatches on these modules live.
from hermes_cli.profiles import get_profile_dir
@@ -1136,34 +1043,29 @@ class GatewayAdapterLifecycleMixin:
platform, profile=profile_name
)
return
# Not installed: a newer reconnect won the slot mid-connect (never replace it),
# shutdown began mid-connect (never republish after the registry drained), or
# connect failed. Release partial resources; stop only for a non-retryable fatal.
# Not installed (newer reconnect won the slot, shutdown began, or connect failed):
# release partial resources; stop only for a non-retryable fatal.
await self._safe_adapter_disconnect(adapter, platform)
if success or (
getattr(adapter, "has_fatal_error", False)
and not getattr(adapter, "fatal_error_retryable", True)
):
return
except asyncio.CancelledError:
if adapter is not None:
await self._safe_adapter_disconnect(adapter, platform)
raise
except Exception:
except BaseException as exc:
if adapter is not None:
await self._safe_adapter_disconnect(adapter, platform)
if not isinstance(exc, Exception):
raise # CancelledError (and other BaseExceptions) propagate after release
logger.debug(
"Secondary %s reconnect attempt failed (profile: %s)", platform.value,
profile_name, exc_info=True,
)
if not self._running:
return
attempts += 1
backoff = _reconnect_backoff(attempts)
logger.info(
"Secondary %s reconnect retry in %ds (profile: %s)", platform.value, backoff,
profile_name,
"Secondary %s reconnect retry in %ds (profile: %s)", platform.value, backoff, profile_name
)
await asyncio.sleep(backoff)
finally:
@@ -1185,8 +1087,7 @@ class GatewayAdapterLifecycleMixin:
if not getattr(adapter, "fatal_error_retryable", True):
return
if is_global_startup_conflict(getattr(adapter, "fatal_error_code", None)):
# A live foreign holder of this profile's token is an ownership conflict, not a blip:
# park it fatal (like ``duplicate_credential``) instead of retry-storming.
# A live foreign token holder is an ownership conflict, not a blip: park it fatal.
logger.error(
"[MULTIPLEX] Profile '%s': %s credential is held by another "
"gateway (%s) — parked, not retried. %s", profile_name, platform.value,
@@ -1199,15 +1100,14 @@ class GatewayAdapterLifecycleMixin:
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.
# A raise here would die as an unretrieved-task exception logged only at GC; surface it.
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:
if self._running: # fast path (also the only path for bare runners without _shutdown_event)
_handoff()
return
# Poll (startup completion has no event); bounded so a wedged startup cannot spin.
@@ -1281,11 +1181,9 @@ class GatewayAdapterLifecycleMixin:
@staticmethod
def _stamp_event_profile(event, profile_name: str) -> None:
"""Best-effort: stamp ``source.profile`` on an inbound event that has none yet."""
try:
with suppress(Exception):
if getattr(event, "source", None) is not None and not event.source.profile:
event.source.profile = profile_name
except Exception:
pass
def _make_profile_message_handler(self, profile_name: str):
"""Message handler that stamps source.profile, then delegates under the profile scope
@@ -1304,8 +1202,7 @@ class GatewayAdapterLifecycleMixin:
"""Stamp an owning adapter's profile before resolving busy policy."""
async def _handler(event, _session_key):
self._stamp_event_profile(event, profile_name)
routed_session_key = self._session_key_for_source(event.source)
return await self._handle_active_session_busy_message(event, routed_session_key)
return await self._handle_active_session_busy_message(event, self._session_key_for_source(event.source))
return _handler
@@ -1346,19 +1243,18 @@ class GatewayAdapterLifecycleMixin:
def _primary_message_handler(self):
"""Return the correctly scoped handler for a primary adapter."""
if getattr(self.config, "multiplex_profiles", False):
return self._make_default_profile_message_handler()
return self._handle_message
return self._make_default_profile_message_handler() if self._multiplex_on() else self._handle_message
def _multiplex_on(self) -> bool:
return bool(getattr(self.config, "multiplex_profiles", False))
async def _handle_gateway_platform_event(self, event: dict, source) -> None:
"""Authorize and publish one normalized adapter event to plugin hooks."""
try:
# Observer failures must never break the adapter's update loop.
with _log_suppressed(logging.DEBUG, "gateway_platform_event hook dispatch failed", exc_info=True):
from hermes_cli.lifecycle import has_hook, invoke_hook
if has_hook("gateway_platform_event") and self._is_user_authorized_for_source(source):
invoke_hook("gateway_platform_event", **event)
except Exception:
# Observer failures must never break the adapter's update loop.
logger.debug("gateway_platform_event hook dispatch failed", exc_info=True)
def _make_profile_platform_event_handler(self, profile_name: str):
"""Bind platform-event auth and hook dispatch to one multiplex profile."""
@@ -1386,7 +1282,7 @@ class GatewayAdapterLifecycleMixin:
return _handler
def _primary_platform_event_handler(self):
if getattr(self.config, "multiplex_profiles", False):
if self._multiplex_on():
return self._make_default_profile_platform_event_handler()
return self._handle_gateway_platform_event
@@ -1414,8 +1310,7 @@ class GatewayAdapterLifecycleMixin:
def _adapter_credential_fingerprint(adapter: Any) -> Optional[str]:
"""Salted, log-safe hash of an adapter's credential; None when none is discoverable
(conflict detection is then skipped)."""
# Many adapters (e.g. Discord) keep the token on `config`; without that fallback the
# same-token check is silently skipped and every profile polls the same bot.
# Many adapters (Discord) keep the token on `config`; without that fallback the check is skipped.
candidates = [
(adapter, attr) for attr in (
"token", "bot_token", "_token", "api_token", "_bot_token",
@@ -1447,7 +1342,6 @@ class GatewayAdapterLifecycleMixin:
config.extra.setdefault(
"thread_sessions_per_user", getattr(self.config, "thread_sessions_per_user", False)
)
with _log_suppressed(logging.DEBUG, "Platform registry lookup for '%s' failed: %s", platform.value):
from gateway.platform_registry import platform_registry
if platform_registry.is_registered(platform.value):
@@ -1468,8 +1362,7 @@ class GatewayAdapterLifecycleMixin:
its scope; for the shared primary (None) the routed profile is stamped so its pairing store
is consulted while allowlist reads stay under the transport home."""
from gateway.run import get_hermes_home
multiplex = bool(getattr(self.config, "multiplex_profiles", False))
transport_home = (Path(get_hermes_home()) if multiplex and profile_name is None else None)
transport_home = Path(get_hermes_home()) if self._multiplex_on() and profile_name is None else None
def check(
user_id: str, chat_type: Optional[str] = None, chat_id: Optional[str] = None, *,
@@ -1481,8 +1374,7 @@ class GatewayAdapterLifecycleMixin:
platform=platform, chat_id=chat_id or "", chat_type=chat_type or "group",
user_id=user_id, thread_id=thread_id, is_bot=bool(is_bot), profile=profile_name,
)
# Same transport provenance ``build_source`` retains, so adapter-level policy reads
# resolve the receiving adapter even once the routed profile is stamped below.
# Same transport provenance as ``build_source``, so policy reads resolve the receiving adapter.
registry = (
(getattr(self, "_profile_adapters", None) or {}).get(profile_name)
if profile_name else getattr(self, "adapters", None)
gateway/run_notifications.py
+97 -238
View File
@@ -39,8 +39,7 @@ _DURABLE_CLAIM_OPS = {
class GatewayNotificationsMixin:
"""Process/completion/update notifications, media delivery and async-delegation delivery for GatewayRunner."""
# Routing-complete coalescing keys: process completions (short-window fan-in) and async
# delegations (originating session + parent session + route).
# Coalescing keys: process completions (short-window fan-in) and async delegations (+ parent session).
_COMPLETION_BATCH_KEY_FIELDS = ("session_key", "platform", "chat_type", "chat_id", "thread_id", "user_id")
_ASYNC_GROUP_KEY_FIELDS = ("session_key", "parent_session_id", *_COMPLETION_BATCH_KEY_FIELDS[1:])
@@ -94,7 +93,6 @@ class GatewayNotificationsMixin:
adapter = self._adapter_for_source(source)
if not adapter:
return
config = getattr(self, "config", None)
chat_id = getattr(source, "chat_id", None)
if config and getattr(source, "platform", None) == Platform.SLACK and _is_slack_ignored_channel(config, chat_id):
@@ -106,16 +104,13 @@ class GatewayNotificationsMixin:
)
metadata = self._thread_metadata_for_source(source)
if notice_delivery == "private" and getattr(source, "user_id", None):
try:
with _log_suppressed(
logging.DEBUG, "[%s] send_private_notice failed, falling back to public",
getattr(source, "platform", "?"), exc_info=True,
):
result = await adapter.send_private_notice(source.chat_id, source.user_id, content, metadata=metadata)
if getattr(result, "success", False):
return
except Exception:
logger.debug(
"[%s] send_private_notice failed, falling back to public",
getattr(source, "platform", "?"), exc_info=True,
)
await adapter.send(source.chat_id, content, metadata=metadata)
async def _resolve_compression_lineage_target(
@@ -127,14 +122,12 @@ class GatewayNotificationsMixin:
except Exception:
logger.debug("Async-delegation compression-tip lookup failed for %s", pinned_session_id, exc_info=True)
target_session_id = None
if not target_session_id or target_session_id == pinned_session_id:
logger.warning(
"Async-delegation completion pinned to compressed session %s "
"without a continuation; dropping injection.", pinned_session_id,
)
return None
try:
tip_row = await session_db.get_session(target_session_id)
except Exception:
@@ -145,11 +138,9 @@ class GatewayNotificationsMixin:
target_session_id, "unknown" if tip_row is None else "ended",
)
return None
route_owns_lineage = session_entry.session_id in {pinned_session_id, target_session_id}
if not route_owns_lineage:
# A delegation may survive several compression rotations: accept an intermediate stale
# route only when its own verified compression tip is the same live target.
# Across several rotations, accept a stale route only when its own tip is the same live target.
try:
route_row = await session_db.get_session(session_entry.session_id)
route_tip = (
@@ -162,7 +153,6 @@ class GatewayNotificationsMixin:
except Exception:
route_tip = None
route_owns_lineage = route_tip == target_session_id
if not route_owns_lineage:
logger.warning(
"Async-delegation completion for compression lineage %s -> %s "
@@ -189,20 +179,17 @@ class GatewayNotificationsMixin:
"dropping injection (#55578 fail-closed)."
)
return None
pinned_row = None
try:
pinned_row = await session_db.get_session(pinned_session_id)
except Exception:
logger.debug("Async-delegation parent lookup failed for %s", pinned_session_id, exc_info=True)
if pinned_row is None:
logger.warning(
"Async-delegation completion has unknown spawning session %s; "
"dropping injection (#55578 fail-closed).", pinned_session_id,
)
return None
target_session_id = pinned_session_id
follows_compression = False
if pinned_row.get("ended_at"):
@@ -215,26 +202,22 @@ class GatewayNotificationsMixin:
)
return None
if _end_reason != "compression":
# Idle/timeout/lifecycle end (scale-to-zero norm): the chat route is still valid and
# ``session_entry`` is its current session, so deliver here rather than drop — otherwise
# the row is acked at adapter acceptance then silently lost.
# Idle/timeout end (scale-to-zero norm): the chat route is still valid, so deliver to its
# current session rather than drop (the row would be acked then silently lost).
logger.info(
"Async-delegation completion pinned to %s-ended session %s; "
"retargeting to the chat's current session %s.",
_end_reason or "idle", pinned_session_id, session_entry.session_id,
)
return session_entry
follows_compression = True
target_session_id = await self._resolve_compression_lineage_target(
session_db, session_entry, pinned_session_id,
)
if target_session_id is None:
return None
if target_session_id == session_entry.session_id:
return session_entry
prior_session_id = session_entry.session_id
if follows_compression:
switched = await self.async_session_store.advance_compression_session(
@@ -245,50 +228,38 @@ class GatewayNotificationsMixin:
if switched is None:
logger.warning(
"Async-delegation completion could not bind routing key %s to "
"owning session %s; dropping injection.",
session_entry.session_key, target_session_id,
"owning session %s; dropping injection.", session_entry.session_key, target_session_id,
)
return None
logger.info(
"Pinned async-delegation completion to owning session %s "
"(was %s) for routing key %s (#57498)",
"Pinned async-delegation completion to owning session %s (was %s) for routing key %s (#57498)",
target_session_id, prior_session_id, session_entry.session_key,
)
return switched
async def _deliver_media_from_response(
self, response: str, event: MessageEvent, adapter,
thread_metadata: Optional[Dict[str, Any]] = None,
self, response: str, event: MessageEvent, adapter, thread_metadata: Optional[Dict[str, Any]] = None
) -> None:
"""Extract explicit MEDIA: tags from an already-streamed response and deliver them.
The text is already delivered; this only handles file attachments. Unlike the non-streaming
path in ``gateway/platforms/base.py`` this rescan is EXPLICIT-ONLY: a bare local path in a
streamed reply was either shown as text or is stale inspected content, and promoting it sent
files the model never asked to deliver. MEDIA tags are NOT deduped against prior turns: a
MEDIA: directive in the final reply is a deliberate attach (incl. user-requested resends);
stale auto-appended tags are deduped upstream (_collect_auto_append_media_tags).
"""
"""Deliver explicit MEDIA: tags from an already-streamed response (text already delivered).
EXPLICIT-ONLY, unlike the non-streaming path in ``gateway/platforms/base.py``: a bare local
path in a streamed reply is shown text or stale inspected content, and promoting it sent
files the model never asked for. MEDIA tags are NOT deduped against prior turns (a final-reply
directive is a deliberate attach); stale auto-appended tags are deduped upstream."""
from urllib.parse import quote as _quote
with _log_suppressed(logging.WARNING, "Post-stream media extraction failed: %s"):
# Capture [[as_document]] before extract_media strips it: image files then go through
# send_document (preserving bytes), not send_multiple_images (Telegram recompresses).
# Capture [[as_document]] before extract_media strips it: images then go via send_document.
force_document_attachments = "[[as_document]]" in response
from gateway.platforms.base import BasePlatformAdapter, should_send_media_as_audio
media_files, cleaned = adapter.extract_media(response)
media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files)
# Strip image URLs for parity with the non-streaming chain; no extract_local_files here.
# Strip image URLs (parity with the non-streaming chain); no extract_local_files here.
adapter.extract_images(cleaned)
_thread_meta = (
dict(thread_metadata)
if thread_metadata is not None
else self._thread_metadata_for_source(event.source, self._reply_anchor_for_event(event))
)
chat_id = event.source.chat_id
# Images go out as one batch (e.g. Signal's multi-attachment RPC) unless [[as_document]].
def _is_photo(media_path: str, is_voice: bool) -> bool:
ext = Path(media_path).suffix.lower()
@@ -296,14 +267,12 @@ class GatewayNotificationsMixin:
image_paths = [p for p, v in media_files if _is_photo(p, v)]
non_image_media = [(p, v) for p, v in media_files if not _is_photo(p, v)]
if image_paths:
try:
images = [(f"file://{_quote(p)}", "") for p in image_paths]
await adapter.send_multiple_images(chat_id=chat_id, images=images, metadata=_thread_meta)
except Exception as e:
logger.warning("[%s] Post-stream image batch delivery failed: %s", adapter.name, e)
for media_path, is_voice in non_image_media:
try:
ext = Path(media_path).suffix.lower()
@@ -329,9 +298,8 @@ class GatewayNotificationsMixin:
if not text_already_delivered:
text_content = _strip_response_attachments_for_direct_send(response, adapter)
if text_content:
# Reconcile-by-edit first: a stream-sealed message whose recorded payload didn't confirm
# the final (post-stream mutation) already carries most of the answer; a plain send here
# would duplicate it.
# Reconcile-by-edit first: a stream-sealed message already carries most of the answer;
# a plain send here would duplicate it.
_reconciled = False
_sc_msg_id = getattr(stream_consumer, "message_id", None)
if (
@@ -353,21 +321,20 @@ class GatewayNotificationsMixin:
logger.debug("Queued-lane reconcile edit failed (%s); falling back to send.", _qe)
if not _reconciled:
await adapter.send(source.chat_id, text_content, metadata=metadata)
# Failed turns deliver their (normalized failure) text but must not upload attachments as if
# they succeeded — mirrors the ``not agent_result.get("failed")`` completed-turn guard.
if not deliver_media:
return
synthetic_event = MessageEvent(text="", source=source, message_id=event_message_id)
await self._deliver_media_from_response(response, synthetic_event, adapter, thread_metadata=metadata)
await self._deliver_media_from_response(
response, MessageEvent(text="", source=source, message_id=event_message_id), adapter,
thread_metadata=metadata,
)
def _schedule_update_notification_watch(self) -> None:
"""Ensure a background task is watching for update completion."""
existing_task = getattr(self, "_update_notification_task", None)
if existing_task and not existing_task.done():
return
try:
self._update_notification_task = asyncio.create_task(self._watch_update_progress())
except RuntimeError:
@@ -378,8 +345,7 @@ class GatewayNotificationsMixin:
from gateway.run import _hermes_home
return cls._UpdatePaths(
pending=_hermes_home / ".update_pending.json",
claimed=_hermes_home / ".update_pending.claimed.json",
output=_hermes_home / ".update_output.txt",
claimed=_hermes_home / ".update_pending.claimed.json", output=_hermes_home / ".update_output.txt",
exit_code=_hermes_home / ".update_exit_code",
prompt=_hermes_home / ".update_prompt.json", response=_hermes_home / ".update_response",
)
@@ -389,7 +355,7 @@ class GatewayNotificationsMixin:
for path in (paths.claimed, paths.pending):
if not path.exists():
continue
try:
with suppress(Exception):
pending = json.loads(path.read_text(encoding="utf-8"))
platform_str = pending.get("platform")
chat_id = pending.get("chat_id")
@@ -405,8 +371,6 @@ class GatewayNotificationsMixin:
return self._UpdateTarget(
adapter, chat_id, session_key or f"{platform_str}:{chat_id}", metadata, platform,
)
except Exception:
pass
return None
def _pending_marker_metadata(self, platform, chat_id, data: dict, adapter):
@@ -419,8 +383,7 @@ class GatewayNotificationsMixin:
async def _watch_update_completion_only(self, paths: "_UpdatePaths", deadline: float, poll_interval: float) -> None:
"""Fallback when no adapter/chat can be resolved: wait for the exit code, then notify."""
logger.warning("Update watcher: cannot resolve adapter/chat_id, falling back to completion-only")
# Keep polling until _send_update_notification actually delivers (True) — it re-resolves the
# adapter each call and returns False (markers kept) while the platform is still reconnecting.
# Poll until _send_update_notification delivers (it returns False while the platform reconnects).
loop = asyncio.get_running_loop()
while paths.any_pending() and loop.time() < deadline:
if paths.exit_code.exists() and await self._send_update_notification():
@@ -474,8 +437,7 @@ class GatewayNotificationsMixin:
f"⚕ **Update needs your input:**\n\n{prompt_text}{default_hint}\n\n"
f"Reply `{_p}approve` (yes) or `{_p}deny` (no), or type your answer directly."
)
# Keep the prompt marker on disk until the user answers so a watcher after a mid-prompt
# gateway restart can recover by re-forwarding it.
# Keep the prompt marker on disk until answered so a restarted watcher can re-forward it.
self._session_state(target.session_key).persistent.update_prompt_pending = True
logger.info("Forwarded update prompt to %s: %s", target.session_key, prompt_text[:80])
@@ -496,13 +458,11 @@ class GatewayNotificationsMixin:
paths = self._update_paths()
loop = asyncio.get_running_loop()
deadline = loop.time() + timeout
target = self._resolve_update_target(paths)
if target is None:
await self._watch_update_completion_only(paths, deadline, poll_interval)
return
session_key = target.session_key
bytes_sent = 0
last_stream_time = loop.time()
buffer = ""
@@ -517,11 +477,9 @@ class GatewayNotificationsMixin:
def _read_new_output() -> None:
nonlocal buffer, bytes_sent
if paths.output.exists():
try:
with suppress(OSError):
chunk, bytes_sent = self._read_update_output_since(paths.output, bytes_sent)
buffer += chunk
except OSError:
pass
while loop.time() < deadline:
if paths.exit_code.exists():
@@ -536,17 +494,13 @@ class GatewayNotificationsMixin:
logger.info("Update finished (exit=%s), notified %s", exit_code, session_key)
self._clear_update_markers(paths, session_key)
return
_read_new_output()
if buffer.strip() and (loop.time() - last_stream_time) >= stream_interval:
await _flush_buffer()
# Forward a prompt only when none is still awaiting a response; otherwise the watcher
# would re-read the same .update_prompt.json every poll and spam duplicate prompts.
# Forward a prompt only when none is pending, else every poll re-forwards the same prompt.
_pending_state = self._peek_session_state(session_key) if session_key else None
if (
paths.prompt.exists() and session_key
and not (_pending_state is not None and _pending_state.persistent.update_prompt_pending)
if paths.prompt.exists() and session_key and not getattr(
getattr(_pending_state, "persistent", None), "update_prompt_pending", False
):
try:
prompt_data = json.loads(paths.prompt.read_text(encoding="utf-8"))
@@ -556,9 +510,7 @@ class GatewayNotificationsMixin:
await self._forward_update_prompt(target, prompt_text, prompt_data.get("default", ""))
except (json.JSONDecodeError, OSError) as e:
logger.debug("Failed to read update prompt: %s", e)
await asyncio.sleep(poll_interval)
if not paths.exit_code.exists():
logger.warning("Update watcher timed out after %.0fs", timeout)
paths.exit_code.write_text("124", encoding="utf-8")
@@ -576,7 +528,6 @@ class GatewayNotificationsMixin:
paths = self._update_paths()
if not paths.any_pending():
return False
cleanup = True
active_pending_path = paths.claimed
@@ -597,23 +548,18 @@ class GatewayNotificationsMixin:
return True
elif not paths.claimed.exists():
return True
pending = json.loads(paths.claimed.read_text(encoding="utf-8"))
platform_str = pending.get("platform")
chat_id = pending.get("chat_id")
if not paths.exit_code.exists():
return _defer("Update notification deferred: update still running")
exit_code = self._update_exit_code(paths)
output = paths.output.read_bytes().decode("utf-8", errors="replace") if paths.output.exists() else ""
platform = Platform(platform_str)
adapter = self.adapters.get(platform)
if chat_id and not adapter:
# The update finished, but the target platform has not reconnected yet (common right
# after the restart that `hermes update` triggers). A definitive skip would delete the
# markers and silently lose the notification; preserve them for a later retry.
# Target platform not reconnected yet (common right after the update's restart): keep the
# markers for a later retry instead of silently losing the notification.
return _defer("Update notification deferred: %s adapter not connected yet", platform_str)
if chat_id:
metadata = self._pending_marker_metadata(platform, chat_id, pending, adapter)
@@ -624,10 +570,11 @@ class GatewayNotificationsMixin:
output = "…" + output[-3500:]
status = "✅ Hermes update finished." if exit_code == 0 else "❌ Hermes update failed."
msg = f"{status}\n\n```\n{output}\n```"
elif exit_code == 0:
msg = "✅ Hermes update finished successfully."
else:
msg = "❌ Hermes update failed. Check the gateway logs or run `hermes update` manually for details."
msg = (
"✅ Hermes update finished successfully." if exit_code == 0 else
"❌ Hermes update failed. Check the gateway logs or run `hermes update` manually for details."
)
await adapter.send(chat_id, msg, metadata=_non_conversational_metadata(metadata, platform=platform))
logger.info("Sent post-update notification to %s:%s (exit=%s)", platform_str, chat_id, exit_code)
except Exception as e:
@@ -636,7 +583,6 @@ class GatewayNotificationsMixin:
if cleanup:
for p in (active_pending_path, paths.claimed, paths.output, paths.exit_code):
p.unlink(missing_ok=True)
return True
async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[str]]]:
@@ -645,30 +591,24 @@ class GatewayNotificationsMixin:
notify_path = _hermes_home / ".restart_notify.json"
if not notify_path.exists():
return None
try:
data = json.loads(notify_path.read_text(encoding="utf-8"))
platform_str = data.get("platform")
chat_id = data.get("chat_id")
thread_id = data.get("thread_id")
if not platform_str or not chat_id:
return None
platform = Platform(platform_str)
transport = resolve_delivery_transport(platform, self.config, self.adapters)
if transport is None:
logger.debug("Restart notification skipped: no live transport for %s", platform_str)
return None
platform_cfg = self.config.platforms.get(platform)
if platform_cfg is not None and not platform_cfg.gateway_restart_notification:
logger.info(
"Restart notification suppressed: %s has gateway_restart_notification=false",
platform_str,
"Restart notification suppressed: %s has gateway_restart_notification=false", platform_str
)
return None
metadata = self._pending_marker_metadata(platform, chat_id, data, transport.adapter)
if data.get("delivered_via_upstream_relay") is True:
metadata = dict(metadata or {})
@@ -686,7 +626,6 @@ class GatewayNotificationsMixin:
"Restart notification to %s:%s was not delivered: %s", platform_str, chat_id, _send_error(result),
)
return None
logger.info("Sent restart notification to %s:%s", platform_str, chat_id)
return str(platform_str), str(chat_id), str(thread_id) if thread_id else None
except Exception as e:
@@ -742,7 +681,6 @@ class GatewayNotificationsMixin:
delivered: set[tuple[str, str, Optional[str]]] = set()
skipped = skip_targets or set()
message = "♻️ Gateway online — Hermes is back and ready."
for platform, platform_cfg, home, transport in self._home_channel_transports():
if not platform_cfg.gateway_restart_notification:
logger.info(
@@ -750,17 +688,14 @@ class GatewayNotificationsMixin:
platform.value,
)
continue
target = _notice_target_key(platform.value, home.chat_id, home.thread_id)
if target in skipped or target in delivered:
continue
if await self._send_home_channel_message(
platform, home, transport, message, "Home-channel startup notification failed for %s:%s: %s",
):
delivered.add(target)
logger.info("Sent home-channel startup notification to %s:%s", platform.value, home.chat_id)
return delivered
async def _send_session_db_warning_notifications(self) -> None:
@@ -773,24 +708,20 @@ class GatewayNotificationsMixin:
error = getattr(self, "_session_db_init_error", None)
if not error:
return
from hermes_state import classify_persistence_error, format_session_db_unavailable
if classify_persistence_error(error) == "corrupt":
message = (
"⚠️ Session database corruption detected. Messages may not be "
"persisted. Recovery options:\n1. Run `hermes doctor --fix`\n"
"2. Salvage with: sqlite3 ~/.hermes/state.db \".recover\" (then replace state.db)\n"
"3. Restore from a backup in ~/.hermes/backups/\n"
"Run `hermes doctor` for sanitized diagnostics."
"⚠️ Session database corruption detected. Messages may not be persisted. Recovery "
"options:\n1. Run `hermes doctor --fix`\n2. Salvage with: sqlite3 ~/.hermes/state.db "
"\".recover\" (then replace state.db)\n3. Restore from a backup in ~/.hermes/backups/\nRun "
"`hermes doctor` for sanitized diagnostics."
)
else:
message = (
f"⚠️ Session database unavailable — messages may not be persisted. "
f"{format_session_db_unavailable()}\nRun `hermes doctor` for diagnostics."
)
logger.warning("Broadcasting state.db failure warning to home channels: %s", error)
for platform, _platform_cfg, home, transport in self._home_channel_transports():
await self._send_home_channel_message(
platform, home, transport, message, "state.db warning notification failed for %s:%s: %s",
@@ -804,7 +735,6 @@ class GatewayNotificationsMixin:
from gateway.run import _parse_session_key
session_key = str(evt.get("session_key") or "").strip()
derived = {}
if session_key:
try:
self.session_store._ensure_loaded()
@@ -813,13 +743,10 @@ class GatewayNotificationsMixin:
return entry.origin
except Exception as exc:
logger.debug("Synthetic process-event session-store lookup failed for %s: %s", session_key, exc)
cached_source = self._get_cached_session_source(session_key)
if cached_source is not None:
return cached_source
derived = _parse_session_key(session_key) or {}
platform_name = str(evt.get("platform") or derived.get("platform") or "").strip().lower()
chat_type = str(evt.get("chat_type") or derived.get("chat_type") or "").strip().lower()
chat_id = str(evt.get("chat_id") or derived.get("chat_id") or "").strip()
@@ -830,11 +757,9 @@ class GatewayNotificationsMixin:
session_key, platform_name, chat_type, chat_id, evt.get("type", "?"),
)
return None
try:
platform = Platform(platform_name)
# Reject arbitrary strings (dynamic pseudo-members): built-ins are always valid, plugin
# platforms must be registered in the platform registry.
# Reject dynamic pseudo-members: plugin platforms must be registered.
if platform.value not in _BUILTIN_PLATFORM_VALUES:
try:
from gateway.platform_registry import platform_registry
@@ -851,15 +776,12 @@ class GatewayNotificationsMixin:
scope_id = _opt("scope_id")
if scope_id is None and chat_type not in ("dm", "thread"):
# Reconstructed (non-persisted) source for a scoped chat with no scope discriminator: a
# relay connector's fail-closed tenant guard may decline the reply unless user_id resolves
# it. Don't fail — DMs/author-bound chats still route and native adapters need no
# scope_id — but warn so a post-restart egress decline isn't silent.
# Reconstructed scoped-chat source without scope_id: a relay connector's tenant guard may
# decline the reply. Warn, don't fail (native adapters need no scope_id).
logger.warning(
"Synthetic event source for %s chat=%s (%s) reconstructed "
"without scope_id; scoped relay egress may be declined by "
"the connector's tenant guard (user_id fallback only).",
platform_name, chat_id, chat_type,
"the connector's tenant guard (user_id fallback only).", platform_name, chat_id, chat_type,
)
return SessionSource(
platform=platform, chat_id=chat_id, chat_type=chat_type, thread_id=_opt("thread_id"),
@@ -876,7 +798,6 @@ class GatewayNotificationsMixin:
watch_events = _drain_gateway_watch_events(completion_queue)
if self._load_background_notifications_mode() == "off":
return
for evt in watch_events:
synth_text = _format_gateway_process_notification(evt)
if not synth_text:
@@ -916,15 +837,9 @@ class GatewayNotificationsMixin:
return False
def _resolve_injection_adapter(self, platform_name: str):
"""Adapter for a synthetic-event platform: transport resolver first, literal scan as fallback.
Alias-aware resolution (relay-plane): one adapter under Platform.RELAY fronts N logical
platforms, so a literal ``p.value == platform_name`` scan misses "slack" and drops the
completion as "no gateway route". Native adapter wins; relay is eligible only when it
advertises fronting the logical platform. The legacy literal scan is still correct for
native adapters and keeps minimal runner stubs (tests) / exotic platform strings working
when the resolver can't run.
"""
"""Adapter for a synthetic-event platform: alias-aware transport resolver first (one
Platform.RELAY adapter fronts N logical platforms; native wins), literal ``p.value`` scan as
fallback for minimal runner stubs / exotic platform strings when the resolver can't run."""
from gateway.run import resolve_delivery_transport
try:
_transport = resolve_delivery_transport(Platform(platform_name), self.config, self.adapters)
@@ -945,8 +860,7 @@ class GatewayNotificationsMixin:
from gateway.wake import adapter_supports_push
source = await asyncio.to_thread(self._build_process_event_source, evt)
if not source:
# API-server sessions bind the RAW X-Hermes-Session-Id key (_bind_api_server_session), not a
# structured ``agent:main:...`` key, so _build_process_event_source returned None above.
# API-server sessions bind the RAW X-Hermes-Session-Id key, not a structured ``agent:...`` key.
raw_sid = str(evt.get("origin_session_id") or "").strip()
_sk = str(evt.get("session_key") or "").strip()
if not raw_sid and _sk and _parse_session_key(_sk) is None:
@@ -970,9 +884,8 @@ class GatewayNotificationsMixin:
if not adapter:
return None
if not adapter_supports_push(adapter):
# Non-push adapter (api_server) resolved WITH routing metadata: its chat_id is the raw session
# id (_bind_api_server_session binds chat_id = session_id), so handle_message would run the
# wake under a build_session_key() key that never matches the raw session — self-post.
# Non-push adapter (api_server): its chat_id IS the raw session id, so handle_message would
# key the wake under a build_session_key() that never matches — self-post instead.
raw_sid = str(evt.get("origin_session_id") or "").strip() or str(source.chat_id or "")
return await self._self_post_api_server(adapter, synth_text, raw_sid, evt)
try:
@@ -988,9 +901,8 @@ class GatewayNotificationsMixin:
"Watch pattern notification — injecting for %s chat=%s thread=%s",
platform_name, source.chat_id, source.thread_id,
)
# Relay-plane egress priming: a synthetic turn injected right after a restart reaches a relay
# adapter whose per-chat routing caches are cold (they warm only on inbound), so its replies
# egress without tenant discriminators and the connector's fail-closed guard declines them.
# Relay egress priming: post-restart routing caches are cold (they warm only on inbound), so
# replies would egress without tenant discriminators and be declined by the connector.
_prime = getattr(adapter, "prime_routing_cache", None)
if callable(_prime):
_prime(synth_event)
@@ -1042,15 +954,10 @@ class GatewayNotificationsMixin:
return seen
async def _classify_completion_target(self, parent_session_id: str) -> str:
"""Classify an async-completion delivery target before adapter acceptance.
- ``"deliver"``: spawning session live (or compression-rotated with a live continuation);
proves deliverability only, the resolver still retargets.
- ``"terminal"``: parent gone for good (unknown / explicit user boundary like /new); drop
the durable row rather than falsely ack or replay forever.
- ``"retry"``: transient uncertainty (DB unavailable, rotation mid-flight); release the
claim for a later consumer; the attempt cap bounds churn.
"""
"""Classify an async-completion target before adapter acceptance: ``"deliver"`` (spawning
session live or compression-rotated with a live continuation; the resolver still retargets),
``"terminal"`` (parent gone for good — unknown / user boundary like /new; drop the durable row
rather than falsely ack), ``"retry"`` (DB unavailable / rotation mid-flight; release the claim)."""
from gateway.run import _USER_BOUNDARY_END_REASONS
session_db = getattr(self, "_session_db", None)
if session_db is None:
@@ -1066,15 +973,13 @@ class GatewayNotificationsMixin:
return "deliver"
end_reason = str(parent.get("end_reason") or "")
if end_reason != "compression":
# Only a USER-closed thread of work (/new, user_exit, session_switch) is unreachable;
# idle/timeout ends are normal on scale-to-zero relays — the chat stays routable and the
# resolver retargets. Boundary set shared with the resolver: no drift.
# Only a USER-closed session (/new, user_exit, session_switch) is unreachable; idle/timeout
# ends stay routable and the resolver retargets. Boundary set shared with the resolver.
return "terminal" if end_reason in _USER_BOUNDARY_END_REASONS else "deliver"
try:
tip_session_id = await session_db.get_compression_tip(parent_session_id)
if not tip_session_id or tip_session_id == parent_session_id:
# Rotation caught mid-flight: parent is compression-ended but its continuation isn't
# visible yet. Retry, don't drop.
# Rotation mid-flight: continuation not visible yet. Retry, don't drop.
return "retry"
tip = await session_db.get_session(tip_session_id)
except Exception:
@@ -1118,10 +1023,8 @@ class GatewayNotificationsMixin:
return claim
elif evt_type != "completion":
return claim
# Background completions carry only session_key, so after /new the OLD session's notification
# would land in the chat's NEW session. Stamped events get the same pre-flight as async
# delegations; unstamped ones deliver.
# would land in the NEW one. Stamped events get the async-delegation pre-flight; unstamped deliver.
parent_session_id = str(evt.get("parent_session_id") or "").strip()
if not parent_session_id:
return claim
@@ -1139,14 +1042,12 @@ class GatewayNotificationsMixin:
logger.warning(
"Background process %s completion targets "
"permanently-gone session %s (user boundary such as "
"/new); dropping notification (output remains "
"available via process(action='log')).",
"/new); dropping notification (output remains available via process(action='log')).",
evt.get("session_id") or "<unknown>", parent_session_id,
)
claim.proceed = False
elif verdict == "retry":
# Transient uncertainty (session DB down / compression rotation mid-flight): tell the
# watcher to re-poll and retry rather than drop or misroute the result.
# Transient uncertainty: tell the watcher to re-poll rather than drop or misroute.
if claim.claim_id:
self._settle_durable_claim("release", claim.delegation_id, claim.claim_id)
claim.proceed, claim.early_result = False, False
@@ -1164,20 +1065,16 @@ class GatewayNotificationsMixin:
return claim.early_result
if identity is not None and self._completion_identity_seen(identity, claim=True):
return None
accepted = False
try:
injection_result = await self._inject_watch_notification(synth_text, evt)
if injection_result is not True:
return injection_result
accepted = True
if identity is not None:
with self._completion_delivery_lock:
self._mark_completions_delivered_locked((identity,))
# The durable async-delegation row is the authoritative replay state — ack it after
# adapter acceptance; no parallel ledger here.
# The durable row is the authoritative replay state — ack it after adapter acceptance.
if claim.claim_id:
try:
from tools.async_delegation import complete_completion_delivery
@@ -1210,8 +1107,7 @@ class GatewayNotificationsMixin:
session_id = str(evt.get("session_id") or "unknown")
exit_code = evt.get("exit_code")
reason = str(evt.get("completion_reason") or "exited")
# Completion output normally passes the terminal redactor at the producer seam, but that is
# configurable and this is user-facing, so keep the unconditional gateway floor. Redact
# Unconditional gateway redaction floor (the producer-seam redactor is configurable). Redact
# BEFORE slicing: truncating first can leave a credential fragment the patterns miss.
output = _redact_gateway_user_facing_secrets(str(evt.get("output") or "")).strip()
if len(output) > 800:
@@ -1242,16 +1138,14 @@ class GatewayNotificationsMixin:
try:
await asyncio.sleep(self._completion_notification_batch_window)
entries = self._completion_notification_batches.pop(key, [])
# Detach before adapter delivery: a completion arriving while this batch is in flight
# must be able to schedule the next flush.
# Detach before delivery so a completion arriving mid-flight can schedule the next flush.
if self._completion_notification_batch_tasks.get(key) is current_task:
self._completion_notification_batch_tasks.pop(key, None)
if not entries:
return
synth_text = entries[0][0] if len(entries) == 1 else self._format_coalesced_process_completions(entries)
# A duplicate primary can legitimately return None from the lifecycle dedupe seam; try the
# next batch identity so a fresh sibling is never discarded with that duplicate.
# A duplicate primary returns None from the dedupe seam; try the next identity so a fresh
# sibling is never discarded with it.
delivered = None
for _text, candidate_evt, _future in entries:
delivered = await self._deliver_completion_notification(synth_text, candidate_evt)
@@ -1260,8 +1154,7 @@ class GatewayNotificationsMixin:
if delivered is True and len(entries) > 1:
self._record_coalesced_completion_siblings([evt for _text, evt, _future in entries])
except asyncio.CancelledError:
# Shutdown may cancel us mid fan-in or while adapter delivery is blocked: recover entries not
# yet detached and resolve every waiter as retryable before adapters are torn down.
# Shutdown cancellation: recover undetached entries and resolve every waiter as retryable.
delivered = False
if not entries:
entries = self._completion_notification_batches.pop(key, [])
@@ -1270,8 +1163,7 @@ class GatewayNotificationsMixin:
logger.exception("Coalesced process completion delivery failed")
delivered = False
finally:
# Never strand watcher futures when formatting, delivery, or cancellation interrupts a batch:
# False follows the existing watcher retry path; None remains the ordinary dedupe result.
# Never strand watcher futures: False = watcher retry path; None = ordinary dedupe result.
self._settle_batch_waiters(entries, delivered)
# Do not remove a newer flush task that reused the same route key.
if self._completion_notification_batch_tasks.get(key) is current_task:
@@ -1295,7 +1187,6 @@ class GatewayNotificationsMixin:
task.cancel()
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
# Defensive cleanup for an orphaned queue with no live flush task.
batches = getattr(self, "_completion_notification_batches", {})
for entries in batches.values():
@@ -1306,22 +1197,17 @@ class GatewayNotificationsMixin:
async def _enqueue_process_completion_notification(self, synth_text: str, evt: dict) -> Optional[bool]:
"""Fan in concurrent process completions that share one conversation."""
# Some unit tests construct GatewayRunner with object.__new__. Keep the
# batching seam lazy so those focused lifecycle tests remain valid.
# Lazy defaults: lifecycle tests build GatewayRunner via object.__new__.
for attr, default in (
("_completion_notification_batches", dict),
("_completion_notification_batch_tasks", dict),
("_completion_notification_batches", dict), ("_completion_notification_batch_tasks", dict),
("_completion_notification_batch_flush_tasks", set),
("_completion_notification_batch_window", lambda: 0.1),
("_completion_notification_batches_stopping", lambda: False),
("_background_tasks", set),
("_completion_notification_batches_stopping", lambda: False), ("_background_tasks", set),
):
if not hasattr(self, attr):
setattr(self, attr, default())
if self._completion_notification_batches_stopping:
return False
key = self._event_route_key(evt, self._COMPLETION_BATCH_KEY_FIELDS)
future = asyncio.get_running_loop().create_future()
self._completion_notification_batches.setdefault(key, []).append((synth_text, evt, future))
@@ -1360,14 +1246,10 @@ class GatewayNotificationsMixin:
logger.debug(fail_msg, exc_info=True)
async def _deliver_async_delegation_group(self, group: list[dict]) -> Optional[bool]:
"""Deliver a same-session batch of async completions as ONE turn.
Single-event groups ride the per-event path. Multi-event groups deliver the primary via
``_deliver_completion_notification`` with consolidated text of every sibling THIS runner
claimed; sibling claims are acked only after adapter acceptance, and siblings claimed by
another consumer are excluded (no double delivery). Returns True after acceptance, False
to requeue the group, None when nothing is deliverable here (retry siblings requeued).
"""
"""Deliver a same-session batch of async completions as ONE turn: the primary carries the
consolidated text of every sibling THIS runner claimed (siblings owned elsewhere are excluded;
their claims are acked only after adapter acceptance). True after acceptance, False to requeue
the group, None when nothing is deliverable here (retry siblings requeued)."""
from gateway.run import _format_gateway_process_notification
from tools.process_registry import process_registry as _pr
deliverable: list[tuple[dict, str]] = []
@@ -1379,13 +1261,11 @@ class GatewayNotificationsMixin:
if identity is not None and self._completion_identity_seen(identity):
continue
deliverable.append((evt, synth_text))
if not deliverable:
return None
if len(deliverable) == 1:
evt, synth_text = deliverable[0]
return await self._deliver_completion_notification(synth_text, evt)
from tools.async_delegation import claim_event_delivery, complete_event_delivery, release_event_delivery
primary_evt, primary_text = deliverable[0]
blocks = [primary_text]
@@ -1393,15 +1273,12 @@ class GatewayNotificationsMixin:
for evt, synth_text in deliverable[1:]:
claim_id = claim_event_delivery(evt, f"gateway-batch:{id(self)}")
if claim_id is None:
# Another consumer owns this row's delivery; keep its result out of our
# consolidated text so it is never double-injected.
# Another consumer owns this row: keep it out of our text so it is never double-injected.
continue
siblings.append((evt, claim_id))
blocks.append(synth_text)
if not siblings:
return await self._deliver_completion_notification(primary_text, primary_evt)
header = (
f"[IMPORTANT: {len(blocks)} background subagent delegations "
"completed for this session. Treat these results as one "
@@ -1419,12 +1296,10 @@ class GatewayNotificationsMixin:
)
self._record_coalesced_completion_siblings([evt for evt, _claim_id in siblings])
else:
# Not delivered — release every sibling claim so a retry or another consumer can claim
# it, honestly leaving the durable rows pending.
# Not delivered: release every sibling claim so a retry or another consumer can take it.
self._settle_sibling_claims(siblings, release_event_delivery, "Could not release coalesced durable claim")
if delivered is None:
# The primary was dropped/owned elsewhere but the siblings still need delivery —
# requeue just them for the next tick.
# Primary dropped/owned elsewhere: requeue just the siblings for the next tick.
for evt, _claim_id in siblings:
_pr.completion_queue.put(evt)
return delivered
@@ -1439,8 +1314,7 @@ class GatewayNotificationsMixin:
from tools.process_registry import process_registry as _pr
while self._running:
with _log_suppressed(logging.DEBUG, "Async delegation watcher error: %s"):
# Peek for async-delegation events only; watch/completion events belong to other drains,
# so requeue anything that isn't ours.
# Take async-delegation events only; requeue watch/completion events for their own drains.
requeue = []
async_events = []
while not _pr.completion_queue.empty():
@@ -1451,9 +1325,8 @@ class GatewayNotificationsMixin:
(async_events if evt.get("type") == "async_delegation" else requeue).append(evt)
for evt in requeue:
_pr.completion_queue.put(evt)
# A same-tick drain often carries several completions for the SAME session (a fan-out
# finishing together); delivering each individually floods it with N synthetic turns.
# Group by full gateway route + parent session: one consolidated turn per group.
# A fan-out finishing together yields N completions for one session; group by full route +
# parent session so each group becomes ONE consolidated turn.
groups: dict[tuple[str, ...], list[dict]] = {}
for evt in async_events:
self._enrich_async_delegation_routing(evt)
@@ -1478,9 +1351,8 @@ class GatewayNotificationsMixin:
if new_output:
from agent.redact import redact_terminal_output
new_output = redact_terminal_output(new_output, getattr(session, "command", "") or "")
# redact_terminal_output() is unforced, so it returns raw text when security.redact_secrets
# is off. This goes straight to the platform adapter, so it needs the same unconditional
# floor as agent-notify.
# redact_terminal_output() is unforced (raw when security.redact_secrets is off); this goes
# straight to the adapter, so apply the same unconditional floor as agent-notify.
new_output = _redact_gateway_user_facing_secrets(new_output)
return new_output
@@ -1503,8 +1375,7 @@ class GatewayNotificationsMixin:
_command = getattr(session, "command", "") or ""
_raw = strip_ansi(session.output_buffer) if session.output_buffer else ""
_raw = redact_terminal_output(_raw, _command)
# Truncate on line boundaries (never start mid-line): keep the last ~2000 chars snapped to
# the preceding newline, prepending a marker when output was cut.
# Keep the last ~2000 chars snapped to a line boundary, with a marker when cut.
_LIMIT = 2000
if len(_raw) > _LIMIT:
_tail = _raw[-_LIMIT:]
@@ -1524,8 +1395,7 @@ class GatewayNotificationsMixin:
"completion_reason": getattr(session, "completion_reason", "exited"),
"termination_source": getattr(session, "termination_source", ""),
"output": _redact_gateway_user_facing_secrets(_out),
# Spawning conversation's session-db id (stamped in terminal_tool); lets delivery
# pre-flight drop this completion if the user closed that session (/new) first.
# Spawning session-db id: lets pre-flight drop this completion if the user /new'd first.
"parent_session_id": (
watcher.get("parent_session_id") or getattr(session, "parent_session_id", "") or ""
),
@@ -1547,13 +1417,10 @@ class GatewayNotificationsMixin:
)
async def _run_process_watcher(self, watcher: dict) -> None:
"""Periodically check a background process and push updates to the user.
Runs as an asyncio task. Stays silent when nothing changed. Auto-removes when the process
exits or is killed. Notification mode (``display.background_process_notifications``):
concise (default, one-line; failures append output tail) / all (running updates + final
raw output) / result (final raw only) / error (final raw only if exit != 0) / off.
"""
"""Poll a background process and push updates until it exits. Mode
(``display.background_process_notifications``): concise (default one-liner; failures append
the output tail) / all (running updates + final raw) / result (final raw) / error (final raw
if exit != 0) / off."""
from tools.process_registry import format_process_notification, process_registry
session_id = watcher["session_id"]
interval = watcher["check_interval"]
@@ -1562,10 +1429,8 @@ class GatewayNotificationsMixin:
thread_id = watcher.get("thread_id", "")
agent_notify = watcher.get("notify_on_complete", False)
notify_mode = self._load_background_notifications_mode()
logger.debug("Process watcher started: %s (every %ss, notify=%s, agent_notify=%s)",
session_id, interval, notify_mode, agent_notify)
silent = notify_mode == "off" and not agent_notify
last_output_len = 0
while True:
@@ -1581,11 +1446,9 @@ class GatewayNotificationsMixin:
current_output_len = len(session.output_buffer)
has_new_output = current_output_len > last_output_len
last_output_len = current_output_len
if session.exited:
# Agent-triggered completion: inject a synthetic message unless the agent already consumed
# the result via wait/log. poll() is read-only and deliberately does NOT mark consumed —
# a status check must not suppress this delivery turn.
# Agent-notify: inject a synthetic message unless the agent already consumed the result via
# wait/log (poll() is read-only and deliberately does NOT mark consumed).
if agent_notify and not process_registry.is_completion_consumed(session_id):
completion_evt = self._build_process_completion_event(watcher, session, session_id)
synth_text = format_process_notification(completion_evt)
@@ -1597,10 +1460,8 @@ class GatewayNotificationsMixin:
# of suppressing the result.
continue
break
# Normal text-only notification. Skip when the agent already consumed this completion via
# wait/log (output returned inline) — the raw "finished" message would be a duplicate.
# The agent_notify skip FALLS THROUGH here, hence this check. poll() is read-only.
# Text-only notification; skip when already consumed via wait/log (the agent_notify branch
# FALLS THROUGH here, hence the re-check).
if process_registry.is_completion_consumed(session_id):
logger.debug(
"Process watcher: completion for %s already consumed "
@@ -1613,7 +1474,6 @@ class GatewayNotificationsMixin:
message_text = self._format_process_final_message(session_id, session, notify_mode)
await self._send_watcher_message(platform_name, chat_id, thread_id, message_text)
break
elif has_new_output and notify_mode == "all" and not agent_notify:
# New output — deliver a status update (only in "all" mode; agent_notify watchers
# only care about completion).
@@ -1622,5 +1482,4 @@ class GatewayNotificationsMixin:
platform_name, chat_id, thread_id,
f"[Background process {session_id} is still running~ New output:\n{new_output}]",
)
logger.debug("Process watcher ended%s: %s", " (silent)" if silent else "", session_id)
gateway/run_shutdown.py
+174 -243
View File
@@ -138,11 +138,15 @@ class GatewayShutdownMixin:
+ self._active_deferred_agent_worker_count()
)
@staticmethod
def _running_cron_job_count() -> int:
from cron.scheduler import get_running_job_ids
return len(get_running_job_ids())
def _active_cron_job_count(self) -> int:
"""Cron jobs currently executing — they run outside ``_running_agents``; 0 if cron can't import."""
try:
from cron.scheduler import get_running_job_ids
return len(get_running_job_ids())
return self._running_cron_job_count()
except Exception:
return 0
@@ -151,8 +155,7 @@ class GatewayShutdownMixin:
Only the primary API server owns the HTTP listener, so only it is a source of this work.
"""
adapter = getattr(self, "adapters", {}).get(Platform.API_SERVER)
helper = getattr(adapter, name, None)
helper = getattr(getattr(self, "adapters", {}).get(Platform.API_SERVER), name, None)
return max(0, int(helper(*args))) if callable(helper) else 0
def _active_api_run_count(self) -> int:
@@ -225,18 +228,18 @@ class GatewayShutdownMixin:
for t in self._background_tasks
):
return True
try:
def _delegations_active() -> bool:
from tools.async_delegation import active_count
if active_count() > 0:
return True
except Exception: # noqa: BLE001 - never let the idle check raise
logger.debug("scale-to-zero async-delegation check failed", exc_info=True)
try:
return active_count() > 0
def _processes_active() -> bool:
from tools.process_registry import process_registry
if process_registry.has_any_active() or process_registry.pending_watchers:
return True
except Exception: # noqa: BLE001 - never let the idle check raise
logger.debug("scale-to-zero bg-work check failed", exc_info=True)
return bool(process_registry.has_any_active() or process_registry.pending_watchers)
for label, probe in (("async-delegation", _delegations_active), ("bg-work", _processes_active)):
with _log_suppressed(logging.DEBUG, f"scale-to-zero {label} check failed", exc_info=True):
if probe():
return True
return False
@staticmethod
@@ -319,9 +322,8 @@ class GatewayShutdownMixin:
return # not opted in — normal, stay quiet
active = [getattr(p, "value", p) for p in self._scale_to_zero_active_messaging_platforms()]
logger.info(
"scale-to-zero: NOT armed despite opt-in — "
"relay_only_or_absent=%s (enabled platforms=%s), wake_url=%s. "
"Need relay-only messaging + a registered wake URL.",
"scale-to-zero: NOT armed despite opt-in — relay_only_or_absent=%s (enabled platforms=%s), "
"wake_url=%s. Need relay-only messaging + a registered wake URL.",
messaging_is_relay_only_or_absent(active), active or "none",
"set" if self._relay_wake_url_or_none() else "MISSING",
)
@@ -330,9 +332,8 @@ class GatewayShutdownMixin:
def _scale_to_zero_is_idle(self) -> bool:
from gateway.scale_to_zero import is_idle
# FULL work aggregate (cron + API runs live outside _running_agents) with fail-AWAKE reads:
# the drain counters swallow errors to 0, which for a suspend predicate would look idle, so
# an unreadable source counts as work here.
# FULL work aggregate with fail-AWAKE reads: the drain counters swallow errors to 0, which a
# suspend predicate would read as idle, so an unreadable source counts as work here.
def _read_or_awake(label: str, fn: Callable[[], Any], busy_sentinel: Any) -> Any:
try:
@@ -341,15 +342,10 @@ class GatewayShutdownMixin:
logger.debug("scale-to-zero: %s unreadable — staying awake", label, exc_info=True)
return busy_sentinel
def _cron_count() -> int:
from cron.scheduler import get_running_job_ids
return len(get_running_job_ids())
cron_count = _read_or_awake("cron work count", _cron_count, 1)
cron_count = _read_or_awake("cron work count", self._running_cron_job_count, 1)
api_count = _read_or_awake("api work count", lambda: self._api_server_hook("active_agent_work_count"), 1)
# An attached dashboard/desktop/TUI client (heartbeat file mtime from the dashboard process) is
# inbound activity too — folded into the inbound clock, not a conjunct, so disconnect gets the
# same idle_timeout grace as a message and a lingering marker cannot pin the box.
# An attached dashboard/desktop/TUI client (heartbeat mtime) is inbound activity — folded into
# the inbound clock, not a conjunct, so a lingering marker cannot pin the box.
last_inbound = self._last_inbound_at
from gateway.scale_to_zero import dashboard_client_last_seen
seen = _read_or_awake("dashboard heartbeat", dashboard_client_last_seen, time.time())
@@ -385,13 +381,10 @@ class GatewayShutdownMixin:
return self.adapters.get(Platform.RELAY)
async def _scale_to_zero_watcher(self, interval: float = 30.0) -> None:
"""Watch for idle, drive the relay dormant, then self-suspend the machine.
On sustained idle: status `draining` (NOT _running=False), relay go_dormant() (socket close,
NOT disconnect()), no mark_resume_pending (suspend preserves RAM), THEN suspend via the flaps
socket. The gateway owns the suspend because Fly autostop sees only INBOUND connections and
would freeze mid-job or before the relay flip. Off-Fly the watcher does not quiesce at all.
"""
"""Watch for idle, drive the relay dormant, then self-suspend. On sustained idle: status
`draining` (NOT _running=False), relay go_dormant() (socket close, NOT disconnect()), no
mark_resume_pending (suspend preserves RAM), THEN suspend via the flaps socket — Fly autostop
sees only INBOUND connections and would freeze mid-job. Off-Fly: no quiesce at all."""
await asyncio.sleep(min(interval, 30.0)) # let startup settle
while self._running:
try:
@@ -410,9 +403,8 @@ class GatewayShutdownMixin:
if not self._scale_to_zero_no_suspend_logged:
self._scale_to_zero_no_suspend_logged = True
logger.info(
"scale-to-zero: idle, but this platform suspends on "
"its own timer (no in-machine suspend API); staying "
"connected rather than quiescing"
"scale-to-zero: idle, but this platform suspends on its own timer (no "
"in-machine suspend API); staying connected rather than quiescing"
)
continue
logger.info(
@@ -432,9 +424,8 @@ class GatewayShutdownMixin:
# After a wake the drained inbound updates _last_inbound_at; give it a window so we
# don't immediately re-go-dormant on the same idle reading before traffic lands.
self._scale_to_zero_cooldown_until = time.time() + max(interval, 60.0)
# Self-suspend ONLY after a clean quiesce: the relay flip (buffered delivery + wake
# poke armed) must be set before the freeze, or inbound black-holes while we sleep.
# Re-check idle one last time — inbound may have landed during the quiesce await.
# Self-suspend ONLY after a clean quiesce (else inbound black-holes while we sleep), and
# re-check idle — inbound may have landed during the quiesce await.
if not dormant_ok:
continue
if not self._scale_to_zero_is_idle():
@@ -463,9 +454,8 @@ class GatewayShutdownMixin:
except Exception: # noqa: BLE001 - suspend is best-effort, never crash
logger.debug("scale-to-zero: self-suspend failed", exc_info=True)
# External drain control: the dashboard's begin/cancel-drain endpoint writes/removes the
# ``.drain_request.json`` marker (gateway/drain_control.py); the watcher flips the gateway
# between accepting and refusing NEW turns WITHOUT exiting (reversible on marker removal).
# External drain control: the dashboard writes/removes ``.drain_request.json`` (gateway/drain_control.py);
# the watcher flips between accepting and refusing NEW turns WITHOUT exiting (reversible).
def _enter_external_drain(self) -> None:
"""Begin external drain: refuse NEW turns (in-flight ones are NOT interrupted). Idempotent."""
if self._external_drain_active:
@@ -473,11 +463,9 @@ class GatewayShutdownMixin:
self._external_drain_active = True
logger.info(
"External drain ENGAGED (.drain_request.json present) — refusing "
"new turns; %d in-flight turn(s) will finish. Process stays up.",
self._active_work_count(),
"new turns; %d in-flight turn(s) will finish. Process stays up.", self._active_work_count(),
)
# Flip persisted lifecycle state so /api/status.gateway_busy / gateway_drainable track the
# drain; active_agents is preserved (read-merge keeps the live count), only state changes.
# Persist "draining" so /api/status tracks it; active_agents is read-merged, only state changes.
self._update_runtime_status("draining")
def _exit_external_drain(self) -> None:
@@ -502,9 +490,8 @@ class GatewayShutdownMixin:
from gateway.drain_control import drain_requested
while self._running:
try:
# drain_requested() does a synchronous read_text() on the marker file: at 1s cadence
# that is a blocking disk read on the event loop ~86k times/day, and under host I/O
# pressure one read can stall 30s+ and take every platform heartbeat down. Off-thread it.
# Off-thread: a synchronous marker read at 1s cadence can stall 30s+ under host I/O
# pressure and take every platform heartbeat down.
if await asyncio.to_thread(drain_requested):
self._enter_external_drain()
# API and cron work live outside messaging's _running_agents map; refresh the
@@ -523,17 +510,16 @@ class GatewayShutdownMixin:
error_code: Optional[str] = None, error_message: Optional[str] = None,
needs_attention: Optional[bool] = None, retrying_since: Any = _UNSET,
) -> None:
with suppress(Exception):
from gateway.status import write_runtime_status
extra: Dict[str, Any] = {}
if needs_attention is not None:
extra["needs_attention"] = needs_attention
if retrying_since is not _UNSET:
extra["retrying_since"] = retrying_since
write_runtime_status(
platform=platform, platform_state=platform_state, error_code=error_code,
error_message=error_message, **extra,
)
from gateway.run import _write_runtime_status_quiet
extra: Dict[str, Any] = {}
if needs_attention is not None:
extra["needs_attention"] = needs_attention
if retrying_since is not _UNSET:
extra["retrying_since"] = retrying_since
_write_runtime_status_quiet(
platform=platform, platform_state=platform_state, error_code=error_code,
error_message=error_message, **extra,
)
# Per-platform circuit breaker (pause/resume): reconnect watcher + /platform pause|resume.
def _pause_failed_platform(self, platform, *, reason: str = "") -> None:
@@ -543,15 +529,14 @@ class GatewayShutdownMixin:
return
info["paused"] = True
info["pause_reason"] = reason or "auto-paused after repeated failures"
# Push next_retry far enough out that a stale code path missing "paused" still never fires.
# next_retry=inf: a stale code path missing "paused" still never fires.
info["next_retry"] = float("inf")
self._update_platform_runtime_status(
platform.value, platform_state="paused", error_code=None, error_message=info["pause_reason"],
)
logger.warning(
"%s paused after %d consecutive failures (%s) — "
"fix the underlying issue then run `/platform resume %s` "
"to retry, or `hermes gateway restart` to restart the gateway.",
"%s paused after %d consecutive failures (%s) — fix the underlying issue then run `/platform "
"resume %s` to retry, or `hermes gateway restart` to restart the gateway.",
platform.value, info.get("attempts", 0), info["pause_reason"], platform.value,
)
@@ -592,14 +577,12 @@ class GatewayShutdownMixin:
self._update_runtime_status("draining")
last_counts, last_status_at = counts, now
# Cron/API-server/deferred work lives outside ``self._running_agents`` — fold it into this
# wait, or a cron job's tool work is killed without warning once it's the only thing running.
# Cron/API/deferred work lives outside ``_running_agents``; fold it in or it is killed unwarned.
_cron0, _api0, _deferred0 = last_counts[1:]
_maybe_update_status(force=True)
if not self._running_agents and not (_cron0 or _api0 or _deferred0):
return snapshot, False
# Cron has its own deadline: ``restart_drain_timeout`` defaults to 0 (an interrupted chat turn
# is announced and resumable) but a cron run killed mid-flight is a permanent failure.
# Cron has its own deadline: a chat turn is announced+resumable; a killed cron run is a permanent failure.
started = loop.time()
deadline = started + timeout
cron_deadline = started + (timeout if cron_timeout is None else cron_timeout)
@@ -609,8 +592,7 @@ class GatewayShutdownMixin:
agents, cron, api, deferred = self._drain_work_counts()
return bool(((agents or api or deferred) and now < deadline) or (cron and now < cron_deadline))
# Both budgets at 0 leave this loop unentered ("interrupt immediately") as an expired deadline,
# not a special case, so timed_out below is always computed from real state.
# Both budgets at 0 = an expired deadline (loop unentered), so timed_out still comes from real state.
while _still_draining():
_maybe_update_status()
await asyncio.sleep(0.1)
@@ -628,12 +610,12 @@ class GatewayShutdownMixin:
logger.debug("Interrupted running agent for session %s during shutdown", session_key)
# API-server / desk turns are adapter-owned and never enter _running_agents, so the loop above
# cannot see them even though _drain_active_agents() waited for them.
interrupted_api = self._interrupt_api_server_runs(reason)
if interrupted_api:
logger.debug("Interrupted %d api_server run(s) during shutdown", interrupted_api)
interrupted_deferred = self._interrupt_deferred_agent_workers(reason)
if interrupted_deferred:
logger.debug("Interrupted %d deferred agent worker(s) during shutdown", interrupted_deferred)
for count, what in (
(self._interrupt_api_server_runs(reason), "api_server run(s)"),
(self._interrupt_deferred_agent_workers(reason), "deferred agent worker(s)"),
):
if count:
logger.debug("Interrupted %d %s during shutdown", count, what)
def _shutdown_interrupt_reason(self) -> str:
from gateway.run import _INTERRUPT_REASON_GATEWAY_RESTART, _INTERRUPT_REASON_GATEWAY_SHUTDOWN
@@ -687,9 +669,8 @@ class GatewayShutdownMixin:
job = get_job(job_id)
if not job:
continue
# deliver=local jobs, and deliver=origin jobs with no resolvable origin, resolve to zero
# targets and must stay silent rather than fall back to a home channel. Interrupted
# notices are failure-category engine status, so they honor failure_deliver.
# deliver=local / unresolvable-origin jobs resolve to zero targets and stay silent (no home-
# channel fallback). Interrupted notices are failure-category status: honor failure_deliver.
targets = _resolve_delivery_targets(job, for_failure=True)
except Exception as e:
logger.debug("Cron interrupt targets unresolved for %s: %s", job_id, e)
@@ -712,17 +693,13 @@ class GatewayShutdownMixin:
dedup_key = (job_id, *_notice_target_key(platform.value, chat_id, thread_id))
if dedup_key in notified:
continue
try:
with _log_suppressed(logging.DEBUG, "Cron interrupt notice to %s:%s raised: %s", platform.value, chat_id):
metadata = self._thread_metadata_for_target(platform, chat_id, thread_id, adapter=adapter)
result = await adapter.send(chat_id, msg, metadata=metadata)
if _send_failed(result):
logger.debug(
"Cron interrupt notice to %s:%s failed: %s", platform.value, chat_id, _send_error(result),
)
continue
notified.add(dedup_key)
except Exception as e:
logger.debug("Cron interrupt notice to %s:%s raised: %s", platform.value, chat_id, e)
if await self._send_notice_logged(
adapter, chat_id, msg, platform.value, "Cron interrupt notice to %s:%s failed: %s",
"Cron interrupt notice to %s:%s raised: %s", metadata=metadata,
):
notified.add(dedup_key)
if notified:
logger.info("Shutdown: delivered %d interrupted-cron-job notice(s)", len(notified))
return len(notified)
@@ -752,19 +729,27 @@ class GatewayShutdownMixin:
) -> bool:
"""Send one shutdown notice; True when delivered. Failures are debug-logged, never raised."""
where = "home channel " if kind == "home channel" else ""
try:
result = await adapter.send(chat_id, msg, **send_kwargs)
if _send_failed(result):
logger.debug(
"Failed to send shutdown notification to %s%s:%s: %s", where, platform_str, chat_id,
_send_error(result),
)
return False
logger.info("Sent shutdown notification to %s %s:%s", kind, platform_str, chat_id)
return True
except Exception as e:
logger.debug("Failed to send shutdown notification to %s%s:%s: %s", where, platform_str, chat_id, e)
fail_fmt = f"Failed to send shutdown notification to {where}%s:%s: %s"
if not await self._send_notice_logged(adapter, chat_id, msg, platform_str, fail_fmt, **send_kwargs):
return False
logger.info("Sent shutdown notification to %s %s:%s", kind, platform_str, chat_id)
return True
@staticmethod
async def _send_notice_logged(
adapter, chat_id: str, msg: str, platform_str: str, fail_fmt: str, raise_fmt: Optional[str] = None, **kw
) -> bool:
"""``adapter.send`` whose failure is debug-logged as ``fmt % (platform, chat, error)`` — ``fail_fmt``
for success=False, ``raise_fmt`` (default ``fail_fmt``) for a raise; True only on a delivered send."""
try:
result = await adapter.send(chat_id, msg, **kw)
except Exception as e:
logger.debug(raise_fmt or fail_fmt, platform_str, chat_id, e)
return False
if _send_failed(result):
logger.debug(fail_fmt, platform_str, chat_id, _send_error(result))
return False
return True
async def _notify_active_sessions_of_shutdown(self) -> None:
"""Send shutdown/restart notifications to active chats and home channels.
@@ -772,14 +757,12 @@ class GatewayShutdownMixin:
Called at the start of stop() while adapters are connected; send failures never block shutdown.
"""
restart_source = self._restart_command_source if self._restart_requested else None
action = "restarting" if self._restart_requested else "shutting down"
hint = (
"Your current task will be interrupted. "
"Send any message after restart and I'll try to resume where you left off."
if self._restart_requested
else "Your current task will be interrupted."
)
msg = f"⚠️ Gateway {action} — {hint}"
msg = "⚠️ Gateway shutting down — Your current task will be interrupted."
if self._restart_requested:
msg = (
"⚠️ Gateway restarting — Your current task will be interrupted. "
"Send any message after restart and I'll try to resume where you left off."
)
restart_key = None
if restart_source is not None:
with suppress(Exception):
@@ -817,19 +800,15 @@ class GatewayShutdownMixin:
if self._restart_requested and restart_source is not None:
logger.debug("Skipping home-channel shutdown notifications for in-chat restart")
return
# Suppress ONLY the home-channel broadcast when the drain asked to be quiet (routine fleet
# auto-update); per-session pings above stay (empty on a drained shutdown, useful on a
# force-interrupt). Current-epoch marker only, so an orphaned marker can't silence a fresh gateway.
try:
# A quiet drain (routine fleet auto-update) suppresses ONLY the home-channel broadcast; per-session
# pings above stay. Current-epoch marker only; a failing check fails toward the louder behaviour.
with _log_suppressed(logging.DEBUG, "drain_notification_suppressed check failed: %s"):
from gateway.drain_control import drain_notification_suppressed
if drain_notification_suppressed():
logger.info(
"Home-channel shutdown broadcast suppressed by drain marker (suppress_notification=true)"
)
return
except Exception as e:
# Never let the suppression check block the broadcast — fail toward the louder behaviour.
logger.debug("drain_notification_suppressed check failed: %s", e)
# Snapshot adapters: adapter.send() can hit a fatal path (_handle_fatal) that pops the adapter
# from self.adapters -> ``RuntimeError: dictionary changed size during iteration``.
for platform, adapter in list(self.adapters.items()):
@@ -849,9 +828,9 @@ class GatewayShutdownMixin:
)
continue
# Home channels omit ``metadata=`` when empty (adapter doubles may not accept the kwarg).
send_kwargs = {"metadata": metadata} if metadata else {}
if await self._send_shutdown_notice(
adapter, str(home.chat_id), msg, "home channel", platform.value, **send_kwargs
adapter, str(home.chat_id), msg, "home channel", platform.value,
**({"metadata": metadata} if metadata else {}),
):
notified.add(dedup_key)
@@ -976,9 +955,8 @@ class GatewayShutdownMixin:
return
with suppress(Exception):
if hasattr(agent, "shutdown_memory_provider"):
# Drain queued memory writes BEFORE teardown: shutdown_all() gives the memory worker
# only ~5s, so a /reset or rotation could drop handed-off writes (stale memory next
# session). Bounded, mirrors CLI exit; a failure never blocks teardown.
# Drain queued memory writes BEFORE teardown (shutdown_all() gives the worker only ~5s, so a
# /reset or rotation could drop them). Bounded; a failure never blocks teardown.
_mm = getattr(agent, "_memory_manager", None)
if _mm is not None and hasattr(_mm, "flush_pending"):
with suppress(Exception):
@@ -1007,7 +985,7 @@ class GatewayShutdownMixin:
@staticmethod
def _read_json_counts(path: Path) -> Optional[dict]:
"""Parsed counter dict, or None when the file is missing/unreadable."""
"""Parsed counter dict, or None when the file is missing/unreadable (no exists() pre-check needed)."""
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
@@ -1017,10 +995,9 @@ class GatewayShutdownMixin:
"""Increment persisted restart-failure counters for active sessions; drop the rest (loop broken)."""
from gateway.run import atomic_json_write
path = self._stuck_loop_counts_path()
counts = (self._read_json_counts(path) if path.exists() else None) or {}
new_counts = {key: counts.get(key, 0) + 1 for key in active_session_keys}
counts = self._read_json_counts(path) or {}
with suppress(Exception):
atomic_json_write(path, new_counts, indent=None)
atomic_json_write(path, {key: counts.get(key, 0) + 1 for key in active_session_keys}, indent=None)
def _suspend_stuck_loop_sessions(self) -> int:
"""Suspend sessions active across too many restarts (startup, AFTER suspend_recently_active())."""
@@ -1032,7 +1009,7 @@ class GatewayShutdownMixin:
return 0
suspended = 0
for session_key in [k for k, v in counts.items() if v >= self._STUCK_LOOP_THRESHOLD]:
try:
with suppress(Exception):
entry = self.session_store._entries.get(session_key)
if entry and not entry.suspended:
entry.suspended = True
@@ -1041,8 +1018,6 @@ class GatewayShutdownMixin:
"Auto-suspended stuck session %s (active across %d consecutive restarts — likely a stuck loop)",
session_key, counts[session_key],
)
except Exception:
pass
if suspended:
with suppress(Exception):
self.session_store._save()
@@ -1085,9 +1060,8 @@ class GatewayShutdownMixin:
)
watcher_env = GatewayShutdownMixin._restart_watcher_env()
project_root = Path(__file__).resolve().parent.parent
# Console python under CREATE_NO_WINDOW owns one hidden console inherited by the restart
# child, so nothing flashes. Do NOT swap in pythonw.exe — a console-less watcher forces
# every console-subsystem descendant to allocate a visible conhost.
# Console python under CREATE_NO_WINDOW: nothing flashes. NOT pythonw.exe — a console-less
# watcher makes every console-subsystem descendant allocate a visible conhost.
watcher_python = sys.executable
venv_dir = Path(watcher_env.get("VIRTUAL_ENV") or project_root / "venv")
site_packages = venv_dir / "Lib" / "site-packages"
@@ -1102,10 +1076,8 @@ class GatewayShutdownMixin:
str(current_pid), str(restart_after_s), *hermes_cmd, "gateway", "restart",
]
popen_kwargs = dict(stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=watcher_env)
# The watcher must break away from the parent CLI's job object (Desktop wrappers, Windows
# Terminal, schtasks) or it is reaped when the CLI exits. A job without
# JOB_OBJECT_LIMIT_BREAKAWAY_OK rejects CREATE_BREAKAWAY_FROM_JOB (OSError); retry once
# without the bit, preserving argv and the scrubbed env.
# Break away from the parent CLI's job object or be reaped when the CLI exits; a job without
# BREAKAWAY_OK rejects CREATE_BREAKAWAY_FROM_JOB (OSError) — retry once without the bit.
try:
subprocess.Popen(watcher_argv, **popen_kwargs, **windows_detach_popen_kwargs())
except OSError:
@@ -1252,22 +1224,19 @@ class GatewayShutdownMixin:
self._restart_detached = detached
self._restart_via_service = via_service
self._restart_task_started = True
# Refuse new turns while in-flight work finishes. Keep ``_running`` True so adapters stay
# connected and the active turn can still deliver its final response.
# Refuse new turns; keep ``_running`` True so the active turn can still deliver its final response.
self._draining = True
async def _run_restart() -> None:
await self._await_active_work_before_restart()
# Launch the detached helper only AFTER the after-turn wait: its drain_timeout+5 deadline
# covers stop() teardown; earlier it would fire the restart mid-turn.
# Detached helper only AFTER the after-turn wait, or its drain_timeout+5 deadline fires mid-turn.
if detached:
with _log_suppressed(logging.ERROR, "Failed to launch detached gateway restart helper: %s"):
await self._launch_detached_restart_command()
await asyncio.sleep(0.05)
await self.stop(restart=True, detached_restart=detached, service_restart=via_service)
# NOT in _background_tasks: _stop_impl cancels every entry there, which would propagate
# CancelledError into _stop_impl and skip _shutdown_event.set() / _exit_code = 75.
# NOT in _background_tasks: _stop_impl cancels those, which would skip _shutdown_event.set() / exit 75.
self._restart_task = asyncio.create_task(_run_restart())
return True
@@ -1315,16 +1284,18 @@ class GatewayShutdownMixin:
def _step(label: str, fn: Callable[[], Any]) -> Any:
return GatewayShutdownMixin._quiet_step(f"{label} ({phase}) error", fn)
def _count_step(fmt: str, fn: Callable[[], int]) -> None:
n = fn()
if n:
logger.info(fmt, phase, n)
def _kill_processes() -> None:
from tools.process_registry import process_registry
_killed = process_registry.kill_all()
if _killed:
logger.info("Shutdown (%s): killed %d tool subprocess(es)", phase, _killed)
_count_step("Shutdown (%s): killed %d tool subprocess(es)", process_registry.kill_all)
def _mark_cron_interrupted() -> list:
# kill_all() is a global sweep, so any cron job dispatched right now lost its tool
# subprocess; its agent thread may still emit a plausible response from truncated
# output. Mark the run interrupted so it can never be reported as success.
# kill_all() is global: a cron job mid-dispatch lost its tool subprocess and its agent thread may
# still emit a plausible response from truncated output — mark it interrupted, never success.
from cron.scheduler import mark_running_jobs_interrupted
_interrupted = mark_running_jobs_interrupted(
f"Gateway shutdown ({phase}) killed the job's tool subprocess before the run finished."
@@ -1338,9 +1309,10 @@ class GatewayShutdownMixin:
def _interrupt_delegations() -> None:
from tools.async_delegation import interrupt_all as _interrupt_async
_async_n = _interrupt_async(reason=f"gateway shutdown ({phase})")
if _async_n:
logger.info("Shutdown (%s): interrupted %d background delegation(s)", phase, _async_n)
_count_step(
"Shutdown (%s): interrupted %d background delegation(s)",
lambda: _interrupt_async(reason=f"gateway shutdown ({phase})"),
)
_step("process_registry.kill_all", _kill_processes)
_marked_cron_jobs = _step("mark_running_jobs_interrupted", _mark_cron_interrupted) or []
@@ -1388,16 +1360,14 @@ class GatewayShutdownMixin:
async def _stop_drain_active_work(self, timeout: float, ctx: "GatewayShutdownMixin._StopContext") -> None:
"""Pre-mark resume_pending, drain agents/cron/API work into ``ctx``."""
from gateway.run import GatewayRunner
# Pre-mark resume_pending BEFORE the drain: if the service manager kills us mid-drain, the
# durable marker already lets the next boot recover the sessions.
# Pre-mark resume_pending BEFORE the drain so a mid-drain SIGKILL still leaves a durable marker.
_pre_drain_keys = await GatewayRunner._mark_running_sessions_resume_pending(
self, "pre-drain mark_resume_pending"
)
_cron_at_start = self._active_cron_job_count()
_api_at_start = self._active_api_run_count()
_deferred_at_start = ctx.deferred_count()
# Cron floor clamped to the watchdog leash so it never costs the post-drain cleanup window.
# getattr-guard: shutdown-path tests drive the phases from bare doubles lacking the class default.
# Cron floor clamped to the watchdog leash; getattr-guard for bare shutdown-path doubles.
_cron_drain_cfg = getattr(self, "_cron_drain_timeout", DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT)
_cron_timeout = resolve_cron_drain_budget(
timeout, _cron_drain_cfg, watchdog_delay=resolve_shutdown_watchdog_delay(timeout),
@@ -1406,18 +1376,15 @@ class GatewayShutdownMixin:
if _cron_at_start and _cron_timeout > timeout:
logger.info(
"Shutdown drain: %d in-flight cron job(s) — waiting up to "
"%.0fs for them (cron_drain_timeout=%.0fs, "
"restart_drain_timeout=%.0fs)",
"%.0fs for them (cron_drain_timeout=%.0fs, restart_drain_timeout=%.0fs)",
_cron_at_start, _cron_timeout, _cron_drain_cfg, timeout,
)
_drain_started_at = time.monotonic()
ctx.active_agents, ctx.timed_out = await self._drain_active_agents(timeout, _cron_timeout)
ctx.drain_elapsed = time.monotonic() - _drain_started_at
logger.info(
"Shutdown phase: drain done at +%.2fs (drain took %.2fs, "
"timed_out=%s, active_at_start=%d, active_now=%d, "
"cron_at_start=%d, cron_now=%d, "
"api_at_start=%d, api_now=%d, "
"Shutdown phase: drain done at +%.2fs (drain took %.2fs, timed_out=%s, active_at_start=%d, "
"active_now=%d, cron_at_start=%d, cron_now=%d, api_at_start=%d, api_now=%d, "
"deferred_at_start=%d, deferred_now=%d)", ctx.elapsed(), ctx.drain_elapsed,
ctx.timed_out, len(ctx.active_agents), self._running_agent_count(), _cron_at_start,
self._active_cron_job_count(), _api_at_start, self._active_api_run_count(),
@@ -1439,15 +1406,12 @@ class GatewayShutdownMixin:
from gateway.run import GatewayRunner
logger.warning(
"Gateway drain timed out after %.1fs with %d active agent(s), "
"%d in-flight cron job(s), %d api_server run(s), and "
"%d deferred agent worker(s); "
"%d in-flight cron job(s), %d api_server run(s), and %d deferred agent worker(s); "
"interrupting remaining work.", ctx.drain_elapsed, self._running_agent_count(),
self._active_cron_job_count(), self._active_api_run_count(), ctx.deferred_count(),
)
# Mark forcibly-interrupted sessions resume_pending BEFORE interrupting so the next message
# auto-resumes instead of becoming a fresh session via suspend_recently_active(); stuck
# sessions still escalate via ``.restart_failure_counts`` (suspended=True wins). Uses the
# CURRENT _running_agents, not the drain-start snapshot, so cleanly finished sessions get no note.
# Mark resume_pending BEFORE interrupting so the next message auto-resumes (stuck sessions
# still escalate via .restart_failure_counts). CURRENT _running_agents, not the drain snapshot.
await GatewayRunner._mark_running_sessions_resume_pending(self, "mark_resume_pending")
reason = GatewayRunner._shutdown_interrupt_reason(self)
self._interrupt_running_agents(reason)
@@ -1459,23 +1423,19 @@ class GatewayShutdownMixin:
def _work_live() -> bool:
return bool(self._running_agents or self._active_api_run_count() or ctx.deferred_count())
# Wait on API-server work too: the interrupt is cooperative; closing the settle window as soon
# as _running_agents is empty would kill an API turn's tool subprocesses before it unwinds.
# Wait on API-server work too, or an API turn's tool subprocesses are killed before it unwinds.
while _work_live() and loop.time() < interrupt_deadline:
self._update_runtime_status("draining")
await asyncio.sleep(0.1)
# Work can materialize AFTER the one-shot interrupt (a /v1/runs task registers only when
# _create_agent returns; a pending sentinel is promoted by track_agent() later). Re-signal so
# it gets a cooperative interrupt instead of a bare tool-subprocess kill.
# Work can materialize AFTER the one-shot interrupt (/v1/runs registers on _create_agent return;
# pending sentinels promote later). Re-signal for a cooperative interrupt, not a bare kill.
if _work_live():
self._interrupt_running_agents(reason)
logger.debug("Re-signaled interrupt for work still live at settle-window exit")
# Kill lingering tool subprocesses NOW, before adapter disconnect / DB close: under systemd
# (TimeoutStopSec ≈ drain_timeout + headroom) deferring risks the cgroup SIGKILL reaping them.
# Kill tool subprocesses NOW: deferring past adapter/DB teardown risks the systemd cgroup SIGKILL.
_interrupted_cron_jobs = GatewayRunner._stop_kill_tool_subprocesses("post-interrupt")
logger.info("Shutdown phase: post-interrupt tool kill done at +%.2fs", ctx.elapsed())
# Last window with the transport up: the cron worker's own "interrupted" notice arrives
# after adapter teardown and is lost.
# Last window with the transport up (the cron worker's own notice arrives after teardown).
with _log_suppressed(logging.DEBUG, "Cron interrupt notification failed: %s"):
await self._notify_interrupted_cron_jobs(_interrupted_cron_jobs)
logger.info("Shutdown phase: cron interrupt notices done at +%.2fs", ctx.elapsed())
@@ -1486,8 +1446,7 @@ class GatewayShutdownMixin:
with _log_suppressed(logging.ERROR, "Failed to launch detached gateway restart: %s"):
await self._launch_detached_restart_command()
await self._finalize_shutdown_agents(ctx.active_agents)
# Idle cached agents too: _finalize_shutdown_agents only covers agents mid-turn at drain
# time, but _agent_cache may hold agents whose MemoryProviders never got on_session_end().
# Idle cached agents too: their MemoryProviders may never have seen on_session_end().
_cache_lock = getattr(self, "_agent_cache_lock", None)
_cache = getattr(self, "_agent_cache", None)
if _cache_lock is not None and _cache is not None:
@@ -1495,13 +1454,11 @@ class GatewayShutdownMixin:
_idle_agents = list(_cache.values())
_cache.clear()
for _entry in _idle_agents:
# Bounded + off-loop so a wedged memory provider can't hang shutdown forever
# (this path is why SIGTERM once failed to kill the process).
# Bounded + off-loop: a wedged memory provider here once made SIGTERM hang forever.
await self._cleanup_agent_resources_off_loop(
_entry[0] if isinstance(_entry, tuple) else _entry, context="shutdown idle-cache"
)
# Cancel + await completion flush tasks (fan-in sleep / adapter delivery) while adapters are
# still alive so every watcher gets a retryable result before platform teardown.
# Settle completion flush tasks while adapters are alive so every watcher gets a retryable result.
cancel_completion_batches = getattr(self, "_cancel_process_completion_batch_tasks", None)
if cancel_completion_batches is not None:
await cancel_completion_batches()
@@ -1520,8 +1477,7 @@ class GatewayShutdownMixin:
"""Cancel background tasks, flush pending messages, clear per-session state, final tool kill."""
from gateway.run import GatewayRunner
for _task in list(self._background_tasks):
# _restart_task is awaiting _stop_task right now; cancelling it would propagate
# CancelledError into _stop_impl and skip _shutdown_event.set(). It self-terminates anyway.
# _restart_task awaits _stop_task: cancelling it would tunnel into _stop_impl and skip _shutdown_event.set().
if _task is self._stop_task or _task is self._restart_task:
continue
_task.cancel()
@@ -1529,36 +1485,30 @@ class GatewayShutdownMixin:
self.adapters.clear()
for _session_key in list(self._running_agents):
self._release_running_agent_state(_session_key)
# Flush pending messages to disk before clearing: under FTS5 corruption the in-memory
# pending text is the only surviving copy; clearing unflushed loses it permanently.
# Flush pending messages before clearing: under FTS5 corruption they are the only surviving copy.
with suppress(Exception):
from gateway.shutdown_flush import flush_pending_to_file
flush_pending_to_file(dict(self._pending_messages), reason="shutdown")
# The FIFO tail lives in SessionState.conversation.queued_events, not the slot dict
# above — flush it too or every follow-up parked in overflow at restart time is lost.
# The overflow FIFO tail lives in SessionState.conversation.queued_events — flush it too.
with suppress(Exception):
from gateway.shutdown_flush import flush_overflow_to_file
flush_overflow_to_file(
{_k: list(_v) for _k, _v in dict(getattr(self, "_queued_events", None) or {}).items() if _v},
reason="shutdown",
)
# On the real runner these are live SessionState views whose clear() resets one field per
# session (never a wholesale dict swap that could lose a concurrent writer's entry).
# Live SessionState views: clear() resets one field per session (never a wholesale dict swap).
self._running_agents.clear()
self._running_agents_ts.clear()
if hasattr(self, "_active_session_leases"):
self._active_session_leases.clear()
self._pending_messages.clear()
self._pending_approvals.clear()
if hasattr(self, "_busy_ack_ts"):
self._busy_ack_ts.clear()
for _attr in ("_active_session_leases", "_busy_ack_ts"): # absent on bare shutdown-path doubles
if hasattr(self, _attr):
getattr(self, _attr).clear()
self._shutdown_event.set()
# Global catch-all subprocess kill (safe to repeat): covers the graceful path and
# anything respawned since the drain-timeout path's post-interrupt kill.
# Global catch-all subprocess kill (safe to repeat) for the graceful path and late respawns.
GatewayRunner._stop_kill_tool_subprocesses("final-cleanup")
logger.info("Shutdown phase: final-cleanup tool kill done at +%.2fs", ctx.elapsed())
# Reap the process-global auxiliary-client cache: per-turn cleanup misses clients bound to
# worker-thread loops that died with their executor (cron ticks) → httpx transports leak to EMFILE.
# Reap the auxiliary-client cache: clients bound to dead worker-thread loops leak httpx transports.
def _reap_aux_clients() -> None:
from agent.auxiliary_client import shutdown_cached_clients
shutdown_cached_clients()
@@ -1568,30 +1518,25 @@ class GatewayShutdownMixin:
def _stop_quiesce_and_close_session_dbs(self, timeout: float, ctx: "GatewayShutdownMixin._StopContext") -> None:
"""Quiesce the executor, then close SessionDB handles only if no worker is still live."""
from gateway.run import GatewayRunner, _EXECUTOR_QUIESCE_TIMEOUT
# Quiesce the thread pool BEFORE closing session DBs: otherwise a coroutine can mint a fresh
# pool (``_executor_closing`` still False) or an already-started run_in_executor future keeps
# writing after ``SessionDB.close()`` checkpointed the WAL — the late write reopens the handle
# and mints a split WAL generation (close-time corruption). Bounded and clamped to what is
# left of the watchdog leash minus a second for the close itself.
# Quiesce the thread pool BEFORE closing session DBs: a late executor write after
# SessionDB.close() checkpointed the WAL reopens the handle and splits the WAL generation
# (close-time corruption). Clamped to the remaining watchdog leash minus 1s for the close.
_exec_quiesce_budget = max(
0.0, min(_EXECUTOR_QUIESCE_TIMEOUT, resolve_shutdown_watchdog_delay(timeout) - ctx.elapsed() - 1.0),
)
_exec_live = GatewayRunner._shutdown_executor(self, drain_timeout=_exec_quiesce_budget)
if _exec_live:
# A live worker may be mid-write: closing now is the exact #101093 corruption sequence,
# so skip the close and let SQLite recover from its WAL on next open (a transient
# "database is locked" on an immediate --replace at worst, not a corrupt file).
# A live worker may be mid-write (the #101093 corruption sequence): skip the close and let
# SQLite recover from its WAL on next open (at worst a transient "database is locked").
logger.warning(
"Shutdown phase: %d executor worker(s) still running after "
"a %.2fs quiesce — skipping the SessionDB close/checkpoint "
"to avoid racing a live write (#101093); handles are left "
"Shutdown phase: %d executor worker(s) still running after a %.2fs quiesce — skipping the "
"SessionDB close/checkpoint to avoid racing a live write (#101093); handles are left "
"open for SQLite to recover on next open", _exec_live, _exec_quiesce_budget,
)
return
logger.info("Shutdown phase: executor quiesced at +%.2fs", ctx.elapsed())
_step = GatewayShutdownMixin._quiet_step
# Close SQLite session DBs so the WAL lock is released; otherwise --replace leaves the old
# connection holding it until exit and the new gateway gets 'database is locked'.
# Close SQLite session DBs so --replace's new gateway does not hit 'database is locked'.
# ``_session_db`` is an AsyncSessionDB facade — unwrap; ``session_store`` holds ``_db``.
_self_db = getattr(self, "_session_db", None)
_self_db = getattr(_self_db, "_db", _self_db)
@@ -1599,18 +1544,15 @@ class GatewayShutdownMixin:
for _db in (_self_db, getattr(store, "_db", None)):
if _db is not None and hasattr(_db, "close"):
_step("SessionDB close error", _db.close)
# A multiplexed session_store caches one SessionDB per profile; ``_db`` above only covered
# the root scope. Sweep the rest so secondary WAL locks are released before --replace.
# Multiplexed session_store caches one SessionDB per profile; sweep the secondary WAL locks too.
_sweep = getattr(store, "close_all_db_handles", None)
if _sweep is not None:
_step("SessionDB handle sweep error", _sweep)
# Same sweep for the runner's own per-profile session_search handles (slash commands
# resolve them under profile scopes).
# Same sweep for the runner's own per-profile session_search handles.
_step("Runner SessionDB handle sweep error", lambda: GatewayRunner.close_all_session_db_handles(self))
def _close_shared() -> None:
# Shared SessionDB instances still held by the process-wide registry (tools, cron,
# mirror, etc. opened via get_shared_session_db but not released above).
# Shared SessionDB instances still held by the process-wide registry (tools, cron, mirror).
from hermes_state import close_shared_session_dbs
closed = close_shared_session_dbs()
if closed:
@@ -1628,20 +1570,17 @@ class GatewayShutdownMixin:
from gateway.status import remove_pid_file, release_gateway_runtime_lock
remove_pid_file()
release_gateway_runtime_lock()
# Clean-shutdown marker: suspend_recently_active() need only run after unexpected exits.
# If the drain timed out and agents were force-interrupted, sessions may be half-finished
# — skip the marker so the next startup suspends them.
# Clean-shutdown marker skips suspend_recently_active() next boot; a timed-out drain left
# half-finished sessions, so no marker — the next startup suspends them.
if not ctx.timed_out:
with suppress(Exception):
(_hermes_home / ".clean_shutdown").touch()
else:
logger.info(
"Skipping .clean_shutdown marker — drain timed out with "
"interrupted agents; next startup will suspend recently "
"active sessions."
"interrupted agents; next startup will suspend recently active sessions."
)
# Stuck-loop detection: the counter increments for sessions active at each restart; at
# the threshold (3 consecutive) the next startup auto-suspends the session.
# Stuck-loop counter: sessions active across 3 consecutive restarts are auto-suspended next boot.
if ctx.active_agents:
self._increment_restart_failure_counts(set(ctx.active_agents.keys()))
if self._restart_requested and self._restart_command_source is None:
@@ -1656,19 +1595,16 @@ class GatewayShutdownMixin:
indent=None,
)
if self._restart_requested and self._restart_via_service:
# Service manager owns restarts: exit 75 + ``RestartForceExitStatus=75`` has systemd
# replace this process without a second helper racing the unit's stop/start job.
# Exit 75 + ``RestartForceExitStatus=75``: systemd replaces us without a racing helper.
self._exit_code = GATEWAY_SERVICE_RESTART_EXIT_CODE
self._exit_reason = self._exit_reason or "Gateway restart requested"
self._draining = False
# Terminal gateway_state: "stopped" by default, "running" on an UNEXPECTED signal (s6 SIGTERM
# on docker restart, OOM-kill) — container_boot.py only auto-starts gateways last seen
# "running". Operator stops write a planned-stop marker BEFORE signalling; restarts persist "stopped".
# Terminal gateway_state: "stopped", or "running" on an UNEXPECTED signal (docker restart,
# OOM) — container_boot.py only auto-starts gateways last seen "running".
if getattr(self, "_signal_initiated_shutdown", False) and not self._restart_requested:
logger.info(
"Gateway stopped by an unexpected signal — persisting "
"gateway_state=running so container_boot auto-starts on "
"the next boot (issue #42675)"
"gateway_state=running so container_boot auto-starts on the next boot (issue #42675)"
)
self._update_runtime_status("running", self._exit_reason)
else:
@@ -1694,22 +1630,18 @@ class GatewayShutdownMixin:
async def _stop_impl(self) -> None:
"""Run every ``_stop_*`` phase under the thread-based shutdown watchdog."""
from gateway.run import GatewayRunner
# Thread-based watchdog (asyncio timeouts cannot recover a frozen loop): if teardown never
# finishes within drain+grace it dumps faulthandler stacks and os._exit so KeepAlive/systemd
# can revive. Skipped under pytest (no delayed hard-exit in the worker).
# Thread-based watchdog (asyncio timeouts cannot recover a frozen loop): dumps stacks and
# os._exit past drain+grace so the service manager revives us. Skipped under pytest.
_watchdog_done = threading.Event()
self._shutdown_watchdog_done = _watchdog_done
# Shutdown-path tests and third-party runner doubles may only implement the older
# drain-count surface (no deferred-worker counter).
# Shutdown-path doubles may lack the deferred-worker counter.
ctx = GatewayShutdownMixin._StopContext(
deferred_count=getattr(self, "_active_deferred_agent_worker_count", lambda: 0)
)
if not os.environ.get("PYTEST_CURRENT_TEST"):
arm_shutdown_watchdog(
resolve_shutdown_watchdog_delay(self._restart_drain_timeout),
done_event=_watchdog_done,
snapshot_fn=lambda: GatewayRunner._shutdown_watchdog_snapshot(self, ctx),
exit_code=1,
resolve_shutdown_watchdog_delay(self._restart_drain_timeout), done_event=_watchdog_done,
snapshot_fn=lambda: GatewayRunner._shutdown_watchdog_snapshot(self, ctx), exit_code=1,
)
try:
await GatewayRunner._stop_begin_teardown(self, ctx)
@@ -1725,8 +1657,7 @@ class GatewayShutdownMixin:
_watchdog_done.set()
async def stop(
self, *, restart: bool = False, detached_restart: bool = False,
service_restart: bool = False,
self, *, restart: bool = False, detached_restart: bool = False, service_restart: bool = False
) -> None:
"""Stop the gateway and disconnect all adapters."""
from gateway.run import GatewayRunner
gateway/run_startup.py
+172 -360
View File
File diff suppressed because it is too large Load Diff