1544 lines
87 KiB
Python
1544 lines
87 KiB
Python
"""Startup sequence, resume/restore and handoff methods for GatewayRunner.
|
|
|
|
Split out of ``gateway/run.py``; bound onto ``GatewayRunner`` via the MRO.
|
|
``gateway.run`` internals are imported lazily inside method bodies (import cycle),
|
|
so ``patch("gateway.run.X")`` keeps intercepting them at call time.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import dataclasses
|
|
import faulthandler
|
|
import logging
|
|
import os
|
|
import signal
|
|
import time
|
|
from contextlib import suppress
|
|
from datetime import datetime
|
|
from gateway.config import Platform
|
|
from gateway.delivery import looks_like_telegram_private_chat_id
|
|
from gateway.platforms.base import BasePlatformAdapter
|
|
from gateway.platforms.event import MessageEvent, MessageType
|
|
from gateway.session import SessionSource, build_session_key
|
|
from gateway.restart import (
|
|
DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT, GATEWAY_FATAL_CONFIG_EXIT_CODE, is_global_startup_conflict
|
|
)
|
|
from gateway.run_shutdown import _log_suppressed, _send_error
|
|
from gateway.shutdown_watchdog import (
|
|
DEFAULT_HEARTBEAT_INTERVAL_S, DEFAULT_LOOP_WATCHDOG_INTERVAL_S,
|
|
DEFAULT_LOOP_WATCHDOG_MAX_STRIKES, DEFAULT_LOOP_WATCHDOG_TIMEOUT_S, loop_heartbeat_forever,
|
|
)
|
|
from typing import Any, Dict, Optional, Tuple
|
|
|
|
# Log-record parity with the origin module.
|
|
logger = logging.getLogger("gateway.run")
|
|
|
|
|
|
class GatewayStartupMixin:
|
|
"""Startup sequence, resume/restore and handoff methods for GatewayRunner."""
|
|
|
|
async def _run_startup_resume_event(
|
|
self, adapter: BasePlatformAdapter, event: MessageEvent, session_key: str,
|
|
) -> None:
|
|
"""Dispatch one synthetic startup resume and wait for its agent turn (inbound stays queued
|
|
until it finishes, else a user message can race it)."""
|
|
from gateway.run import _AGENT_PENDING_SENTINEL
|
|
try:
|
|
await adapter.handle_message(event)
|
|
session_tasks = getattr(adapter, "_session_tasks", {})
|
|
task = session_tasks.get(session_key) if isinstance(session_tasks, dict) else None
|
|
if task is not None:
|
|
await asyncio.shield(task)
|
|
finally:
|
|
# Release the pre-claimed slot if handle_message raised before _handle_message took ownership.
|
|
_pre_state = self._peek_session_state(session_key)
|
|
if (_pre_state.turn.agent if _pre_state else None) is _AGENT_PENDING_SENTINEL:
|
|
self._release_running_agent_state(session_key)
|
|
|
|
def _queue_startup_restore_event(self, event: MessageEvent) -> None:
|
|
queue = getattr(self, "_startup_restore_queue", None)
|
|
if queue is None:
|
|
queue = self._startup_restore_queue = []
|
|
queue.append(event)
|
|
with suppress(Exception):
|
|
source = event.source
|
|
logger.info(
|
|
"Queued inbound message during gateway startup restore: platform=%s chat=%s",
|
|
source.platform.value if source and source.platform else "unknown",
|
|
source.chat_id if source else "unknown",
|
|
)
|
|
|
|
async def _drain_startup_restore_queue(self) -> int:
|
|
"""Replay inbound messages queued while startup auto-resume ran."""
|
|
drained = 0
|
|
queue = getattr(self, "_startup_restore_queue", None) or []
|
|
while queue:
|
|
event = queue.pop(0)
|
|
source = getattr(event, "source", None)
|
|
adapter = self._adapter_for_source(source)
|
|
if adapter is None:
|
|
logger.debug(
|
|
"Dropping startup-restore queued message: adapter unavailable for %s",
|
|
getattr(getattr(source, "platform", None), "value", None),
|
|
)
|
|
continue
|
|
# Mark the replay so _handle_message does not re-queue it while the restore gate is closed.
|
|
with suppress(Exception):
|
|
setattr(event, "_hermes_startup_restore_replay", True)
|
|
await adapter.handle_message(event)
|
|
drained += 1
|
|
return drained
|
|
|
|
@staticmethod
|
|
def _start_free_tier_bootstrap() -> None:
|
|
"""One bootstrap per process. `run_bootstrap` already records its own failure in the boot record
|
|
and never raises, so this is a plain call; it exists as a method so tests can seam it."""
|
|
from hermes_cli.free_tier_bootstrap import run_bootstrap
|
|
run_bootstrap(announce=False)
|
|
|
|
def _start_startup_warmup(self) -> None:
|
|
"""Kick off the boot turn-machinery warm-up so it overlaps the network-bound platform
|
|
connects; ``_finish_startup_restore`` awaits it (bounded)."""
|
|
from gateway.run import _startup_warmup_timeout_secs
|
|
if _startup_warmup_timeout_secs() <= 0:
|
|
self._startup_warmup_task = None
|
|
return
|
|
self._startup_warmup_task = asyncio.ensure_future(self._warm_turn_prerequisites())
|
|
|
|
async def _warm_turn_prerequisites(self) -> None:
|
|
"""Initialize turn machinery on an executor thread before the gate opens. Never raises: a
|
|
failed warm-up degrades to lazy init and must not block startup."""
|
|
from gateway.run import _warm_turn_machinery_sync
|
|
with _log_suppressed(
|
|
logging.WARNING, "Turn-machinery warm-up failed; first inbound turn will initialize lazily",
|
|
exc_info=True,
|
|
):
|
|
loop = asyncio.get_running_loop()
|
|
t0 = time.monotonic()
|
|
tool_count = await loop.run_in_executor(None, _warm_turn_machinery_sync)
|
|
logger.info(
|
|
"Turn machinery warmed in %.1fs (%d tool schema(s) materialized)",
|
|
time.monotonic() - t0, tool_count,
|
|
)
|
|
|
|
async def _await_startup_warmup(self) -> None:
|
|
"""Bounded wait for the boot warm-up. On timeout the gate opens anyway (availability outranks
|
|
prompt completeness for a WEDGED init); the warm-up continues and a late failure is logged."""
|
|
from gateway.run import _startup_warmup_timeout_secs
|
|
task = getattr(self, "_startup_warmup_task", None)
|
|
if task is None or task.done():
|
|
return
|
|
timeout = _startup_warmup_timeout_secs()
|
|
if timeout <= 0:
|
|
return
|
|
await self._wait_bounded_or_release(
|
|
{task}, timeout,
|
|
"Turn-machinery warm-up still running after %.0fs; opening inbound gate anyway — the "
|
|
"first turn may see lazily initialized machinery (#99373). Warm-up continues in the background.",
|
|
"boot turn-machinery warm-up failed after gate release", level=logging.DEBUG,
|
|
)
|
|
|
|
async def _wait_bounded_or_release(
|
|
self, tasks: set, timeout: float, warn_fmt: str, late_msg: str, *,
|
|
level: int = logging.WARNING, track: bool = False,
|
|
) -> set:
|
|
"""``asyncio.wait`` (which, unlike wait_for/gather+timeout, does NOT cancel on timeout) with
|
|
the gate-release warning + late-failure callback every boot gate uses. Returns ``done``.
|
|
``warn_fmt`` takes ``%.0fs`` (timeout) and optionally ``%d`` (pending count). A non-positive
|
|
``timeout`` opts out of the bound ("wait forever")."""
|
|
if timeout <= 0:
|
|
await asyncio.gather(*tasks, return_exceptions=True)
|
|
return set(tasks)
|
|
done, pending = await asyncio.wait(tasks, timeout=timeout)
|
|
if pending:
|
|
args = (timeout, len(pending)) if warn_fmt.count("%") > 1 else (timeout,)
|
|
logger.warning(warn_fmt, *args)
|
|
late = self._late_failure_callback(late_msg, level=level)
|
|
for task in pending:
|
|
task.add_done_callback(late)
|
|
if track:
|
|
self._retain_background_task(task)
|
|
return done
|
|
|
|
async def _finish_startup_restore(self) -> None:
|
|
"""Wait (BOUNDED by ``_startup_restore_drain_timeout_secs``) for startup auto-resume, then
|
|
release + drain inbound. On timeout the gate opens and resume turns finish in the background
|
|
(NOT cancelled) — safe because ``_schedule_resume_pending_sessions`` claims each
|
|
``_running_agents`` slot SYNCHRONOUSLY first, so drained inbound queues behind."""
|
|
from gateway.run import _startup_restore_drain_timeout_secs
|
|
tasks = list(getattr(self, "_startup_restore_tasks", []) or [])
|
|
if tasks:
|
|
# Tasks outliving the gate get a late-failure callback (their done-callback only discards them).
|
|
done = await self._wait_bounded_or_release(
|
|
set(tasks), _startup_restore_drain_timeout_secs(),
|
|
"Startup-restore gate released after %.0fs with %d boot auto-resume turn(s) "
|
|
"still running; draining inbound queue now (resume slots already claimed, so no "
|
|
"duplicate agents). Slow turn(s) continue in the background.",
|
|
"background startup auto-resume task failed after gate release", level=logging.DEBUG,
|
|
)
|
|
report = self._late_failure_callback("startup auto-resume task failed", level=logging.DEBUG)
|
|
for task in done:
|
|
report(task)
|
|
self._startup_restore_tasks = []
|
|
# Warm the turn machinery BEFORE the queue drains: inbound turns must not build skeleton prompts.
|
|
await self._await_startup_warmup()
|
|
drained = await self._drain_startup_restore_queue()
|
|
self._startup_restore_in_progress = False
|
|
if drained:
|
|
logger.info("Drained %d inbound message(s) queued during startup restore", drained)
|
|
|
|
@staticmethod
|
|
def _late_failure_callback(message: str, *, level: int = logging.WARNING):
|
|
"""Done-callback for boot-path tasks that outlive the startup-restore gate: surface a late
|
|
failure otherwise swallowed once the task leaves ``_background_tasks``. Cancellation is
|
|
expected (shutdown), not an error."""
|
|
def _report(task: "asyncio.Task") -> None:
|
|
if task.cancelled():
|
|
return
|
|
exc = task.exception()
|
|
if exc is not None:
|
|
logger.log(level, message, exc_info=(type(exc), exc, exc.__traceback__))
|
|
return _report
|
|
|
|
async def _await_startup_boot_sends(self, *, planned_restart_notification_pending: bool) -> None:
|
|
"""Run boot-path sends without letting them pin the inbound restore gate (one Telegram
|
|
flood-control sleep must not freeze inbound on every platform): same bounded wait as the
|
|
resume gate, sends finish in the background on timeout. The ledger claim + ``resume_pending``
|
|
clear run INLINE before the send task exists — deferring it let a hung notification expire the
|
|
gate with zero rows claimed, so answered turns were replayed AND redelivered.
|
|
|
|
``_send_restart_notification`` and ``_redeliver_pending_obligations`` used to be awaited inline
|
|
*before* ``_finish_startup_restore`` released the gate. See #91969.
|
|
"""
|
|
from gateway.run import _clear_planned_restart_notification, _startup_restore_drain_timeout_secs
|
|
claimed = await self._claim_pending_obligations()
|
|
|
|
async def _boot_sends() -> None:
|
|
await self._send_restart_notification()
|
|
if planned_restart_notification_pending:
|
|
try:
|
|
await self._send_home_channel_startup_notifications(skip_targets=None)
|
|
finally:
|
|
_clear_planned_restart_notification()
|
|
await self._redeliver_claimed_obligations(claimed)
|
|
|
|
boot_task = asyncio.create_task(_boot_sends())
|
|
timeout = _startup_restore_drain_timeout_secs()
|
|
if timeout <= 0:
|
|
await boot_task # unbounded: a failing send surfaces here (unlike the gate path)
|
|
return
|
|
await self._wait_bounded_or_release(
|
|
{boot_task}, timeout,
|
|
"Boot-path sends still running after %.0fs; releasing inbound gate so other platforms are not "
|
|
"frozen. Restart notification / obligation redelivery continue in the background.",
|
|
"background boot-path send failed after gate release: see traceback", track=True,
|
|
)
|
|
|
|
async def _clear_resume_pending_for_claimed_obligations(
|
|
self, claimed: list, *, require_success: bool = False
|
|
) -> list:
|
|
"""Clear resume flags and return rows safe to redeliver. Startup recovery is best-effort;
|
|
runtime reconnect recovery (``require_success``) is stricter: if the session-store write
|
|
fails the response must not be sent, or the turn could be resumed too."""
|
|
sendable = []
|
|
for row in claimed:
|
|
session_key = row.get("session_key") or ""
|
|
if not session_key:
|
|
sendable.append(row)
|
|
continue
|
|
try:
|
|
await self.async_session_store.clear_resume_pending(session_key)
|
|
except Exception:
|
|
logger.debug("clear_resume_pending failed for %s", session_key, exc_info=True)
|
|
if require_success:
|
|
continue
|
|
sendable.append(row)
|
|
return sendable
|
|
|
|
async def _claim_pending_obligations(self) -> list:
|
|
"""Claim recoverable delivery-ledger rows and clear their ``resume_pending`` flags (pure DB
|
|
work, no sends). Must run INLINE BEFORE ``_schedule_resume_pending_sessions`` and the
|
|
abandonable boot-send task: these sessions already produced their answer, so the resume path
|
|
must not re-run (re-pay for) the turn however long the sends take. Mid-send / rejected rows
|
|
carry a visible recovered-reply marker (gateway/delivery_ledger.py). Returns the rows.
|
|
|
|
A session with a recoverable obligation already produced its answer — the turn completed and only
|
|
delivery is owed — so clearing ``resume_pending`` here prevents the resume path from re-running (and
|
|
re-paying for) a turn whose output we hold, regardless of how long the sends ahead of redelivery
|
|
take (#91969). Flood-refused rows adopted inside their wait come back flagged ``adopted``: their
|
|
flags are cleared here too, and ``_redeliver_claimed_obligations`` leaves them to the timer.
|
|
"""
|
|
try:
|
|
from gateway.delivery_ledger import ledger_enabled, sweep_recoverable
|
|
if not await asyncio.to_thread(ledger_enabled):
|
|
return []
|
|
# Claim only rows whose exact transport owner is connected: platform-only filtering would spend
|
|
# a disconnected bot's retry budget because another bot on that platform is online.
|
|
_profile_adapters = getattr(self, "_profile_adapters", None) or {}
|
|
_pval = lambda p: getattr(p, "value", str(p)) # noqa: E731
|
|
_deliverable_targets = {(_pval(p), "default") for p in self.adapters}
|
|
# Legacy rows (no adapter_profile) are unambiguous only without multiplexing; else fail closed.
|
|
if not _profile_adapters:
|
|
_deliverable_targets.update((_pval(p), None) for p in self.adapters)
|
|
for _profile, _adapters in _profile_adapters.items():
|
|
_deliverable_targets.update((_pval(p), _profile) for p in _adapters)
|
|
claimed = await asyncio.to_thread(
|
|
sweep_recoverable, None,
|
|
deliverable_platforms={platform for platform, _ in _deliverable_targets},
|
|
deliverable_targets=_deliverable_targets,
|
|
)
|
|
except Exception:
|
|
logger.debug("delivery ledger sweep failed", exc_info=True)
|
|
return []
|
|
if not claimed:
|
|
return []
|
|
# Clear resume_pending for EVERY claimed row before any send: the answer is in the ledger.
|
|
# Claiming already spent one of the row's redelivery attempts — the answer is in the ledger, so the
|
|
# resume path must never re-run these turns (#91969).
|
|
await self._clear_resume_pending_for_claimed_obligations(claimed)
|
|
return claimed
|
|
|
|
@staticmethod
|
|
async def _release_runtime_claim_quiet(obligation_id, log_fmt: str, error: str = "send_path_degraded") -> None:
|
|
"""Release an unsent runtime delivery-ledger claim; log-only on failure. ``error`` is what the row
|
|
goes back to ``failed`` with: the claim's own pre-claim error when the caller knows it, so a
|
|
flood-refused row keeps its ``flood_control:<seconds>`` and stays on the flood timer's list (the
|
|
release re-stamps ``updated_at``, so it waits the platform's figure once more); else
|
|
``send_path_degraded``."""
|
|
from gateway.delivery_ledger import release_runtime_claim
|
|
try:
|
|
await asyncio.to_thread(release_runtime_claim, obligation_id, error)
|
|
except Exception:
|
|
logger.debug(log_fmt, obligation_id, exc_info=True)
|
|
|
|
def _schedule_flood_redelivery(self, platform, *, profile: Optional[str] = None) -> None:
|
|
"""Wake one deadline-driven ledger worker per bot identity, never sleep in a send."""
|
|
from gateway.delivery_ledger import flood_retry_delay, pending_flood_retries
|
|
target = platform if isinstance(platform, Platform) else Platform(str(platform))
|
|
key = (target.value, profile or "default")
|
|
pending = getattr(self, "_flood_redelivery_tasks", None)
|
|
if pending is None:
|
|
pending = self._flood_redelivery_tasks = {}
|
|
self._flood_redelivery_wakes = {}
|
|
wakes = self._flood_redelivery_wakes
|
|
if key in pending:
|
|
# Remember refusals arriving during a threaded SELECT or a send. An empty stale
|
|
# snapshot cannot retire this worker until it has observed the wake.
|
|
wakes[key].set()
|
|
return None
|
|
wake = wakes[key] = asyncio.Event()
|
|
|
|
async def _redeliver_after_wait():
|
|
try:
|
|
while getattr(self, "_running", False):
|
|
wake.clear()
|
|
waiting = await asyncio.to_thread(pending_flood_retries)
|
|
deadlines = [r["not_before"] for r in waiting
|
|
if (r["platform"], r["profile"]) == key]
|
|
if not deadlines:
|
|
if wake.is_set():
|
|
continue
|
|
return
|
|
delay = flood_retry_delay(min(deadlines) - time.time())
|
|
try:
|
|
await asyncio.wait_for(wake.wait(), timeout=delay)
|
|
continue # A shorter sibling may now be due first.
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
if getattr(self, "_running", False):
|
|
await self._redeliver_failed_obligations_for_platform(target, profile=profile)
|
|
finally:
|
|
pending.pop(key, None)
|
|
wakes.pop(key, None)
|
|
|
|
task = asyncio.create_task(_redeliver_after_wait(), name="flood-redelivery:%s:%s" % key)
|
|
pending[key] = task
|
|
# The gateway's ordinary shutdown drain must cancel sleeping timers too.
|
|
background = getattr(self, "_background_tasks", None)
|
|
if background is not None:
|
|
self._track_task_in(background, task)
|
|
|
|
async def _arm_flood_timers_for_waiting_rows(self) -> None:
|
|
"""Recover adopted, newly refused and unsent released rows without blocking the loop."""
|
|
from gateway.delivery_ledger import pending_flood_retries
|
|
for row in await asyncio.to_thread(pending_flood_retries):
|
|
self._schedule_flood_redelivery(row["platform"], profile=row["profile"])
|
|
|
|
async def _redeliver_claimed_obligations(self, claimed: list) -> int:
|
|
"""Redeliver final responses for claimed rows (network half of the split): runs inside the
|
|
bounded boot-send task, so a flood-limited send can be abandoned by the restore gate without
|
|
reopening the turn-replay window. Returns the redelivered count."""
|
|
# No early return on an empty claim: the boot sweep may have ADOPTED flood-refused rows that are
|
|
# not due yet, and those still need their timer armed below.
|
|
try:
|
|
from gateway.delivery_ledger import RECOVERED_MARKER, mark_delivered, mark_failed
|
|
except Exception:
|
|
logger.debug("delivery ledger import failed", exc_info=True)
|
|
return 0
|
|
redelivered = 0
|
|
for row in claimed:
|
|
if row.get("adopted"):
|
|
# Adopted at boot inside its flood wait: its resume flag is cleared with the others, and
|
|
# the timer armed below sends it once the platform's deadline has passed.
|
|
continue
|
|
adapter = await self._obligation_adapter(row)
|
|
if adapter is None:
|
|
continue
|
|
content = row["content"]
|
|
if row.get("needs_marker"):
|
|
content = row.get("marker", RECOVERED_MARKER) + content
|
|
metadata = {"thread_id": row["thread_id"]} if row.get("thread_id") else None
|
|
try:
|
|
result = await adapter.send(chat_id=row["chat_id"], content=content, metadata=metadata)
|
|
except Exception as send_err:
|
|
logger.warning("obligation %s: redelivery send raised: %s", row["obligation_id"], send_err)
|
|
result = None
|
|
with _log_suppressed(logging.DEBUG, "delivery ledger update failed", exc_info=True):
|
|
if result is not None and getattr(result, "success", False):
|
|
await asyncio.to_thread(mark_delivered, row["obligation_id"])
|
|
redelivered += 1
|
|
logger.info(
|
|
"Redelivered recovered final response to %s:%s (obligation %s, attempt %d)",
|
|
row["platform"], row["chat_id"], row["obligation_id"], row["attempts"],
|
|
)
|
|
else:
|
|
await asyncio.to_thread(
|
|
mark_failed, row["obligation_id"], str(getattr(result, "error", "") or "send failed")
|
|
)
|
|
# Whatever is still waiting on a flood penalty (adopted at boot, skipped as not yet due, refused
|
|
# again just now) gets a timer, so no flood-refused reply waits for the next restart.
|
|
with _log_suppressed(logging.DEBUG, "arming flood redelivery timers failed", exc_info=True):
|
|
await self._arm_flood_timers_for_waiting_rows()
|
|
return redelivered
|
|
|
|
async def _obligation_adapter(self, row: dict):
|
|
"""Resolve the adapter for a claimed ledger row, or None when it cannot be delivered now."""
|
|
try:
|
|
platform = Platform(row["platform"])
|
|
except Exception:
|
|
logger.debug("obligation %s: unknown platform %r", row["obligation_id"], row.get("platform"))
|
|
return None
|
|
if "profile" in row:
|
|
adapter = self._authorization_adapter(platform, row.get("profile"))
|
|
else:
|
|
# Startup rows preserve the historical default-adapter route.
|
|
adapter = self.adapters.get(platform)
|
|
# A runtime claim whose reconnect vanished before dispatch is released without spending an
|
|
# attempt; startup claims keep their state (attempts cap + stale cutoff bound retries).
|
|
if adapter is None and row.get("runtime_recovery"):
|
|
await self._release_runtime_claim_quiet(
|
|
row["obligation_id"], "failed to release undispatched runtime obligation %s",
|
|
error=row.get("last_error") or "send_path_degraded",
|
|
)
|
|
return adapter
|
|
|
|
async def _redeliver_pending_obligations(self) -> int:
|
|
"""Claim + redeliver in one call. Stable shape for tests/external callers; the startup path
|
|
calls the halves separately so the DB half runs inline before the abandonable send task."""
|
|
return await self._redeliver_claimed_obligations(await self._claim_pending_obligations())
|
|
|
|
async def _redeliver_failed_obligations_for_platform(
|
|
self, platform: Platform, *, profile: Optional[str] = None,
|
|
) -> int:
|
|
"""Replay one adapter identity's transient failures after reconnect: the startup sweep cannot
|
|
claim live-owner rows, so ``send_path_degraded`` responses would otherwise stay failed until
|
|
the next restart. Best-effort; reuses the startup redelivery contract."""
|
|
try:
|
|
from gateway.delivery_ledger import ledger_enabled, sweep_failed_for_runtime
|
|
if not await asyncio.to_thread(ledger_enabled):
|
|
return 0
|
|
claimed = await asyncio.to_thread(sweep_failed_for_runtime, platform.value, profile=profile)
|
|
except Exception:
|
|
logger.debug(
|
|
"runtime delivery ledger sweep failed after %s reconnect", platform.value, exc_info=True,
|
|
)
|
|
return 0
|
|
if not claimed:
|
|
return 0
|
|
# Clear before any send so the reconnect path cannot both redeliver AND resume the same turn.
|
|
sendable = await self._clear_resume_pending_for_claimed_obligations(claimed, require_success=True)
|
|
sendable_ids = {row["obligation_id"] for row in sendable}
|
|
for row in claimed:
|
|
if row["obligation_id"] not in sendable_ids:
|
|
await self._release_runtime_claim_quiet(
|
|
row["obligation_id"], "failed to release runtime delivery claim %s",
|
|
error=row.get("last_error") or "send_path_degraded",
|
|
)
|
|
return await self._redeliver_claimed_obligations(sendable)
|
|
|
|
def _resume_pending_candidates(self, platform=None) -> Optional[list]:
|
|
"""Snapshot resume-pending entries (optionally scoped to ``platform``); None when
|
|
enumeration failed or the restart-loop breaker tripped for this boot."""
|
|
try:
|
|
with self.session_store._lock: # noqa: SLF001 — snapshot under lock
|
|
self.session_store._ensure_loaded_locked() # noqa: SLF001
|
|
candidates = [
|
|
entry for entry in self.session_store._entries.values() # noqa: SLF001
|
|
if entry.resume_pending
|
|
and not entry.suspended
|
|
and entry.origin is not None
|
|
and entry.resume_reason in self._AUTO_RESUME_REASONS
|
|
and (platform is None or entry.origin.platform == platform)
|
|
]
|
|
except Exception as exc:
|
|
logger.warning("Failed to enumerate resume-pending sessions: %s", exc)
|
|
return None
|
|
# Restart-loop breaker: only boots WITH restart-interrupted sessions count; when tripped, skip
|
|
# auto-resume for THIS boot only (inbound still served; sessions stay resume_pending).
|
|
if candidates:
|
|
try:
|
|
from gateway import restart_loop_guard as _rlg
|
|
_max_restarts, _window, _max_gap = self._restart_loop_guard_config()
|
|
if _rlg.check_and_record(_max_restarts, _window, max_gap_seconds=_max_gap):
|
|
return None
|
|
except Exception as exc: # noqa: BLE001 — breaker must fail OPEN
|
|
logger.debug("Restart-loop guard check skipped: %s", exc)
|
|
return candidates
|
|
|
|
def _resume_owner_authorized(self, session_key: str, source) -> bool:
|
|
"""Validate the session owner against the CURRENT allowlist: a session created before the
|
|
allowlist existed (or whose owner was since removed) must not silently receive a full agent
|
|
response just because it carries a resume marker."""
|
|
try:
|
|
if self._is_user_authorized(source):
|
|
return True
|
|
logger.warning(
|
|
"Skipping auto-resume for %s: session owner is no "
|
|
"longer authorized under the current allowlist", session_key,
|
|
)
|
|
except Exception as exc:
|
|
logger.warning("Skipping auto-resume for %s: authorization check failed: %s", session_key, exc)
|
|
return False
|
|
|
|
def _schedule_resume_pending_sessions(self, platform=None) -> int:
|
|
"""Auto-continue fresh restart-interrupted sessions: synthesize an empty-text turn (the
|
|
``_is_resume_pending`` injection path owns the wording). Sessions whose adapter is offline stay
|
|
``resume_pending`` for the reconnect watcher, which re-calls this scoped to that ``platform``;
|
|
sessions with a running agent are skipped so none is resumed twice."""
|
|
from gateway.run import _AGENT_PENDING_SENTINEL, _auto_continue_freshness_window
|
|
window = _auto_continue_freshness_window()
|
|
candidates = self._resume_pending_candidates(platform)
|
|
if candidates is None:
|
|
return 0
|
|
now = datetime.now()
|
|
scheduled = 0
|
|
for entry in candidates:
|
|
marker = entry.last_resume_marked_at or entry.updated_at
|
|
if marker is not None and (now - marker).total_seconds() > window:
|
|
continue
|
|
# Already being resumed (e.g. scheduled at startup, still in-flight) — no second turn.
|
|
if self._is_session_running(entry.session_key):
|
|
continue
|
|
source = entry.origin
|
|
adapter = self._adapter_for_source(source)
|
|
if adapter is None:
|
|
logger.debug(
|
|
"Skipping auto-resume for %s: adapter not ready for %s", entry.session_key,
|
|
getattr(source.platform, "value", source.platform),
|
|
)
|
|
continue
|
|
if not self._resume_owner_authorized(entry.session_key, source):
|
|
continue
|
|
# Claim the slot *before* spawning so an inbound message arriving before the task's first
|
|
# await queues instead of building a duplicate AIAgent.
|
|
_resume_state = self._session_state(entry.session_key)
|
|
_resume_state.turn.agent = _AGENT_PENDING_SENTINEL
|
|
_resume_state.turn.started_ts = time.time()
|
|
self._persist_active_agents()
|
|
# Empty-text internal event: the _is_resume_pending branch prepends the reason-aware note.
|
|
event = MessageEvent(text="", message_type=MessageType.TEXT, source=source, internal=True)
|
|
task = self._retain_background_task(
|
|
asyncio.create_task(self._run_startup_resume_event(adapter, event, entry.session_key))
|
|
)
|
|
if getattr(self, "_startup_restore_in_progress", False):
|
|
tasks = getattr(self, "_startup_restore_tasks", None)
|
|
if tasks is None:
|
|
tasks = self._startup_restore_tasks = []
|
|
tasks.append(task)
|
|
scheduled += 1
|
|
if scheduled:
|
|
logger.info("Scheduled auto-resume for %d restart-interrupted session(s)", scheduled)
|
|
return scheduled
|
|
|
|
def _startup_should_abort(self) -> bool:
|
|
return self._restart_requested or self._draining or self._shutdown_event.is_set()
|
|
|
|
async def _startup_teardown_adapter(self, adapter, platform) -> None:
|
|
"""Cancel an adapter's background tasks (best-effort) then disconnect it."""
|
|
with _log_suppressed(logging.DEBUG, "✗ %s background-task cancel error: %s", platform.value):
|
|
await adapter.cancel_background_tasks()
|
|
await self._safe_adapter_disconnect(adapter, platform)
|
|
|
|
def _startup_retry_entry(self, platform, adapter, platform_config, *, queued: bool = True) -> dict:
|
|
"""``_failed_platforms`` entry for a platform that failed at startup (first retry in 30s)."""
|
|
return self._reconnect_queue_entry(platform, adapter, platform_config, attempts=1, delay=30, queued=queued)
|
|
|
|
async def _abort_startup_if_shutdown_requested(
|
|
self, adapter: Optional[BasePlatformAdapter] = None, platform: Optional[Platform] = None
|
|
) -> bool:
|
|
"""Clean up and exit startup when restart/shutdown begins mid-startup."""
|
|
if not self._startup_should_abort():
|
|
return False
|
|
if adapter is not None and platform is not None:
|
|
await self._startup_teardown_adapter(adapter, platform)
|
|
stop_task = self._stop_task
|
|
current_task = asyncio.current_task()
|
|
if stop_task is not None and stop_task is not current_task:
|
|
await stop_task
|
|
elif not self._shutdown_event.is_set():
|
|
await self.stop(
|
|
restart=self._restart_requested, detached_restart=self._restart_detached,
|
|
service_restart=self._restart_via_service,
|
|
)
|
|
return True
|
|
|
|
def _start_loop_liveness_guards(self, loop: asyncio.AbstractEventLoop) -> None:
|
|
"""Arm the selector floor and out-of-loop watchdog before adapters. Disabled entirely with
|
|
``gateway.loop_watchdog: false`` in config.yaml (config-only knob).
|
|
|
|
See #69089.
|
|
"""
|
|
from gateway.shutdown_watchdog import _arm_loop_floor_timer, start_loop_liveness_watchdog
|
|
config = getattr(self, "config", None)
|
|
if config is not None and not getattr(config, "loop_watchdog", True):
|
|
return
|
|
if getattr(self, "_loop_floor_timer_handle", None) is None:
|
|
with _log_suppressed(logging.DEBUG, "Failed to arm gateway loop floor timer", exc_info=True):
|
|
self._loop_floor_timer_handle = _arm_loop_floor_timer(loop)
|
|
watchdog = getattr(self, "_loop_liveness_watchdog", None)
|
|
if watchdog is None or not watchdog.is_alive():
|
|
try:
|
|
# getattr defaults cover config=None test paths; loaded values are already clamped.
|
|
self._loop_liveness_watchdog = start_loop_liveness_watchdog(
|
|
loop,
|
|
probe_interval=float(getattr(
|
|
config, "loop_watchdog_probe_interval_s", DEFAULT_LOOP_WATCHDOG_INTERVAL_S
|
|
)),
|
|
probe_timeout=float(getattr(
|
|
config, "loop_watchdog_probe_timeout_s", DEFAULT_LOOP_WATCHDOG_TIMEOUT_S
|
|
)),
|
|
max_strikes=int(getattr(
|
|
config, "loop_watchdog_max_strikes", DEFAULT_LOOP_WATCHDOG_MAX_STRIKES
|
|
)),
|
|
)
|
|
except Exception:
|
|
logger.debug("Failed to start gateway loop liveness watchdog", exc_info=True)
|
|
|
|
def _stop_loop_liveness_guards(self) -> None:
|
|
"""Disarm lifetime liveness guards before shutdown can load the loop — including the
|
|
heartbeat writer, which would otherwise make a draining gateway look healthy to probes."""
|
|
for attr, method, what in (
|
|
("_loop_liveness_watchdog", "stop", "stop gateway loop liveness watchdog"),
|
|
("_loop_floor_timer_handle", "cancel", "cancel gateway loop floor timer"),
|
|
("_loop_heartbeat_task", "cancel", "cancel gateway loop heartbeat task"),
|
|
):
|
|
guard = getattr(self, attr, None)
|
|
setattr(self, attr, None)
|
|
if guard is not None:
|
|
with _log_suppressed(logging.DEBUG, "Failed to %s", what, exc_info=True):
|
|
getattr(guard, method)()
|
|
|
|
async def _consume_clean_shutdown_marker(self, marker_path) -> int:
|
|
"""Discard orphan turn markers before consuming a clean-exit receipt. Raises (fail closed):
|
|
continuing with the old receipt would let a later unclean exit masquerade as clean."""
|
|
discarded = await self.async_session_store.discard_active_turn_markers()
|
|
marker_path.unlink()
|
|
return discarded
|
|
|
|
async def _recover_unclean_sessions(self) -> tuple[int, int]:
|
|
"""Recover exact active turns, then run the legacy recency fallback."""
|
|
from gateway.run import _float_env
|
|
exact = 0
|
|
fallback = 0
|
|
with _log_suppressed(logging.WARNING, "Exact active-turn recovery on startup failed: %s"):
|
|
agent_timeout = max(1.0, _float_env("HERMES_AGENT_TIMEOUT", 1800))
|
|
exact = await self.async_session_store.recover_interrupted_turns(
|
|
max_age_seconds=max(60 * 60, int(agent_timeout * 2))
|
|
)
|
|
with _log_suppressed(logging.WARNING, "Legacy session recovery on startup failed: %s"):
|
|
fallback = await self.async_session_store.suspend_recently_active(max_age_seconds=120)
|
|
return exact, fallback
|
|
|
|
@staticmethod
|
|
def _start_hosted_room_worker_sync():
|
|
"""Start the local Group Chat worker without importing the dashboard."""
|
|
import tui_gateway.server # noqa: F401
|
|
from tui_gateway import methods_groups
|
|
service = methods_groups.get_hosted_room_service()
|
|
if service is None:
|
|
service = methods_groups.start_hosted_room_service()
|
|
if service is None:
|
|
raise RuntimeError("Group Chat worker has no bound session backend")
|
|
status = service.runtime.status()
|
|
if not status.get("running") or status.get("stopping"):
|
|
raise RuntimeError("Group Chat worker did not start")
|
|
return service
|
|
|
|
async def _ensure_hosted_room_worker(self):
|
|
return await asyncio.to_thread(self._start_hosted_room_worker_sync)
|
|
|
|
async def _hosted_room_worker_watcher(self, interval: float = 1.0) -> None:
|
|
"""Keep the room worker alive for the messaging gateway lifetime."""
|
|
while self._running:
|
|
await self._ensure_hosted_room_worker()
|
|
await asyncio.sleep(interval)
|
|
|
|
async def _stop_hosted_room_worker(self, timeout: float = 5.0) -> bool:
|
|
"""Pause room execution durably without interrupting accepted turns."""
|
|
from tui_gateway import methods_groups
|
|
return await asyncio.to_thread(methods_groups.stop_hosted_room_service, timeout=timeout)
|
|
|
|
def _start_loop_heartbeat_task(self) -> None:
|
|
"""Start the loop-liveness heartbeat task (idempotent, best-effort). An asyncio task so a
|
|
frozen loop stops refreshing ``state/gateway.heartbeat``; cancelled with the others in stop().
|
|
|
|
See #66892.
|
|
"""
|
|
with _log_suppressed(logging.DEBUG, "Failed to start gateway loop heartbeat", exc_info=True):
|
|
_existing_hb = getattr(self, "_loop_heartbeat_task", None)
|
|
if _existing_hb is not None and not _existing_hb.done():
|
|
return
|
|
task = self._loop_heartbeat_task = asyncio.create_task(
|
|
loop_heartbeat_forever(
|
|
interval_s=DEFAULT_HEARTBEAT_INTERVAL_S,
|
|
start_time=getattr(self, "_gateway_started_at", 0.0),
|
|
)
|
|
)
|
|
# PERMANENT watcher tag so the scale-to-zero idle check doesn't count it as busy forever.
|
|
task._hermes_supervised_watcher = True # type: ignore[attr-defined]
|
|
_bg = getattr(self, "_background_tasks", None)
|
|
if _bg is not None:
|
|
self._track_task_in(_bg, task)
|
|
|
|
def _open_faulthandler_log(self):
|
|
"""Open (append) ``<log_dir>/gateway_faulthandler.log``, creating the directory."""
|
|
from gateway.run import get_hermes_home
|
|
log_dir = getattr(self.config, "log_dir", None) or os.path.join(str(get_hermes_home()), "logs")
|
|
os.makedirs(log_dir, exist_ok=True)
|
|
return open(os.path.join(log_dir, "gateway_faulthandler.log"), "a", encoding="utf-8")
|
|
|
|
def _start_install_faulthandler(self) -> None:
|
|
"""Enable faulthandler (stderr or a log file) plus the SIGUSR2 stack-dump hook."""
|
|
# sys.stderr may be None (Windows VBS / pythonw / detached service): fall back to a log file.
|
|
try:
|
|
# Enable faulthandler for stack dumps on freezes/crashes (#70344). Falls back to a log file when
|
|
# sys.stderr is None (Windows VBS / pythonw / detached service) — otherwise the gateway would
|
|
# die here and take every adapter offline. See #71671.
|
|
faulthandler.enable()
|
|
except (RuntimeError, ValueError, OSError):
|
|
with _log_suppressed(logging.DEBUG, "faulthandler.enable() unavailable", exc_info=True):
|
|
faulthandler.enable(file=self._open_faulthandler_log(), all_threads=True)
|
|
# SIGUSR2 stack dump to file for service managers that drop stderr; POSIX-only.
|
|
_sigusr2 = getattr(signal, "SIGUSR2", None)
|
|
if _sigusr2 is not None and hasattr(faulthandler, "register"):
|
|
with _log_suppressed(logging.DEBUG, "Could not set up faulthandler file logging", exc_info=True):
|
|
faulthandler.register(
|
|
_sigusr2, file=self._open_faulthandler_log(), all_threads=True, chain=True,
|
|
)
|
|
|
|
def _start_log_startup_environment(self) -> None:
|
|
"""Bind the gateway loop, disarm the startup watchdog, and log the startup environment."""
|
|
from gateway.run import _write_runtime_status_quiet
|
|
try:
|
|
self._gateway_loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
self._gateway_loop = None
|
|
if self._gateway_loop is not None:
|
|
self._start_loop_liveness_guards(self._gateway_loop)
|
|
# Loop live: the loop-liveness watchdog takes over from the startup watchdog. Disarm even
|
|
# when loop guards are config-disabled; only inside this branch (no live loop = stay armed).
|
|
with _log_suppressed(logging.DEBUG, "Startup watchdog disarm failed", exc_info=True):
|
|
from hermes_startup_watchdog import disarm_startup_watchdog
|
|
disarm_startup_watchdog()
|
|
logger.info("Session storage: %s", self.config.sessions_dir)
|
|
self._start_log_systemd_timing_alignment()
|
|
# Log the resolved max_iterations so operators can verify the config.yaml → env bridge.
|
|
with suppress(Exception):
|
|
logger.info(
|
|
"Agent budget: max_iterations=%d (agent.max_turns from config.yaml, "
|
|
"or HERMES_MAX_ITERATIONS from .env, or default 500)",
|
|
int(os.getenv("HERMES_MAX_ITERATIONS", "500")),
|
|
)
|
|
# Warn prominently when redaction is opted out; the redactor snapshots its state at import time,
|
|
# so this line is the source of truth for the process lifetime.
|
|
with suppress(Exception):
|
|
# Redaction status: ON by default (#17691).
|
|
_redact_raw = os.getenv("HERMES_REDACT_SECRETS", "true")
|
|
if _redact_raw.lower() in {"1", "true", "yes", "on"}:
|
|
logger.info(
|
|
"Secret redaction: ENABLED (tool output, logs, and chat "
|
|
"responses are scrubbed before delivery)"
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"Secret redaction: DISABLED (HERMES_REDACT_SECRETS=%s). API keys and tokens may appear "
|
|
"verbatim in chat output, session JSONs, and logs. Set security.redact_secrets: true "
|
|
"in config.yaml to re-enable.", _redact_raw,
|
|
)
|
|
with suppress(Exception):
|
|
from hermes_cli.profiles import get_active_profile_name
|
|
_profile = get_active_profile_name()
|
|
if _profile and _profile != "default":
|
|
logger.info("Active profile: %s", _profile)
|
|
_write_runtime_status_quiet(gateway_state="starting", exit_reason=None, clear_profile_platforms=True)
|
|
with _log_suppressed(logging.DEBUG, "gateway health OTLP export startup failed", exc_info=True):
|
|
from hermes_cli.config import load_config
|
|
from agent.monitoring.gateway_health_export import start_gateway_health_export
|
|
self._gateway_health_export_runtime = start_gateway_health_export(load_config())
|
|
if getattr(self._gateway_health_export_runtime, "enabled", False):
|
|
logger.info("Gateway health OTLP export: enabled")
|
|
# Supply-chain advisories: log only (never block startup or surface to users; only the operator can act).
|
|
with _log_suppressed(logging.DEBUG, "security advisory check failed at gateway startup", exc_info=True):
|
|
from hermes_cli.security_advisories import detect_compromised, gateway_log_message
|
|
_adv_msg = gateway_log_message(detect_compromised())
|
|
if _adv_msg:
|
|
logger.warning("%s", _adv_msg)
|
|
logger.warning("Run `hermes doctor` on the gateway host for full remediation steps.")
|
|
|
|
def _start_log_systemd_timing_alignment(self) -> None:
|
|
"""Warn when systemd's TimeoutStopSec does not cover the drain window (a unit file from before
|
|
an upgrade may encode the old default, so SIGKILL hits mid-drain). Never raises."""
|
|
with _log_suppressed(logging.DEBUG, "check_systemd_timing_alignment failed: %s"):
|
|
from gateway.shutdown_forensics import check_systemd_timing_alignment
|
|
_alignment = check_systemd_timing_alignment(
|
|
self._restart_drain_timeout,
|
|
getattr(self, "_cron_drain_timeout", DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT),
|
|
)
|
|
if _alignment is not None and _alignment.get("mismatch"):
|
|
logger.warning(
|
|
"Stale systemd unit detected: %s has TimeoutStopSec=%.0fs but drain_timeout=%.0fs "
|
|
"cron_drain_timeout=%.0fs (expected >=%.0fs). systemd may SIGKILL the gateway "
|
|
"mid-drain. Run `hermes gateway install --force` to regenerate the unit, or shorten "
|
|
"agent.restart_drain_timeout / agent.cron_drain_timeout.",
|
|
_alignment.get("unit", "(unknown)"), _alignment["timeout_stop_sec"],
|
|
_alignment["drain_timeout"],
|
|
_alignment.get("cron_drain_timeout", DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT),
|
|
_alignment["expected_min"],
|
|
)
|
|
|
|
# Builtin platforms whose ``<P>_ALLOWED_USERS`` / ``<P>_ALLOW_ALL_USERS`` env vars count as an
|
|
# allowlist / open-access opt-in; plugin platforms are appended at check time.
|
|
_ALLOWLIST_ENV_PLATFORMS = (
|
|
"TELEGRAM", "DISCORD", "WHATSAPP", "WHATSAPP_CLOUD", "SLACK", "SIGNAL", "EMAIL", "SMS",
|
|
"MATTERMOST", "MATRIX", "DINGTALK", "FEISHU", "WECOM", "WECOM_CALLBACK", "WEIXIN",
|
|
"BLUEBUBBLES", "QQ", "YUANBAO",
|
|
)
|
|
_BUILTIN_ALLOWED_USERS_VARS = tuple(f"{p}_ALLOWED_USERS" for p in _ALLOWLIST_ENV_PLATFORMS) + (
|
|
"SIGNAL_GROUP_ALLOWED_USERS", "TELEGRAM_GROUP_ALLOWED_USERS",
|
|
"TELEGRAM_GROUP_ALLOWED_CHATS", "GATEWAY_ALLOWED_USERS",
|
|
)
|
|
_BUILTIN_ALLOW_ALL_VARS = tuple(f"{p}_ALLOW_ALL_USERS" for p in _ALLOWLIST_ENV_PLATFORMS)
|
|
|
|
def _start_check_access_policy(self) -> bool:
|
|
"""Warn about missing allowlists; return True when startup must be refused."""
|
|
from gateway.run import (
|
|
_OWN_POLICY_OPEN_ENV, _own_policy_open_startup_violation, _write_runtime_status_quiet
|
|
)
|
|
# Plugin platforms declare their own allowed_users_env / allow_all_env.
|
|
allowed_vars = list(self._BUILTIN_ALLOWED_USERS_VARS)
|
|
allow_all_vars = ["GATEWAY_ALLOW_ALL_USERS", *self._BUILTIN_ALLOW_ALL_VARS]
|
|
with suppress(Exception):
|
|
from gateway.platform_registry import platform_registry
|
|
entries = platform_registry.plugin_entries()
|
|
allowed_vars += [e.allowed_users_env for e in entries if e.allowed_users_env]
|
|
allow_all_vars += [e.allow_all_env for e in entries if e.allow_all_env]
|
|
if not any(os.getenv(v) for v in allowed_vars) and not any(
|
|
os.getenv(v, "").lower() in {"true", "1", "yes"} for v in allow_all_vars
|
|
):
|
|
logger.warning(
|
|
"No env user allowlists configured. Messaging platforms default to pairing/allowlist "
|
|
"policies and will deny unknown senders unless you configure platform allowlists (e.g., "
|
|
"TELEGRAM_ALLOWED_USERS=your_id) or explicitly opt in with GATEWAY_ALLOW_ALL_USERS=true "
|
|
"plus dm_policy/group_policy: open on the platform."
|
|
)
|
|
reason = _own_policy_open_startup_violation(self.config)
|
|
if reason:
|
|
platform_value = reason.split(":", 1)[0]
|
|
allow_all_env = next(
|
|
(env[2] for p, env in _OWN_POLICY_OPEN_ENV.items() if p.value == platform_value), None
|
|
)
|
|
logger.error(
|
|
"Refusing to start: %s has dm_policy/group_policy set to 'open' "
|
|
"but neither GATEWAY_ALLOW_ALL_USERS nor %s is enabled.", platform_value,
|
|
allow_all_env or "a platform allow-all flag",
|
|
)
|
|
_write_runtime_status_quiet(gateway_state="startup_failed", exit_reason=reason)
|
|
self._request_clean_exit(reason)
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def _start_register_plugins_relay_hooks() -> None:
|
|
"""Plugin discovery, relay registration and shell-hook/webhook registration. Never raises."""
|
|
# Discover plugins before shell hooks (plugin block decisions win ties). Explicit: the gateway
|
|
# lazily imports run_agent, so model_tools' discover_plugins() side-effect may not have run.
|
|
with _log_suppressed(logging.WARNING, "plugin discovery failed at gateway startup", exc_info=True):
|
|
from hermes_cli.plugins import discover_plugins
|
|
discover_plugins()
|
|
# Relay entrypoints share the effective profile opt-out, including when a
|
|
# deployment injects a URL. No URL or explicitly disabled -> no side effects.
|
|
try:
|
|
from gateway.relay import (
|
|
register_relay_adapter, relay_url, self_provision_relay, send_relay_policy
|
|
)
|
|
# Relay self-provision sets GATEWAY_RELAY_* in os.environ BEFORE registration reads them.
|
|
self_provision_relay()
|
|
if register_relay_adapter():
|
|
logger.info("relay adapter registered (connector at %s)", relay_url())
|
|
# Declare the relevance policy to the connector so the SAME behavior governs relay delivery.
|
|
send_relay_policy()
|
|
except Exception:
|
|
logger.warning("relay adapter registration failed at gateway startup", exc_info=True)
|
|
GatewayStartupMixin._register_config_hooks("shell-hook registration failed at gateway startup")
|
|
|
|
@staticmethod
|
|
def _register_config_hooks(fail_fmt: str, *fail_args, level: int = logging.DEBUG) -> None:
|
|
"""Register declarative shell hooks + outbound webhooks from the CURRENT scope's config.
|
|
|
|
Gateway has no TTY, so consent must come from --accept-hooks, HERMES_ACCEPT_HOOKS, or
|
|
hooks_auto_accept: true; ``accept_hooks=False`` lets register_from_config resolve env + config.
|
|
Never raises (logged at ``level``).
|
|
"""
|
|
try:
|
|
from hermes_cli.config import load_config
|
|
from agent.shell_hooks import register_from_config
|
|
from agent.outbound_webhooks import register_from_config as register_outbound_webhooks
|
|
_hooks_cfg = load_config()
|
|
register_from_config(_hooks_cfg, accept_hooks=False)
|
|
register_outbound_webhooks(_hooks_cfg)
|
|
except Exception:
|
|
logger.log(level, fail_fmt, *fail_args, exc_info=True)
|
|
|
|
async def _start_recover_previous_run(self) -> None:
|
|
"""Plugins, relay, hooks, then crash/clean-exit recovery of processes and sessions."""
|
|
from gateway.run import _hermes_home
|
|
self._start_register_plugins_relay_hooks()
|
|
self.hooks.discover_and_load()
|
|
# Recover background processes from checkpoint (crash recovery)
|
|
with _log_suppressed(logging.WARNING, "Process checkpoint recovery: %s"):
|
|
from tools.process_registry import process_registry
|
|
recovered = process_registry.recover_from_checkpoint()
|
|
if recovered:
|
|
logger.info("Recovered %s background process(es) from previous run", recovered)
|
|
# Recover sessions active at last exit (exact turn markers + 120s recency fallback for
|
|
# marker-less older turns). SKIP after a clean exit — the previous process already drained.
|
|
_clean_marker = _hermes_home / ".clean_shutdown"
|
|
if _clean_marker.exists():
|
|
logger.info("Previous gateway exited cleanly — skipping session suspension")
|
|
try:
|
|
discarded = await self._consume_clean_shutdown_marker(_clean_marker)
|
|
except Exception as exc:
|
|
logger.error(
|
|
"Clean-start marker cleanup failed; refusing startup so the "
|
|
"clean-exit receipt cannot mask a later unclean exit: %s", exc,
|
|
)
|
|
raise RuntimeError("clean-start recovery cleanup failed") from exc
|
|
if discarded:
|
|
logger.info("Discarded %d orphan active-turn marker(s) after clean shutdown", discarded)
|
|
else:
|
|
exact, fallback = await self._recover_unclean_sessions()
|
|
if exact + fallback:
|
|
logger.info(
|
|
"Marked %d in-flight session(s) as resumable from previous run "
|
|
"(%d exact, %d legacy)", exact + fallback, exact, fallback,
|
|
)
|
|
# Stuck-loop detection: a session active across 3+ consecutive restarts is auto-suspended.
|
|
with _log_suppressed(logging.DEBUG, "Stuck-loop detection failed: %s"):
|
|
# Auto-suspend it so the user gets a clean slate on the next message. See #7536.
|
|
stuck = self._suspend_stuck_loop_sessions()
|
|
if stuck:
|
|
logger.warning("Auto-suspended %d stuck-loop session(s)", stuck)
|
|
|
|
async def _start_prefilter_platforms(self) -> Tuple[bool, int, list, list]:
|
|
"""Create + wire an adapter per enabled platform (no connects). Returns
|
|
(aborted, enabled_platform_count, multiplex_skipped_platforms, pending_connects)."""
|
|
from gateway.run import _platform_has_bot_credential
|
|
enabled_platform_count = 0
|
|
_multiplex_on = self._multiplex_on()
|
|
_multiplex_skipped_platforms: list[Platform] = []
|
|
_pending_connects = [] # (platform, platform_config, adapter); connected concurrently later
|
|
for platform, platform_config in self.config.platforms.items():
|
|
if await self._abort_startup_if_shutdown_requested():
|
|
return True, enabled_platform_count, _multiplex_skipped_platforms, _pending_connects
|
|
if not platform_config.enabled:
|
|
continue
|
|
# Multiplex: a platform enabled in the shared config.yaml may hold its token only in a
|
|
# secondary profile's .env; an empty primary would queue a reconnect loop that never heals.
|
|
# Starting that primary adapter with an empty token fails immediately and queues an infinite
|
|
# reconnect loop that can never heal (#64674). Secondary profiles still start their own adapters
|
|
# under _profile_runtime_scope with the real token -- skip the empty primary instead of failing
|
|
# loudly.
|
|
if _multiplex_on and not _platform_has_bot_credential(platform, platform_config):
|
|
logger.info(
|
|
"Skipping %s on default profile: no bot credential in this "
|
|
"profile's secrets. Secondary multiplexed profiles that "
|
|
"provide the token will still connect.", platform.value,
|
|
)
|
|
_multiplex_skipped_platforms.append(platform)
|
|
continue
|
|
enabled_platform_count += 1
|
|
adapter = self._create_adapter(platform, platform_config)
|
|
if not adapter:
|
|
# Distinguish between missing builtin deps and missing plugin
|
|
if platform.value in {m.value for m in Platform.__members__.values()}:
|
|
logger.warning("No adapter available for %s", platform.value)
|
|
else:
|
|
logger.warning(
|
|
"No adapter for '%s' -- is the plugin installed? "
|
|
"(platform is enabled in config.yaml but no plugin registered it)", platform.value,
|
|
)
|
|
continue
|
|
# Under multiplexing the default profile needs the same whole-handler runtime scope as a
|
|
# secondary (authorization and prompt rendering run before the agent-turn scope).
|
|
self._wire_adapter_handlers(adapter)
|
|
_pending_connects.append((platform, platform_config, adapter))
|
|
return False, enabled_platform_count, _multiplex_skipped_platforms, _pending_connects
|
|
|
|
async def _start_connect_pending(self, _pending_connects: list) -> Optional[list]:
|
|
"""Connect the pre-filtered adapters concurrently. Returns the raw per-platform results, or
|
|
None when a restart/shutdown aborted startup mid-connect (adapters already torn down)."""
|
|
async def _connect_one_startup(p, p_cfg, adp):
|
|
"""Connect a single platform; never let one block the others (#83791)."""
|
|
if await self._abort_startup_if_shutdown_requested(adp, p):
|
|
return (p, adp, p_cfg, "aborted", None)
|
|
logger.info("Connecting to %s...", p.value)
|
|
self._update_platform_runtime_status(
|
|
p.value, platform_state="connecting", error_code=None, error_message=None,
|
|
)
|
|
try:
|
|
ok = await self._connect_initial_adapter_with_timeout(adp, p)
|
|
except Exception as _exc: # noqa: BLE001 - surfaced below as a retryable error
|
|
return (p, adp, p_cfg, "exception", _exc)
|
|
return (p, adp, p_cfg, "ok" if ok else "failed", None)
|
|
|
|
if not _pending_connects:
|
|
return []
|
|
# Abort-aware concurrent wait: a restart/shutdown mid-connect cancels pending connects, tears down
|
|
# completed ones, and aborts startup.
|
|
_task_map = {
|
|
asyncio.ensure_future(_connect_one_startup(p, c, a)): (p, c, a) for (p, c, a) in _pending_connects
|
|
}
|
|
_pending_tasks = set(_task_map)
|
|
while _pending_tasks:
|
|
_done, _pending_tasks = await asyncio.wait(_pending_tasks, timeout=0.05)
|
|
if _pending_tasks and self._startup_should_abort():
|
|
break
|
|
else:
|
|
return [_t.exception() or _t.result() for _t in _task_map]
|
|
# Settle in-flight connects FIRST so a completed adapter's disconnect cannot unblock a sibling.
|
|
for _t in _pending_tasks:
|
|
_t.cancel()
|
|
await asyncio.gather(*_pending_tasks, return_exceptions=True)
|
|
# Then tear down adapters whose connect succeeded — never registered, so stop() won't reach them.
|
|
_connected_ok = [
|
|
_t for _t in _task_map
|
|
if _t not in _pending_tasks and not _t.cancelled() and _t.exception() is None and _t.result()[3] == "ok"
|
|
]
|
|
for _t in [*_pending_tasks, *_connected_ok]:
|
|
await self._startup_teardown_adapter(_task_map[_t][2], _task_map[_t][0])
|
|
await self._abort_startup_if_shutdown_requested()
|
|
return None
|
|
|
|
def _startup_queue_transient_failure(
|
|
self, platform, adapter, platform_config, message: str, startup_retryable_errors: list
|
|
) -> None:
|
|
"""Mark a platform ``retrying`` with ``message`` and queue it in ``_failed_platforms``."""
|
|
self._update_platform_runtime_status(
|
|
platform.value, platform_state="retrying", error_code=None, error_message=message,
|
|
)
|
|
startup_retryable_errors.append(f"{platform.value}: {message}")
|
|
self._failed_platforms[platform] = self._startup_retry_entry(platform, adapter, platform_config)
|
|
|
|
async def _start_aggregate_connect_results(
|
|
self, _raw: list, startup_retryable_errors: list, startup_nonretryable_errors: list
|
|
) -> int:
|
|
"""Apply connect outcomes to shared state single-threaded (exactly as the original serial
|
|
loop did); returns the connected adapter count."""
|
|
connected_count = 0
|
|
for _item in _raw:
|
|
if isinstance(_item, Exception):
|
|
# Unexpected escape from _connect_one_startup (shouldn't happen); log and skip.
|
|
logger.error("Unexpected startup connect error: %s", _item)
|
|
continue
|
|
platform, adapter, platform_config, outcome, exc = _item
|
|
if outcome == "aborted":
|
|
continue
|
|
if outcome == "exception":
|
|
logger.error("\u2717 %s error: %s", platform.value, exc)
|
|
# An adapter that raised mid-connect may still hold a ClientSession/subprocess; treat
|
|
# unexpected exceptions as transient and queue for retry.
|
|
await self._safe_adapter_disconnect(adapter, platform)
|
|
self._startup_queue_transient_failure(
|
|
platform, adapter, platform_config, str(exc), startup_retryable_errors
|
|
)
|
|
continue
|
|
if outcome == "ok":
|
|
self._publish_primary_adapter(platform, adapter)
|
|
connected_count += 1
|
|
# connect() may return True on a degraded (unconfirmed) receive path; don't stamp "connected".
|
|
_degraded = adapter.send_path_degraded
|
|
self._update_platform_runtime_status(
|
|
platform.value, platform_state="retrying" if _degraded else "connected", error_code=None,
|
|
error_message=adapter.DEGRADED_STATUS_MESSAGE if _degraded else None,
|
|
)
|
|
logger.info("\u2713 %s connected%s", platform.value, " (degraded)" if _degraded else "")
|
|
continue
|
|
# outcome == "failed"
|
|
logger.warning("\u2717 %s failed to connect", platform.value)
|
|
# A failed connect() may have allocated ClientSessions / poll tasks / subprocesses.
|
|
await self._safe_adapter_disconnect(adapter, platform)
|
|
if not adapter.has_fatal_error:
|
|
# No fatal error info means likely a transient issue -- queue for retry
|
|
self._startup_queue_transient_failure(
|
|
platform, adapter, platform_config, "failed to connect", startup_retryable_errors
|
|
)
|
|
continue
|
|
# A live foreign token holder is an ownership conflict, not a blip: retryable only for
|
|
# MID-RUN reconnects; at startup route it non-retryable so the gateway exits 78, not deaf.
|
|
_retryable = adapter.fatal_error_retryable and not is_global_startup_conflict(adapter.fatal_error_code)
|
|
self._update_platform_runtime_status(
|
|
platform.value, platform_state="retrying" if _retryable else "fatal",
|
|
error_code=adapter.fatal_error_code, error_message=adapter.fatal_error_message,
|
|
)
|
|
target = startup_retryable_errors if _retryable else startup_nonretryable_errors
|
|
target.append(f"{platform.value}: {adapter.fatal_error_message}")
|
|
if _retryable:
|
|
self._failed_platforms[platform] = self._startup_retry_entry(
|
|
platform, adapter, platform_config, queued=False
|
|
)
|
|
return connected_count
|
|
|
|
def _startup_fail_fatal_config(self, reason: str) -> None:
|
|
"""Record a fatal-config startup failure (exit 78) and request a clean exit."""
|
|
from gateway.run import _write_runtime_status_quiet
|
|
_write_runtime_status_quiet(gateway_state="startup_failed", exit_reason=reason)
|
|
self._exit_code = GATEWAY_FATAL_CONFIG_EXIT_CODE
|
|
self._request_clean_exit(reason)
|
|
self._startup_restore_in_progress = False
|
|
|
|
async def _start_secondary_profiles(
|
|
self, connected_count: int, _multiplex_skipped_platforms: list
|
|
) -> Tuple[bool, int]:
|
|
"""Bring up multiplexed secondary-profile adapters. Returns (aborted, connected_count)."""
|
|
from gateway.run import MultiplexConfigError
|
|
# Secondary-profile adapters connect under their own home + credential scope.
|
|
try:
|
|
connected_count += await self._start_secondary_profile_adapters()
|
|
except MultiplexConfigError as e:
|
|
# Invalid multiplexer config — abort cleanly rather than run a half-wired gateway.
|
|
logger.error("Gateway multiplexer config error: %s", str(e))
|
|
self._startup_fail_fatal_config(str(e))
|
|
return True, connected_count
|
|
except Exception as e:
|
|
logger.error("Secondary-profile adapter startup failed: %s", e, exc_info=True)
|
|
finally:
|
|
# Startup authority is one phase: from here on every adapter retry is non-evicting.
|
|
self._platform_lock_takeover_on_start = False
|
|
# A platform skipped on the primary should have been picked up by a secondary owning the token;
|
|
# if none did it is enabled yet silently unserved — say so loudly.
|
|
# If none did, the platform is enabled in config.yaml yet silently unserved — surface it loudly so
|
|
# the operator sees a config problem instead of a quiet dead channel (#64674 follow-up).
|
|
for _skipped in _multiplex_skipped_platforms:
|
|
if not any(_skipped in _profile_map for _profile_map in self._profile_adapters.values()):
|
|
logger.warning(
|
|
"%s is enabled but no profile (default or secondary) provided a bot credential for it — "
|
|
"the platform is not being served. Add its token to the profile that should "
|
|
"own it, or disable the platform.", _skipped.value,
|
|
)
|
|
return False, connected_count
|
|
|
|
def _start_handle_no_connections(
|
|
self, connected_count: int, enabled_platform_count: int, startup_retryable_errors: list,
|
|
startup_nonretryable_errors: list,
|
|
) -> bool:
|
|
"""Log/degrade when nothing connected; return True when startup must exit."""
|
|
from gateway.run import _write_runtime_status_quiet
|
|
if connected_count != 0:
|
|
return False
|
|
if startup_nonretryable_errors and not startup_retryable_errors:
|
|
reason = "; ".join(startup_nonretryable_errors)
|
|
logger.error("Gateway hit a non-retryable startup conflict: %s", reason)
|
|
self._startup_fail_fatal_config(reason)
|
|
return True
|
|
if startup_nonretryable_errors:
|
|
# Mixed (some fatal, some transient): exiting 78 would take the gateway PERMANENTLY down
|
|
# over a blip. Log the fatal side loudly and fall through to the degraded/retry path.
|
|
logger.error(
|
|
# WhatsApp enabled but never paired) while others hit merely transient errors (e.g. Telegram
|
|
# TimedOut during polling startup). Exiting with GATEWAY_FATAL_CONFIG_EXIT_CODE here is
|
|
# wrong in both supervision worlds: under supervisors that honor the exit-78 contract
|
|
# (systemd RestartPreventExitStatus, s6 finish→125 since #51228) the gateway goes
|
|
# PERMANENTLY down over a network blip; under anything else it crash-loops. Either way the
|
|
# retryable platforms never get their retry. Log the fatal side loudly, then fall through to
|
|
# the degraded/retry path below: the reconnect watcher recovers the retryable platforms; the
|
|
# non-retryable ones remain fatal-parked and visible in runtime status.
|
|
"%d platform(s) fatally misconfigured and parked: %s. "
|
|
"Staying alive so retryable platforms can recover.",
|
|
len(startup_nonretryable_errors), "; ".join(startup_nonretryable_errors),
|
|
)
|
|
if enabled_platform_count <= 0:
|
|
logger.warning("No messaging platforms enabled.")
|
|
logger.info("Gateway will continue running for cron job execution.")
|
|
return False
|
|
if startup_retryable_errors:
|
|
# All retryable: stay alive (cron runs, watcher recovers) rather than systemd restart-loop.
|
|
logger.warning(
|
|
"Gateway started with no connected platforms — %d platform(s) queued for retry: %s",
|
|
len(self._failed_platforms), "; ".join(startup_retryable_errors),
|
|
)
|
|
_write_runtime_status_quiet(gateway_state="degraded", exit_reason=None)
|
|
# No adapter for any enabled platform: fleet nodes share one config.yaml but hold a subset of
|
|
# credentials, so degrade gracefully.
|
|
logger.warning(
|
|
# Fall through to the normal "running" state — reconnect watcher takes it from here. In fleet
|
|
# deployments the same config.yaml is shared across nodes that may only have credentials for a
|
|
# subset of platforms. Rather than failing hard, degrade gracefully and allow cron jobs to run
|
|
# (#5196).
|
|
"No adapter could be created for any of the %d configured platform(s). "
|
|
"Check that required dependencies are installed and credentials are set. "
|
|
"Gateway will continue for cron job execution.", enabled_platform_count,
|
|
)
|
|
return False
|
|
|
|
async def _start_post_connect_services(self, connected_count: int) -> None:
|
|
"""Room worker, heartbeat, gateway:startup hook, channel directory, /update notice."""
|
|
from gateway.run import _hermes_home
|
|
try:
|
|
await self._ensure_hosted_room_worker()
|
|
except Exception:
|
|
logger.error(
|
|
"Group Chat worker failed to start; mutating Group Chat commands "
|
|
"will fail closed until supervision recovers it", exc_info=True,
|
|
)
|
|
self._spawn_supervised(self._hosted_room_worker_watcher, "hosted_room_worker")
|
|
self._start_loop_heartbeat_task()
|
|
from gateway.run_heartbeat_restore import restore_heartbeat_watches
|
|
self._start_heartbeat_poller() # Keep retrying even when the first scan is empty.
|
|
await restore_heartbeat_watches(self)
|
|
hook_count = len(self.hooks.loaded_hooks)
|
|
if hook_count:
|
|
logger.info("%s hook(s) loaded", hook_count)
|
|
await self.hooks.emit("gateway:startup", {"platforms": [p.value for p in self.adapters]})
|
|
if connected_count > 0:
|
|
logger.info("Gateway running with %s platform(s)", connected_count)
|
|
# Initial channel directory for send_message name resolution
|
|
with _log_suppressed(logging.WARNING, "Channel directory build failed: %s"):
|
|
from gateway.channel_directory import build_channel_directory
|
|
directory = await build_channel_directory(self.adapters)
|
|
ch_count = sum(len(chs) for chs in directory.get("platforms", {}).values())
|
|
logger.info("Channel directory built: %d target(s)", ch_count)
|
|
# Restarting after a /update still in progress: keep watching so we notify when it finishes.
|
|
notified = await self._send_update_notification()
|
|
if not notified and any(
|
|
(_hermes_home / name).exists()
|
|
for name in (".update_pending.json", ".update_pending.claimed.json")
|
|
):
|
|
self._schedule_update_notification_watch()
|
|
|
|
async def _start_finish_wiring(self, connected_count: int) -> None:
|
|
"""Post-connect wiring: services, boot notifications, startup restore, recovered watchers."""
|
|
from gateway.run import _planned_restart_notification_pending, _restart_notification_pending
|
|
await self._start_post_connect_services(connected_count)
|
|
# Let fresh adapters settle before lifecycle sends (helps Discord thread deliveries).
|
|
if connected_count > 0:
|
|
await asyncio.sleep(1.0)
|
|
# Before _send_restart_notification() unlinks the marker: did we boot from a chat /restart?
|
|
# One-shot signal for _is_stale_restart_redelivery.
|
|
if _restart_notification_pending():
|
|
self._booted_from_restart = True
|
|
# Boot-path adapter.send() calls must not pin the inbound restore gate (a Telegram flood-
|
|
# control sleep here once froze every platform).
|
|
# Restart notification, home-channel startup notice, and obligation redelivery all call
|
|
# adapter.send(). Bound them the same way _finish_startup_restore bounds resume turns. See #91969.
|
|
await self._await_startup_boot_sends(
|
|
planned_restart_notification_pending=_planned_restart_notification_pending(),
|
|
)
|
|
# Auto-resume restart-interrupted sessions (ledger-answered ones were cleared above); a failed
|
|
# auto-resume stays visible on the next user message.
|
|
self._schedule_resume_pending_sessions()
|
|
await self._finish_startup_restore()
|
|
# Surface state.db init failures to messaging platforms before the user loses data.
|
|
# See #88235.
|
|
await self._send_session_db_warning_notifications()
|
|
# Resume recovered process watchers. Detach the batch atomically (fresh list, not clear(): a
|
|
# concurrent append during the yield must not be lost); yield every 100 to keep the loop live.
|
|
with _log_suppressed(logging.ERROR, "Recovered watcher setup error: %s"):
|
|
from tools.process_registry import process_registry
|
|
watchers = process_registry.pending_watchers
|
|
process_registry.pending_watchers = []
|
|
for i, watcher in enumerate(watchers):
|
|
self._spawn_supervised(
|
|
lambda w=watcher: self._run_process_watcher(w),
|
|
f"process_watcher:{watcher.get('session_id')}", restart=False,
|
|
)
|
|
logger.info("Resumed watcher for recovered process %s", watcher.get("session_id"))
|
|
if i % 100 == 99:
|
|
await asyncio.sleep(0)
|
|
|
|
# Long-lived supervised watchers spawned at the end of start(), in order; supervised name = method
|
|
# name minus the leading underscore.
|
|
_PRE_RECONNECT_WATCHERS = (
|
|
"_session_housekeeping_watcher", "_model_catalog_refresh_watcher", "_session_stall_watcher",
|
|
"_kanban_notifier_watcher", "_kanban_dispatcher_watcher",
|
|
)
|
|
_POST_RECONNECT_WATCHERS = ("_handoff_watcher", "_async_delegation_watcher", "_loop_wakeup_watcher")
|
|
|
|
def _start_spawn_background_watchers(self) -> None:
|
|
"""Spawn the long-lived supervised background watchers."""
|
|
for method in self._PRE_RECONNECT_WATCHERS:
|
|
self._spawn_supervised(getattr(self, method), method[1:])
|
|
if self._failed_platforms:
|
|
logger.info(
|
|
"Starting reconnection watcher for %d failed platform(s): %s",
|
|
len(self._failed_platforms), ", ".join(p.value for p in self._failed_platforms),
|
|
)
|
|
# Supervised: an escaping exception is restarted with backoff instead of stranding queued
|
|
# platforms (the ensure hook only runs on a NEW fatal). ``on_spawn`` keeps the handle current.
|
|
self._spawn_reconnect_watcher()
|
|
for method in self._POST_RECONNECT_WATCHERS:
|
|
self._spawn_supervised(getattr(self, method), method[1:])
|
|
# Scale-to-zero watcher ONLY when opted in, messaging is relay-only/absent, and a wakeUrl exists.
|
|
try:
|
|
if self._scale_to_zero_should_arm():
|
|
logger.info(
|
|
"scale-to-zero: armed (idle timeout %.0fs) — watching for idle",
|
|
self._scale_to_zero_idle_timeout_seconds(),
|
|
)
|
|
self._spawn_supervised(self._scale_to_zero_watcher, "scale_to_zero_watcher")
|
|
else:
|
|
# Say WHY an OPTED-IN instance didn't arm (non-opted stays silent).
|
|
self._log_scale_to_zero_not_armed_reason()
|
|
except Exception: # noqa: BLE001 - arming must never block startup
|
|
logger.debug("scale-to-zero: arm check failed at startup", exc_info=True)
|
|
# Drain-control watcher: reconciles new-turn acceptance with the dashboard's ``.drain_request.json``
|
|
# marker (prior-instantiation markers are ignored via epoch).
|
|
self._spawn_supervised(self._drain_control_watcher, "drain_control_watcher")
|
|
|
|
async def start(self) -> bool:
|
|
"""Start the gateway and all configured platform adapters."""
|
|
logger.info("Starting Hermes Gateway...")
|
|
self._start_install_faulthandler()
|
|
self._start_log_startup_environment()
|
|
if await self._abort_startup_if_shutdown_requested():
|
|
return True
|
|
if self._start_check_access_policy():
|
|
return True
|
|
await self._start_recover_previous_run()
|
|
# The gateway is a boot owner of the Nous free tier, beside `cmd_chat` and `hermes serve`: every
|
|
# demand-time site (provider resolution, /login, the connector token) is a read that needs the
|
|
# identity to already exist. Blocking here, before any adapter connects, is what keeps a fast
|
|
# first DM from arriving with nothing to resolve. With the launch gate unset this is a local
|
|
# inventory and no network.
|
|
await asyncio.get_running_loop().run_in_executor(None, self._start_free_tier_bootstrap)
|
|
# Serialize startup restore against inbound: adapters receive as soon as they connect, so inbound
|
|
# queues until every synthetic resume turn has finished.
|
|
self._startup_restore_in_progress = True
|
|
self._startup_restore_queue = []
|
|
self._startup_restore_tasks = []
|
|
# Fresh boot: the gate opens while the turn machinery is still cold (skeleton prompts). Warm NOW
|
|
# to overlap the connects; _finish_startup_restore awaits it (bounded).
|
|
self._start_startup_warmup()
|
|
startup_nonretryable_errors: list[str] = []
|
|
startup_retryable_errors: list[str] = []
|
|
(
|
|
_aborted, enabled_platform_count, _multiplex_skipped_platforms, _pending_connects
|
|
) = await self._start_prefilter_platforms()
|
|
if _aborted:
|
|
return True
|
|
if await self._abort_startup_if_shutdown_requested():
|
|
return True
|
|
_raw = await self._start_connect_pending(_pending_connects)
|
|
if _raw is None:
|
|
return True
|
|
connected_count = await self._start_aggregate_connect_results(
|
|
_raw, startup_retryable_errors, startup_nonretryable_errors
|
|
)
|
|
if await self._abort_startup_if_shutdown_requested():
|
|
return True
|
|
_aborted, connected_count = await self._start_secondary_profiles(
|
|
connected_count, _multiplex_skipped_platforms
|
|
)
|
|
if _aborted:
|
|
return True
|
|
if self._start_handle_no_connections(
|
|
connected_count, enabled_platform_count, startup_retryable_errors, startup_nonretryable_errors
|
|
):
|
|
return True
|
|
if await self._abort_startup_if_shutdown_requested():
|
|
return True
|
|
self.delivery_router.adapters = self.adapters
|
|
self._wire_teams_pipeline_runtime()
|
|
self._running = True
|
|
self._install_plugin_message_injector()
|
|
self._update_runtime_status("running")
|
|
await self._start_finish_wiring(connected_count)
|
|
self._start_spawn_background_watchers()
|
|
logger.info("Press Ctrl+C to stop")
|
|
return True
|
|
|
|
@dataclasses.dataclass
|
|
class _HandoffDestination:
|
|
"""Resolved destination for one handoff row."""
|
|
platform: Platform
|
|
platform_name: str
|
|
transport: Any
|
|
home: Any
|
|
home_chat_id: str
|
|
effective_thread_id: Optional[str]
|
|
source: SessionSource
|
|
handoff_config: Any
|
|
|
|
def _handoff_resolve_scope(self, profile_name: Optional[str]):
|
|
"""Return (config, adapters) for the profile that queued the handoff. For a secondary
|
|
profile the watcher already entered _profile_runtime_scope, so a fresh load resolves THAT
|
|
profile's config; fail closed — self.config would deliver to the WRONG chat."""
|
|
from gateway.run import load_gateway_config
|
|
if not profile_name or profile_name == "default":
|
|
return self.config, self.adapters
|
|
secondary = (self._profile_adapters or {}).get(profile_name)
|
|
if not secondary:
|
|
raise RuntimeError(f"profile '{profile_name}' has no live adapters in this gateway")
|
|
try:
|
|
return load_gateway_config(), secondary
|
|
except Exception as exc:
|
|
logger.error(
|
|
"Handoff: could not load config for profile %s; failing the handoff instead of "
|
|
"delivering via the primary's config", profile_name, exc_info=True,
|
|
)
|
|
raise RuntimeError(f"could not load config for profile '{profile_name}': {exc}") from exc
|
|
|
|
async def _handoff_resolve_destination(
|
|
self, row: Dict[str, Any], profile_name: Optional[str]
|
|
) -> "GatewayStartupMixin._HandoffDestination":
|
|
"""Resolve platform, transport, home channel, thread and destination source for a row."""
|
|
from gateway.delivery import resolve_delivery_transport
|
|
cli_session_id = row["id"]
|
|
platform_name = (row.get("handoff_platform") or "").strip().lower()
|
|
if not platform_name:
|
|
raise RuntimeError("handoff_platform is empty")
|
|
try:
|
|
platform = Platform(platform_name)
|
|
except (ValueError, KeyError):
|
|
raise RuntimeError(f"unknown platform '{platform_name}'")
|
|
handoff_config, handoff_adapters = self._handoff_resolve_scope(profile_name)
|
|
# Alias-aware transport: a relay-fronted gateway registers ONE Platform.RELAY adapter fronting
|
|
# N logical platforms, so a literal adapters.get() would miss a deliverable one.
|
|
transport = resolve_delivery_transport(platform, handoff_config, handoff_adapters)
|
|
if not transport:
|
|
raise RuntimeError(f"platform '{platform_name}' is not active in this gateway")
|
|
home = handoff_config.get_home_channel(platform)
|
|
if not home or not home.chat_id:
|
|
raise RuntimeError(
|
|
f"no home channel configured for {platform_name}; run /sethome on the desired chat first"
|
|
)
|
|
home_chat_id = str(home.chat_id)
|
|
# Fresh thread for the handoff's own scrollback; None when unsupported or creation failed.
|
|
cli_title = row.get("title") or cli_session_id[:8]
|
|
try:
|
|
new_thread_id = await transport.adapter.create_handoff_thread(
|
|
home_chat_id, f"Hermes — {cli_title}",
|
|
)
|
|
except Exception as exc:
|
|
logger.debug("Handoff: create_handoff_thread raised on %s: %s", platform_name, exc, exc_info=True)
|
|
new_thread_id = None
|
|
effective_thread_id = new_thread_id or (str(home.thread_id) if home.thread_id else None)
|
|
# Telegram private-chat DM topics use the DM-topic source shape (user_id == chat_id) so the
|
|
# synthetic turn binds the same key later inbound turns arrive on (`dm`, not `thread`).
|
|
is_telegram_private_chat = (
|
|
platform == Platform.TELEGRAM and looks_like_telegram_private_chat_id(home_chat_id)
|
|
)
|
|
is_thread = bool(new_thread_id) and not is_telegram_private_chat
|
|
# Discord builds in-thread messages with ``chat_id == thread id``: key on the thread's OWN id.
|
|
dest_source = SessionSource(
|
|
platform=platform,
|
|
chat_id=str(effective_thread_id) if (
|
|
is_thread and platform == Platform.DISCORD and effective_thread_id
|
|
) else home_chat_id,
|
|
chat_name=home.name,
|
|
chat_type="thread" if is_thread else "dm",
|
|
user_id=home_chat_id if is_telegram_private_chat else "system:handoff",
|
|
user_name="Handoff", thread_id=effective_thread_id, profile=profile_name,
|
|
)
|
|
return self._HandoffDestination(
|
|
platform=platform, platform_name=platform_name, transport=transport, home=home,
|
|
home_chat_id=home_chat_id, effective_thread_id=effective_thread_id, source=dest_source,
|
|
handoff_config=handoff_config,
|
|
)
|
|
|
|
def _handoff_session_key(self, dest, profile_name: Optional[str]) -> str:
|
|
"""Destination session_key by the adapters' own rules. Thread keys omit user_id so the next
|
|
message shares it. Namespaced to the queuing profile (else a multiplexed gateway builds
|
|
``agent:main:...`` while the profile's adapter routes on ``agent:<profile>:...``); the store
|
|
resolver is only the root fallback. The isinstance check is load-bearing: a Mock store returns
|
|
a truthy MagicMock."""
|
|
platform_cfg = dest.handoff_config.platforms.get(dest.platform)
|
|
extra = platform_cfg.extra if platform_cfg else {}
|
|
handoff_profile = profile_name if (profile_name and profile_name != "default") else None
|
|
if handoff_profile is None:
|
|
try:
|
|
store = getattr(self.async_session_store, "_store", self.async_session_store)
|
|
resolver = getattr(store, "_resolve_profile_for_key", None)
|
|
# Resolve the bound text channel's channel_prompt so voice input gets the same per-channel
|
|
# context as typed messages (#50149).
|
|
if callable(resolver):
|
|
resolved = resolver(dest.source)
|
|
if isinstance(resolved, str) and resolved.strip():
|
|
handoff_profile = resolved
|
|
except Exception:
|
|
logger.debug("Handoff: could not resolve profile namespace", exc_info=True)
|
|
return build_session_key(
|
|
dest.source, group_sessions_per_user=extra.get("group_sessions_per_user", True),
|
|
thread_sessions_per_user=extra.get("thread_sessions_per_user", False), profile=handoff_profile,
|
|
)
|
|
|
|
async def _process_handoff(self, row: Dict[str, Any], profile_name: Optional[str] = None) -> None:
|
|
"""Execute one handoff row; raises on failure (caller marks failed). ``profile_name`` (None =
|
|
root) is the profile whose store queued it — load-bearing under multiplex: secondaries live in
|
|
``_profile_adapters`` and the key must be namespaced ``agent:<profile>:...`` or nobody reads it."""
|
|
cli_session_id = row["id"]
|
|
dest = await self._handoff_resolve_destination(row, profile_name)
|
|
session_key = self._handoff_session_key(dest, profile_name)
|
|
# Ensure a session_store entry exists for this key; switch_session then re-points it.
|
|
await self.async_session_store.get_or_create_session(dest.source)
|
|
# switch_session ends the prior session and reopens the CLI session under the new key.
|
|
switched = await self.async_session_store.switch_session(session_key, cli_session_id)
|
|
if switched is None:
|
|
raise RuntimeError(f"could not switch session key {session_key} → {cli_session_id}")
|
|
# Evict the cached AIAgent (rebuild against the CLI session_id, like /resume) and clear stale
|
|
# running-agent state so the synthetic turn isn't queued behind it.
|
|
self._evict_cached_agent(session_key)
|
|
self._release_running_agent_state(session_key)
|
|
cli_title = row.get("title") or cli_session_id[:8]
|
|
synthetic_event = MessageEvent(
|
|
text=(
|
|
f"[Session was just handed off from CLI (\"{cli_title}\") to this "
|
|
f"channel. The full prior conversation history is loaded above. "
|
|
f"Briefly confirm you're working here and summarize what we were "
|
|
f"working on, so the user can continue from this device.]"
|
|
),
|
|
source=dest.source,
|
|
internal=True,
|
|
)
|
|
logger.info(
|
|
"Handoff: dispatching synthetic turn for CLI session %s → %s "
|
|
"(home=%s, thread=%s, session_key=%s)",
|
|
cli_session_id, dest.platform_name, dest.home.chat_id, dest.effective_thread_id, session_key,
|
|
)
|
|
# Inline _handle_message keeps success/failure observable (handle_message would detach it).
|
|
response_text = await self._handle_message(synthetic_event)
|
|
if not response_text:
|
|
# Streaming may have delivered inline; the agent ran without raising — success.
|
|
return
|
|
# Reply into the new thread (else the home channel) via the resolved transport, so a relay-fronted
|
|
# logical platform is stamped on the outbound frame.
|
|
send_metadata = {"thread_id": dest.effective_thread_id} if dest.effective_thread_id else None
|
|
try:
|
|
result = await dest.transport.send(
|
|
dest.platform, str(dest.home.chat_id), response_text, send_metadata,
|
|
)
|
|
except Exception as exc:
|
|
raise RuntimeError(f"adapter.send failed: {exc}") from exc
|
|
if not getattr(result, "success", True):
|
|
raise RuntimeError(f"adapter.send failed: {_send_error(result)}")
|