feat(gateway): multiplexer hot-serves profiles created while it runs, unroutes deleted ones

A `gateway.multiplex_profiles` gateway enumerated `profiles/` once at boot, so a profile
created afterwards (CLI, dashboard, Desktop, TUI) was never served until `hermes gateway
restart`; Desktop and the dashboard gave no reminder, so a new profile's bot simply never
connected.

The served set is now reconciled at runtime (`gateway/run_profile_reconcile.py`):
- `hermes_cli/profiles.py` create/delete ping the multiplexer over its control socket
  (new `rescan-profiles` verb); a supervised watcher rescans every 30s as the safety net.
- A new profile gets its adapters under its own runtime scope from its config/.env
  (`_start_one_profile_adapters`, same duplicate-credential guard as boot, now seeded
  with the LIVE secondaries' claims), `served_profiles` in gateway_state.json is
  updated, MCP discovery + log routing run for it. Other profiles' adapters are never
  touched.
- A served profile whose config.yaml/.env changed is re-scanned so a token added after
  create builds the adapter; already-live/queued platforms are skipped (no second poller).
- A deleted profile (tombstone) has its reconnects cancelled, adapters torn down,
  pairing/busy bookkeeping and cached agents dropped, and this process's SQLite /
  memory-store handles released so the deleter's rmtree succeeds.
- The in-process cron ticker takes a live enumerator so new profiles' jobs fire.
- PUT /api/messaging/platforms/<id>?profile=X returns `hot_served` when a live
  multiplexer rebuilt X's adapters; Desktop/dashboard skip the restart banner then.
- `hermes profile create` confirms hot-serve; the restart reminder stays for a gateway
  that did not pick the profile up (older build / signal failed).
This commit is contained in:
teknium1
2026-09-12 07:55:28 -07:00
committed by Teknium
parent 0d6b59fb74
commit d1dbb0ac9e
25 changed files with 585 additions and 37 deletions
+10 -2
View File
@@ -22,12 +22,20 @@ export function getMessagingPlatforms(profile?: null | string): Promise<Messagin
})
}
/** `hot_served`: a live multiplexer serving this named profile rebuilt its adapters from the new
* credentials right away — no gateway restart is needed for the change to take effect. */
export interface MessagingPlatformUpdateResponse {
hot_served?: boolean
ok: boolean
platform: string
}
export function updateMessagingPlatform(
platformId: string,
body: MessagingPlatformUpdate,
profile?: null | string
): Promise<{ ok: boolean; platform: string }> {
return hermesApi<{ ok: boolean; platform: string }>({
): Promise<MessagingPlatformUpdateResponse> {
return hermesApi<MessagingPlatformUpdateResponse>({
...profileScoped(profile),
path: `/api/messaging/platforms/${encodeURIComponent(platformId)}`,
method: 'PUT',
+19 -8
View File
@@ -166,6 +166,17 @@ export function MessagingView({ setStatusbarItemGroup: _setStatusbarItemGroup, .
}
}, [])
// A multiplexed named profile is re-served from its new config at once (`hot_served`): no restart
// banner; re-read status once the adapter had a moment to connect. Anything else needs a restart.
const settleAfterUpdate = useCallback((hotServed: boolean | undefined) => {
if (hotServed) {
window.setTimeout(() => void refreshPlatformsRef.current(true), 4000)
return
}
setRestartNeeded(true)
}, [])
const refreshPlatforms = useCallback(
async (silent = false) => {
if (!silent) {
@@ -320,7 +331,7 @@ export function MessagingView({ setStatusbarItemGroup: _setStatusbarItemGroup, .
setSaving(`enabled:${platform.id}`)
try {
await updateMessagingPlatform(platform.id, { enabled }, scopeProfile)
const result = await updateMessagingPlatform(platform.id, { enabled }, scopeProfile)
setPlatforms(
current =>
current?.map(row =>
@@ -333,11 +344,11 @@ export function MessagingView({ setStatusbarItemGroup: _setStatusbarItemGroup, .
: row
) ?? current
)
setRestartNeeded(true)
settleAfterUpdate(result.hot_served)
notify({
kind: 'success',
title: enabled ? m.platformEnabled(platform.name) : m.platformDisabled(platform.name),
message: m.restartToApply
message: result.hot_served ? m.appliedLive : m.restartToApply
})
} catch (err) {
notifyError(err, m.failedUpdate(platform.name))
@@ -356,14 +367,14 @@ export function MessagingView({ setStatusbarItemGroup: _setStatusbarItemGroup, .
setSaving(`env:${platform.id}`)
try {
await updateMessagingPlatform(platform.id, { env }, scopeProfile)
const result = await updateMessagingPlatform(platform.id, { env }, scopeProfile)
setEdits(current => ({ ...current, [platform.id]: {} }))
await refreshPlatforms()
setRestartNeeded(true)
settleAfterUpdate(result.hot_served)
notify({
kind: 'success',
title: m.setupSaved(platform.name),
message: m.restartToReconnect
message: result.hot_served ? m.connectingLive : m.restartToReconnect
})
} catch (err) {
notifyError(err, m.failedSave(platform.name))
@@ -376,7 +387,7 @@ export function MessagingView({ setStatusbarItemGroup: _setStatusbarItemGroup, .
setSaving(`clear:${key}`)
try {
await updateMessagingPlatform(platform.id, { clear_env: [key] }, scopeProfile)
const result = await updateMessagingPlatform(platform.id, { clear_env: [key] }, scopeProfile)
setEdits(current => ({
...current,
[platform.id]: {
@@ -385,7 +396,7 @@ export function MessagingView({ setStatusbarItemGroup: _setStatusbarItemGroup, .
}
}))
await refreshPlatforms()
setRestartNeeded(true)
settleAfterUpdate(result.hot_served)
notify({ kind: 'success', title: m.keyCleared(key), message: m.setupUpdated(platform.name) })
} catch (err) {
notifyError(err, m.failedClear(key))
+2
View File
@@ -1422,6 +1422,8 @@ export const ar = defineLocale({
restartToApply: 'أعد التشغيل لتطبيق التغييرات.',
setupSaved: name => `تم حفظ إعداد ${name}`,
restartToReconnect: 'أعد التشغيل لإعادة الاتصال.',
appliedLive: 'تم التطبيق على البوابة قيد التشغيل.',
connectingLive: 'البوابة قيد التشغيل تتصل باستخدام بيانات الاعتماد الجديدة.',
keyCleared: key => `تم مسح ${key}`,
setupUpdated: name => `تم تحديث إعداد ${name}`,
failedUpdate: name => `فشل تحديث ${name}`,
+2
View File
@@ -2037,6 +2037,8 @@ export const en: Translations = {
restartToApply: 'This change takes effect after a gateway restart.',
setupSaved: name => `${name} setup saved`,
restartToReconnect: 'New credentials take effect after a gateway restart.',
appliedLive: 'Applied to the running gateway.',
connectingLive: 'The running gateway is connecting with the new credentials.',
keyCleared: key => `${key} cleared`,
setupUpdated: name => `${name} setup was updated.`,
failedUpdate: name => `Failed to update ${name}`,
+2
View File
@@ -1744,6 +1744,8 @@ export const ja = defineLocale({
restartToApply: 'この変更はゲートウェイの再起動後に有効になります。',
setupSaved: name => `${name} の設定を保存しました`,
restartToReconnect: '新しい認証情報はゲートウェイの再起動後に有効になります。',
appliedLive: '実行中のゲートウェイに適用されました。',
connectingLive: '実行中のゲートウェイが新しい認証情報で接続しています。',
keyCleared: key => `${key} をクリアしました`,
setupUpdated: name => `${name} の設定が更新されました。`,
failedUpdate: name => `${name} の更新に失敗しました`,
+2
View File
@@ -1926,6 +1926,8 @@ export const ru = defineLocale({
restartToApply: 'Это изменение вступит в силу после перезапуска шлюза.',
setupSaved: name => `Настройка ${name} сохранена`,
restartToReconnect: 'Новые учётные данные вступят в силу после перезапуска шлюза.',
appliedLive: 'Применено к работающему шлюзу.',
connectingLive: 'Работающий шлюз подключается с новыми учётными данными.',
keyCleared: key => `${key} очищено`,
setupUpdated: name => `Настройка ${name} обновлена.`,
failedUpdate: name => `Не удалось обновить ${name}`,
+2
View File
@@ -1803,6 +1803,8 @@ export interface Translations {
restartToApply: string
setupSaved: (name: string) => string
restartToReconnect: string
appliedLive: string
connectingLive: string
keyCleared: (key: string) => string
setupUpdated: (name: string) => string
failedUpdate: (name: string) => string
+2
View File
@@ -1683,6 +1683,8 @@ export const zhHant = defineLocale({
restartToApply: '此變更將在閘道重新啟動後生效。',
setupSaved: name => `${name} 設定已儲存`,
restartToReconnect: '新憑證將在閘道重新啟動後生效。',
appliedLive: '已套用到執行中的閘道。',
connectingLive: '執行中的閘道正在使用新憑證連線。',
keyCleared: key => `${key} 已清除`,
setupUpdated: name => `${name} 設定已更新。`,
failedUpdate: name => `更新 ${name} 失敗`,
+2
View File
@@ -2203,6 +2203,8 @@ export const zh = defineLocale({
restartToApply: '此更改将在网关重启后生效。',
setupSaved: name => `${name} 设置已保存`,
restartToReconnect: '新凭据将在网关重启后生效。',
appliedLive: '已应用到正在运行的网关。',
connectingLive: '正在运行的网关正在使用新凭据连接。',
keyCleared: key => `${key} 已清除`,
setupUpdated: name => `${name} 设置已更新。`,
failedUpdate: name => `更新 ${name} 失败`,
+15 -5
View File
@@ -66,6 +66,14 @@ def _existing_profile_homes(profile_homes: list) -> list:
profile's home untouched, which is the correct invariant: a home that does not exist cannot hold jobs to
fire.
"""
if callable(profile_homes):
# Live enumerator (multiplex gateway): a profile created after startup is ticked without a
# restart; a raising enumerator keeps this cycle at zero homes rather than killing the ticker.
try:
profile_homes = list(profile_homes())
except Exception:
logger.warning("cron profile enumeration failed; skipping this cycle", exc_info=True)
return []
return [entry for entry in profile_homes if Path(_profile_entry(entry)[1]).is_dir()]
@@ -393,7 +401,7 @@ class InProcessCronScheduler(CronScheduler):
# jobs actually fire instead of languishing in a store no ticker owns (#69377). Without this, only
# the process-global HERMES_HOME (the default profile) is ticked. Heartbeats and recovery are also
# scoped per profile so `hermes cron status` reflects liveness for every profile independently.
if profile_homes:
if profile_homes is not None and (callable(profile_homes) or profile_homes):
self._start_multiplex(
stop_event, profile_homes=profile_homes, adapters=adapters, loop=loop,
interval=interval, can_dispatch=can_dispatch, profile_adapters=profile_adapters,
@@ -462,10 +470,12 @@ class InProcessCronScheduler(CronScheduler):
)
from cron.jobs import clear_ticker_error, record_ticker_error, record_ticker_heartbeat
initial_homes = _existing_profile_homes(profile_homes)
logger.info(
"Multiplex cron scheduler started for %d profile(s): %s",
len(profile_homes),
[p[0] if isinstance(p, tuple) else p for p in profile_homes],
"Multiplex cron scheduler started for %d profile(s): %s%s",
len(initial_homes),
[p[0] if isinstance(p, tuple) else p for p in initial_homes],
" (re-enumerated every cycle)" if callable(profile_homes) else "",
)
def tick_adapters_for(profile_name):
@@ -482,7 +492,7 @@ class InProcessCronScheduler(CronScheduler):
# Recovery + heartbeat per profile; one broken store must not abort startup for the others.
# A profile may have been deleted since this snapshot was taken; never recreate a deleted home's
# cron workspace via the heartbeat below (#47368).
for entry in _existing_profile_homes(profile_homes):
for entry in initial_homes:
_, home = _profile_entry(entry)
try:
with _profile_cron_scope(home):
+8
View File
@@ -340,3 +340,11 @@ def pause_gateway_for_update(home: Path, *, timeout: float = _DEFAULT_CLIENT_TIM
Step 2 of the socket migration (#92091).
"""
return query_gateway_control(home, "pause-for-update", timeout=timeout)
def rescan_gateway_profiles(home: Path, *, timeout: float = 8.0) -> Optional[dict[str, Any]]:
"""Ask the multiplexer serving ``home`` to reconcile ``profiles/`` now (hot-serve a created profile,
unroute a deleted one). Returns its ``{"served_profiles", "added", "removed", ...}`` answer, or None
when no gateway answers / the gateway predates the verb — callers then rely on the periodic rescan
(or the restart reminder)."""
return query_gateway_control(home, "rescan-profiles", timeout=timeout)
+22 -3
View File
@@ -2149,6 +2149,7 @@ from gateway.run_notifications import GatewayNotificationsMixin
from gateway.run_inbound import GatewayInboundMixin
from gateway.run_goals import GatewayGoalsMixin
from gateway.run_agent_cache import GatewayAgentCacheMixin
from gateway.run_profile_reconcile import GatewayProfileReconcileMixin
from gateway.platforms.base import (
BasePlatformAdapter,
_reply_anchor_for_event,
@@ -3329,7 +3330,7 @@ class GatewayRunner(
GatewayVoiceMixin, GatewayAdapterLifecycleMixin, GatewayTopicThreadsMixin, GatewayTurnMixin,
GatewayShutdownMixin, GatewayBusySessionMixin, GatewayConfigLoadersMixin, GatewayStartupMixin,
GatewaySessionWatchersMixin, GatewayNotificationsMixin, GatewayInboundMixin, GatewayGoalsMixin,
GatewayAgentCacheMixin):
GatewayAgentCacheMixin, GatewayProfileReconcileMixin):
"""Main gateway controller: manages adapter lifecycles, routes messages to/from the agent."""
# Class-level defaults so partial construction in tests doesn't blow up on attribute access.
@@ -5115,8 +5116,24 @@ async def _start_gateway_start_control_socket(runner):
"pausing": accepted, "already_stopping": not accepted,
"pid": os.getpid(), "drain_timeout": _drain}
def _rescan_profiles_handler() -> dict:
"""``hermes profile create/delete`` asks the multiplexer to reconcile ``profiles/`` now
(the watcher also rescans periodically). Runs on the socket executor: marshal onto the loop
and wait briefly so the caller learns whether the profile is served."""
if not getattr(runner.config, "multiplex_profiles", False):
return {"multiplex": False, "served_profiles": runner.served_profile_names()}
future = asyncio.run_coroutine_threadsafe(
runner.reconcile_served_profiles(reason="control-socket"), _main_loop)
try:
# Bounded: a token-less create reconciles in milliseconds; a credential-add whose adapter
# connect outlasts this keeps running and the caller sees ``pending`` (not an error).
return {"multiplex": True, **future.result(timeout=5.0)}
except concurrent.futures.TimeoutError:
return {"multiplex": True, "pending": True, "served_profiles": runner.served_profile_names()}
_control_server = GatewayControlServer(
verb_handlers={"pause-for-update": _pause_for_update_handler})
verb_handlers={"pause-for-update": _pause_for_update_handler,
"rescan-profiles": _rescan_profiles_handler})
if not await _control_server.start():
_control_server = None
else:
@@ -5145,7 +5162,9 @@ def _start_gateway_start_cron_and_housekeeping(runner):
try:
profile_homes = _cron_tick_profile_homes(runner.config)
if profile_homes:
cron_start_kwargs["profile_homes"] = profile_homes
# Live enumerator: the ticker re-reads profiles/ every cycle so a profile created while
# the multiplexer runs gets its jobs fired without a restart (hot-serve).
cron_start_kwargs["profile_homes"] = lambda: _cron_tick_profile_homes(runner.config)
# Per-profile adapters so each profile's cron output goes via its own bot, not the default's.
cron_start_kwargs["profile_adapters"] = getattr(runner, "_profile_adapters", None)
# runner.adapters belongs to the LAUNCH profile (``default``, or the ``--profile``
+6
View File
@@ -873,6 +873,7 @@ class GatewayAdapterLifecycleMixin:
from gateway.status import write_runtime_status
from gateway.pairing import PairingStore
served = [active] + sorted(name for name, _home in profile_homes if name != active)
self._note_served_profiles(profile_homes)
for name in served:
if name and name not in self.pairing_stores:
self.pairing_stores[name] = (
@@ -956,6 +957,11 @@ class GatewayAdapterLifecycleMixin:
for platform, platform_config in profile_cfg.platforms.items():
if not platform_config.enabled:
continue
# Runtime re-scan of a served profile (config/.env changed): only platforms that are not
# already live or queued for reconnect are built — never a second poller on the same bot.
if platform in profile_map or platform in (
(getattr(self, "_profile_failed_platforms", None) or {}).get(profile_name) or {}):
continue
# 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(
+203
View File
@@ -0,0 +1,203 @@
"""Hot-serve for ``gateway.multiplex_profiles``: keep the served-profile set in step with ``profiles/``
while the multiplexer runs, instead of snapshotting it once at boot.
Three things were start-time snapshots: the secondary adapter set (``_start_secondary_profile_adapters``),
the ``served_profiles`` record in ``gateway_state.json`` (``_record_served_profiles``) and the cron
ticker's ``profile_homes`` list. Everything else (``/p/<profile>/`` prefixes, profile-route eligibility,
handoff/kanban watchers, shared ingress) already reads ``profiles_to_serve()`` / ``_profile_adapters``
live, so reconciling those three is enough for a profile created after boot to be served.
``reconcile_served_profiles`` runs on the loop under one lock, triggered by the ``rescan-profiles`` control
verb (``hermes_cli/profiles.py`` create/delete fire it through the control socket) and by the supervised
``_profile_reconcile_watcher`` every ``_PROFILE_RESCAN_INTERVAL_SECS`` as the safety net. A served profile whose ``config.yaml``/``.env``
changed since its adapters were last built is re-scanned too: creators make the profile first and add the
bot token afterwards, and without this an adapter-less profile would stay adapter-less forever.
"""
from __future__ import annotations
import asyncio
import logging
import os
from pathlib import Path
from typing import Any, Dict, Optional
from gateway.run_shutdown import _log_suppressed
logger = logging.getLogger(__name__)
_PROFILE_RESCAN_INTERVAL_SECS = 30.0
_PROFILE_SIGNATURE_FILES = ("config.yaml", ".env")
def profile_serve_signature(home: "Path") -> tuple:
"""Cheap change detector for a served profile's credentials/config: (mtime_ns, size) per file."""
sig = []
for name in _PROFILE_SIGNATURE_FILES:
try:
st = os.stat(Path(home) / name)
sig.append((st.st_mtime_ns, st.st_size))
except OSError:
sig.append(None)
return tuple(sig)
class GatewayProfileReconcileMixin:
"""Runtime reconciliation of the multiplexed served-profile set (hot add / unroute / credential-add)."""
_served_profile_homes: Optional[Dict[str, "Path"]] = None
_served_profile_signatures: Optional[Dict[str, tuple]] = None
_profile_reconcile_lock: Optional[asyncio.Lock] = None
# ── state helpers ─────────────────────────────────────────────────────────────────────────────
def _reconcile_lock(self) -> asyncio.Lock:
if self._profile_reconcile_lock is None:
self._profile_reconcile_lock = asyncio.Lock()
return self._profile_reconcile_lock
def served_profile_names(self) -> list:
"""Profiles this multiplexer currently serves (active first), from the live bookkeeping."""
homes = self._served_profile_homes or {}
active = getattr(self, "_primary_profile_name", None) or "default"
return ([active] if active in homes or not homes else []) + sorted(n for n in homes if n != active)
def _note_served_profiles(self, profile_homes) -> None:
"""Called by ``_record_served_profiles``: remember the served set and each home's signature."""
homes = {str(name): Path(home) for name, home in profile_homes}
self._served_profile_homes = homes
sigs = self._served_profile_signatures if isinstance(self._served_profile_signatures, dict) else {}
self._served_profile_signatures = {name: sigs.get(name) or profile_serve_signature(home)
for name, home in homes.items()}
# ── watcher ───────────────────────────────────────────────────────────────────────────────────
async def _profile_reconcile_watcher(self, interval: float = _PROFILE_RESCAN_INTERVAL_SECS) -> None:
"""Supervised safety net: rescan ``profiles/`` every ``interval`` seconds (creators signal the
control socket for an immediate rescan). Returns at once, never respawned, when multiplexing is off."""
if not self._multiplex_on():
return
while self._running:
await asyncio.sleep(interval)
if not self._running:
return
try:
await self.reconcile_served_profiles(reason="watcher")
except asyncio.CancelledError:
raise
except Exception:
logger.warning("Served-profile reconcile failed; retrying next cycle", exc_info=True)
# ── reconcile ─────────────────────────────────────────────────────────────────────────────────
async def reconcile_served_profiles(self, *, reason: str = "request") -> Dict[str, Any]:
"""Diff ``profiles/`` against the served set: start adapters for new profiles, tear down and
unroute deleted ones, (re)build adapters for served profiles whose config/.env changed. Other
profiles' adapters are never touched. Returns ``{"added", "removed", "rescanned", "served_profiles"}``."""
from gateway.run import MultiplexConfigError, _multiplex_profile_homes
result: Dict[str, Any] = {"added": [], "removed": [], "rescanned": [], "reason": reason}
if not self._multiplex_on():
return {**result, "multiplex": False, "served_profiles": self.served_profile_names()}
if not self._running or self._served_profile_homes is None:
# Startup enumerates profiles/ itself; a rescan before it finishes has nothing to diff against.
return {**result, "pending": True, "served_profiles": self.served_profile_names()}
async with self._reconcile_lock():
active = getattr(self, "_primary_profile_name", None) or "default"
current = {str(name): Path(home) for name, home in _multiplex_profile_homes(self.config)}
known = dict(self._served_profile_homes or {})
sigs = self._served_profile_signatures or {}
added = [n for n in current if n not in known and n != active]
removed = [n for n in known if n not in current and n != active]
changed = [n for n in current if n in known and n != active and n not in added
and profile_serve_signature(current[n]) != sigs.get(n)]
if not (added or removed or changed):
return {**result, "served_profiles": self.served_profile_names()}
for name in removed:
await self._unserve_profile(name, known[name])
result["removed"].append(name)
claimed = self._live_resource_claims(active)
for name in added + changed:
try:
connected = await self._start_one_profile_adapters(name, current[name], claimed)
except MultiplexConfigError as exc:
# Boot refuses to run with such a profile; at runtime we park just this profile.
logger.error("[MULTIPLEX] Profile '%s' not served: %s", name, exc)
connected = 0
except Exception:
logger.error("[MULTIPLEX] Failed to start adapters for profile '%s'", name, exc_info=True)
connected = 0
sigs[name] = profile_serve_signature(current[name])
if name in added:
logger.info("[MULTIPLEX] Now serving profile '%s' (%s adapter(s) connected; %s)", name, connected, reason)
result["added"].append(name)
else:
logger.info("[MULTIPLEX] Re-scanned profile '%s' after config/.env change (%s adapter(s) connected)", name, connected)
result["rescanned"].append(name)
self._served_profile_signatures = sigs
self._record_served_profiles(active, list(current.items()))
if added:
await self._after_profiles_added([(n, current[n]) for n in added])
result["served_profiles"] = self.served_profile_names()
return result
def _live_resource_claims(self, active: str) -> Dict[tuple, str]:
"""Startup's ``claimed`` map rebuilt from what is live now: primary claims plus every connected
secondary's credential/listener, so a hot-added profile reusing a token is parked, never a
second poller."""
claimed = self._primary_resource_claims(active)
for profile_name, adapters in (getattr(self, "_profile_adapters", None) or {}).items():
for platform, adapter in list(adapters.items()):
for claim in (self._adapter_credential_claim(platform, adapter),
self._adapter_listener_claim(platform, adapter)):
if claim is not None:
claimed[claim] = profile_name
return claimed
async def _after_profiles_added(self, profile_homes) -> None:
"""Per-profile startup side effects for hot-added profiles: log routing + scoped MCP discovery."""
from gateway.run import _enable_multiplex_log_routing, _profile_runtime_scope
from contextvars import copy_context
with _log_suppressed(logging.DEBUG, "log routing refresh failed", exc_info=True):
_enable_multiplex_log_routing(self.config)
loop = asyncio.get_running_loop()
for profile_name, profile_home in profile_homes:
try:
from tools.mcp_tool_discovery import discover_mcp_tools
with _profile_runtime_scope(Path(profile_home)):
await loop.run_in_executor(None, copy_context().run, discover_mcp_tools)
except Exception:
logger.warning("MCP tool discovery failed for profile '%s'", profile_name, exc_info=True)
async def _unserve_profile(self, name: str, home: "Path") -> None:
"""Stop and unroute one deleted profile: cancel its reconnects, tear down its adapters, drop its
bookkeeping and release this process's handles into its home so the deleter's rmtree succeeds."""
from gateway.run import _write_runtime_status_quiet
pending = (getattr(self, "_profile_failed_platforms", None) or {}).pop(name, None) or {}
tasks = [t for t in pending.values() if isinstance(t, asyncio.Task) and not t.done()]
for task in tasks:
task.cancel()
if tasks:
await asyncio.wait(tasks, timeout=self._adapter_disconnect_timeout_secs())
adapters = (getattr(self, "_profile_adapters", None) or {}).pop(name, None) or {}
for platform, adapter in list(adapters.items()):
await self._bounded_adapter_teardown(adapter, platform, profile=name)
_write_runtime_status_quiet(platform=f"{name}:{platform.value}", platform_state="stopped")
for attr in ("pairing_stores", "_busy_text_modes_by_profile", "_busy_input_modes_by_profile"):
store = getattr(self, attr, None)
if isinstance(store, dict):
store.pop(name, None)
if isinstance(self._served_profile_homes, dict):
self._served_profile_homes.pop(name, None)
if isinstance(self._served_profile_signatures, dict):
self._served_profile_signatures.pop(name, None)
prefix = f"agent:{name}:"
cache = getattr(self, "_agent_cache", None)
for key in [k for k in list(cache or {}) if str(k).startswith(prefix)]:
with _log_suppressed(logging.DEBUG, "agent eviction failed for %s", key, exc_info=True):
self._evict_cached_agent(key)
with _log_suppressed(logging.DEBUG, "profile handle release failed", exc_info=True):
from hermes_state_registry import close_all_under
close_all_under(home)
with _log_suppressed(logging.DEBUG, "memory-store release failed", exc_info=True):
from plugins.memory.holographic.store import MemoryStore
MemoryStore.release_all_under(home)
logger.info("[MULTIPLEX] Profile '%s' deleted — %d adapter(s) stopped and unrouted", name, len(adapters))
+3 -1
View File
@@ -1305,7 +1305,9 @@ class GatewayStartupMixin:
"_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")
_POST_RECONNECT_WATCHERS = (
"_handoff_watcher", "_async_delegation_watcher", "_loop_wakeup_watcher", "_profile_reconcile_watcher",
)
def _start_spawn_background_watchers(self) -> None:
"""Spawn the long-lived supervised background watchers."""
+2 -2
View File
@@ -406,8 +406,8 @@ def build_migration_plan() -> MigrationPlan:
for check in _PREFLIGHT_CHECKS:
check(plan, configs)
plan.notices.append(
"Profiles created after the migration are served after `hermes gateway restart` "
"(the multiplexer snapshots the profile set at startup)."
"Profiles created after the migration are served by the running multiplexer as soon as "
"they exist (it rescans profiles/ on create/delete and every 30s)."
)
return plan
+20
View File
@@ -72,3 +72,23 @@ def served_profile_ingress_urls(profile: Optional[str] = None) -> dict[str, dict
def format_ingress_url_lines(urls: dict[str, str], indent: str = " ") -> list[str]:
"""One ``<indent><platform>: <url>`` line per platform, sorted."""
return [f"{indent}{platform}: {url}" for platform, url in sorted(urls.items())]
def notify_multiplexer_profiles_changed(profile_name: str, *, timeout: float = 8.0) -> Optional[list[str]]:
"""Tell the live default multiplexer that ``profiles/`` changed (``profile_name`` was created or
deleted) so it hot-serves / unroutes it now instead of at its next periodic rescan. Returns the
served-profile list the gateway answered with, or None when no multiplexer answered (no live default
gateway, single-profile gateway, or a gateway predating the verb). Never raises."""
try:
from hermes_constants import get_default_hermes_root
from gateway.control_socket import rescan_gateway_profiles
if live_default_gateway_pid() is None:
return None
answer = rescan_gateway_profiles(get_default_hermes_root(), timeout=timeout)
except Exception:
logger.debug("multiplexer rescan notification failed for %r", profile_name, exc_info=True)
return None
if not isinstance(answer, dict) or answer.get("multiplex") is False:
return None
served = answer.get("served_profiles")
return [str(p) for p in served] if isinstance(served, list) else None
+6 -2
View File
@@ -209,8 +209,12 @@ def _profile_create(args):
print(f" {name} setup Configure API keys and model")
print(f" {name} chat Start chatting")
from hermes_cli.gateway_multiplex_served import live_default_gateway_pid, recorded_served_profiles
if live_default_gateway_pid() is not None and recorded_served_profiles() is not None:
# The multiplexer snapshots the profile set at startup: a new profile is served only after a restart.
from hermes_cli.profiles import normalize_profile_name
served = recorded_served_profiles() if live_default_gateway_pid() is not None else None
if served is not None and normalize_profile_name(name) in {normalize_profile_name(p) for p in served}:
print(" (served now by the running multiplexed gateway — add its bot token and it connects)")
elif served is not None:
# The multiplexer did not pick the profile up (older gateway or the signal failed): a restart serves it.
print(" hermes gateway restart Serve this profile from the running multiplexed gateway")
else:
print(f" {name} gateway start Start the messaging gateway")
+11
View File
@@ -868,9 +868,17 @@ def create_profile(
# `hermes -p <profile> gateway start` supervises via `s6-svc -u` instead of a bare
# process. No-op on host (systemd/launchd/windows unit generation handles lifecycle).
_maybe_register_gateway_service(canon)
# A running multiplexer enumerates profiles/ at boot: ask it to serve this one now (it also
# rescans periodically, so a missed signal only delays serving).
_notify_multiplexer(canon)
return profile_dir
def _notify_multiplexer(canon: str) -> None:
from hermes_cli.gateway_multiplex_served import notify_multiplexer_profiles_changed
notify_multiplexer_profiles_changed(canon)
def seed_profile_skills(profile_dir: Path, quiet: bool = False) -> Optional[dict]:
"""Seed bundled skills into a profile via subprocess (sync_skills() caches HERMES_HOME at
module level). Returns the sync result dict, or None on failure. ``--no-skills`` profiles
@@ -1153,6 +1161,9 @@ def delete_profile(name: str, yes: bool = False) -> Path:
# Tombstone before rmtree so a stale serve/logging mkdir cannot relist this name live.
mark_named_profile_deleted(profile_dir)
# The multiplexer sees the tombstone, stops this profile's adapters and releases its handles
# into the directory before we remove it.
_notify_multiplexer(canon)
# Release this process's holographic memory-store connections into the profile. The
# Desktop's main serve process opens memory_store.db for every profile and is
+12 -1
View File
@@ -885,7 +885,18 @@ async def update_messaging_platform(platform_id: str, body: MessagingPlatformUpd
"env_keys=%s cleared_keys=%s",
platform_id, target_profile or "current", body.enabled, sorted(body.env), sorted(body.clear_env),
)
return {"ok": True, "platform": platform_id}
# A live multiplexer serving this named profile builds the adapter from the new token now
# (its periodic rescan would otherwise pick it up within a cycle); no gateway restart.
hot_served = bool(target_profile) and await asyncio.to_thread(_notify_multiplexer_hot_serve, target_profile)
return {"ok": True, "platform": platform_id, "hot_served": hot_served}
def _notify_multiplexer_hot_serve(profile: str) -> bool:
from hermes_cli.gateway import named_profile_served_by_running_multiplexer
from hermes_cli.gateway_multiplex_served import notify_multiplexer_profiles_changed
if not named_profile_served_by_running_multiplexer(profile):
return False
return notify_multiplexer_profiles_changed(profile) is not None
@router.post("/api/messaging/platforms/{platform_id}/test")
+39
View File
@@ -897,3 +897,42 @@ def test_multiplex_recovery_isolates_profile_failures(tmp_path):
assert recovery_homes == [str(failing_home), str(healthy_home)]
# The failing profile stays in rotation: its ledger may still hold jobs.
assert set(tick_homes) == {str(failing_home), str(healthy_home)}
def test_multiplex_ticker_reenumerates_profiles_each_cycle(tmp_path):
"""Hot-serve: with a callable ``profile_homes`` the ticker re-reads the served set every cycle,
so a profile created after the multiplexer started gets its jobs fired without a restart."""
import threading
from unittest.mock import patch
from cron.scheduler_provider import InProcessCronScheduler
from hermes_constants import get_hermes_home
alpha = tmp_path / "alpha"
gamma = tmp_path / "gamma"
(alpha / "cron").mkdir(parents=True)
homes = [("alpha", alpha)]
stop = threading.Event()
ticked: list[str] = []
def _tick(*args, **kwargs):
ticked.append(str(get_hermes_home()))
if len(ticked) == 1: # "hermes profile create gamma" happens between two cycles
(gamma / "cron").mkdir(parents=True)
homes.append(("gamma", gamma))
if len(ticked) >= 4:
stop.set()
return 0
provider = InProcessCronScheduler()
with patch("cron.scheduler.tick", side_effect=_tick):
thread = threading.Thread(
target=provider.start, args=(stop,),
kwargs={"interval": 0, "profile_homes": lambda: list(homes)}, daemon=True,
)
thread.start()
thread.join(timeout=5)
stop.set()
thread.join(timeout=5)
assert not thread.is_alive()
assert str(gamma) in ticked, ticked
+164
View File
@@ -0,0 +1,164 @@
"""Hot-serve invariants for ``gateway.multiplex_profiles`` (``gateway/run_profile_reconcile.py``).
The multiplexer used to enumerate ``profiles/`` once at boot; these pin the runtime reconcile: a
profile created afterwards is served, a deleted one is torn down and unrouted, a served profile whose
config/.env changed (bot token added after create) gets its adapters, and none of it touches the other
profiles' live adapters. The cron ticker's live enumerator is covered in ``tests/cron``.
"""
import asyncio
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from gateway.config import GatewayConfig, Platform
from gateway.run import GatewayRunner
from gateway.run_profile_reconcile import profile_serve_signature
class _Adapter:
platform = Platform.DISCORD
def __init__(self, token):
self.token = token
self.disconnected = False
self.cancelled = False
async def disconnect(self):
self.disconnected = True
async def cancel_background_tasks(self):
self.cancelled = True
def _runner(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
(home / "profiles").mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
runner = object.__new__(GatewayRunner)
runner.config = GatewayConfig(multiplex_profiles=True)
runner._running = True
runner._primary_profile_name = "default"
runner.adapters = {}
runner._profile_adapters = {}
runner._profile_failed_platforms = {}
runner._failed_platforms = {}
runner._agent_cache = {}
runner._agent_cache_lock = None
runner.pairing_store = MagicMock()
runner.pairing_stores = {}
runner._adapter_disconnect_timeout_secs = lambda: 0.5
started = []
async def _start(profile_name, profile_home, claimed):
started.append(profile_name)
token = (profile_home / ".env").read_text() if (profile_home / ".env").exists() else ""
if "DISCORD_BOT_TOKEN" not in token:
return 0
runner._profile_adapters.setdefault(profile_name, {})[Platform.DISCORD] = _Adapter(token)
return 1
runner._start_one_profile_adapters = _start
runner._adapter_credential_fingerprint = lambda adapter: getattr(adapter, "token", None)
runner._started = started
return runner, home
def _mkprofile(home, name, env=""):
d = home / "profiles" / name
d.mkdir(parents=True, exist_ok=True)
(d / "config.yaml").write_text("model: {default: m}\n")
(d / ".env").write_text(env)
return d
def _served_record(home):
return json.loads((home / "gateway_state.json").read_text()).get("served_profiles")
@pytest.mark.asyncio
async def test_created_then_credentialed_profile_is_served_without_restart(tmp_path, monkeypatch):
runner, home = _runner(tmp_path, monkeypatch)
alpha_dir = _mkprofile(home, "alpha", "DISCORD_BOT_TOKEN=alpha-token\n")
with patch("hermes_cli.profiles.get_active_profile_name", return_value="default"):
await runner._start_secondary_profile_adapters()
alpha_adapter = runner._profile_adapters["alpha"][Platform.DISCORD]
assert _served_record(home) == ["default", "alpha"]
# 1. Created while running, no token yet: served (routes/prefixes/cron), zero adapters.
gamma_dir = _mkprofile(home, "gamma")
result = await runner.reconcile_served_profiles()
assert result["added"] == ["gamma"]
assert _served_record(home) == ["default", "alpha", "gamma"]
assert "gamma" in runner.pairing_stores
assert Platform.DISCORD not in runner._profile_adapters.get("gamma", {})
# 2. Token added afterwards: the rescan builds the adapter (never "adapter-less forever").
(gamma_dir / ".env").write_text("DISCORD_BOT_TOKEN=gamma-token\n")
result = await runner.reconcile_served_profiles()
assert result["rescanned"] == ["gamma"]
assert runner._profile_adapters["gamma"][Platform.DISCORD].token.strip().endswith("gamma-token")
# 3. A no-op rescan and the whole sequence never touched alpha's live adapter.
assert await runner.reconcile_served_profiles() == {
"added": [], "removed": [], "rescanned": [], "reason": "request",
"served_profiles": ["default", "alpha", "gamma"],
}
assert runner._profile_adapters["alpha"][Platform.DISCORD] is alpha_adapter
assert alpha_adapter.disconnected is False
assert runner._started.count("alpha") == 1
assert profile_serve_signature(alpha_dir) == runner._served_profile_signatures["alpha"]
@pytest.mark.asyncio
async def test_deleted_profile_is_torn_down_and_unrouted_others_untouched(tmp_path, monkeypatch):
runner, home = _runner(tmp_path, monkeypatch)
_mkprofile(home, "alpha", "DISCORD_BOT_TOKEN=alpha-token\n")
gamma_dir = _mkprofile(home, "gamma", "DISCORD_BOT_TOKEN=gamma-token\n")
with patch("hermes_cli.profiles.get_active_profile_name", return_value="default"):
await runner._start_secondary_profile_adapters()
alpha_adapter = runner._profile_adapters["alpha"][Platform.DISCORD]
gamma_adapter = runner._profile_adapters["gamma"][Platform.DISCORD]
runner._agent_cache = {"agent:gamma:discord:dm:1": ("agent",), "agent:alpha:discord:dm:1": ("agent",)}
evicted = []
runner._evict_cached_agent = evicted.append
reconnect = asyncio.get_running_loop().create_task(asyncio.sleep(3600))
runner._profile_failed_platforms = {"gamma": {Platform.TELEGRAM: reconnect}}
from hermes_constants import mark_named_profile_deleted
mark_named_profile_deleted(gamma_dir) # what ``delete_profile`` does before rmtree
result = await runner.reconcile_served_profiles()
assert result["removed"] == ["gamma"]
assert gamma_adapter.disconnected is True and gamma_adapter.cancelled is True
assert "gamma" not in runner._profile_adapters
assert "gamma" not in runner.pairing_stores
assert reconnect.cancelled()
assert evicted == ["agent:gamma:discord:dm:1"]
assert _served_record(home) == ["default", "alpha"]
assert runner._profile_adapters["alpha"][Platform.DISCORD] is alpha_adapter
assert alpha_adapter.disconnected is False
@pytest.mark.asyncio
async def test_hot_added_profile_cannot_double_claim_a_live_secondary_token(tmp_path, monkeypatch):
"""Boot's duplicate-credential guard sees every profile at once; a hot add must see the LIVE
secondaries' claims too, or the new profile starts a second poller on alpha's bot."""
runner, home = _runner(tmp_path, monkeypatch)
_mkprofile(home, "alpha", "DISCORD_BOT_TOKEN=shared\n")
seen_claims = {}
async def _start(profile_name, profile_home, claimed):
seen_claims[profile_name] = dict(claimed)
runner._profile_adapters.setdefault(profile_name, {})[Platform.DISCORD] = _Adapter("shared")
return 1
runner._start_one_profile_adapters = _start
with patch("hermes_cli.profiles.get_active_profile_name", return_value="default"):
await runner._start_secondary_profile_adapters()
_mkprofile(home, "dupe", "DISCORD_BOT_TOKEN=shared\n")
await runner.reconcile_served_profiles()
fp = GatewayRunner._adapter_credential_fingerprint(_Adapter("shared"))
assert seen_claims["dupe"].get((Platform.DISCORD, fp)) == "alpha"
+3 -1
View File
@@ -875,8 +875,10 @@ export const api = {
// Messaging platforms (gateway channels)
getMessagingPlatforms: () =>
fetchJSON<MessagingPlatformsResponse>("/api/messaging/platforms"),
// `hot_served`: a live multiplexer serving the selected named profile rebuilt its adapters from the
// new credentials right away (no gateway restart needed).
updateMessagingPlatform: (id: string, body: MessagingPlatformUpdate) =>
fetchJSON<{ ok: boolean; platform: string }>(
fetchJSON<{ ok: boolean; platform: string; hot_served?: boolean }>(
`/api/messaging/platforms/${encodeURIComponent(id)}`,
{
method: "PUT",
+12 -5
View File
@@ -215,11 +215,17 @@ export default function ChannelsPage() {
setSaving(true);
try {
const body: MessagingPlatformUpdate = { env, enabled: true };
await api.updateMessagingPlatform(editing.id, body);
showToast(`${editing.name} saved`, "success");
const result = await api.updateMessagingPlatform(editing.id, body);
showToast(
result.hot_served
? `${editing.name} saved; the running gateway is connecting`
: `${editing.name} saved`,
"success",
);
setEditing(null);
setRestartNeeded(true);
if (!result.hot_served) setRestartNeeded(true);
await load();
if (result.hot_served) setTimeout(() => void load(), 4000);
} catch (e) {
showToast(`Failed to save: ${e}`, "error");
} finally {
@@ -231,7 +237,7 @@ export default function ChannelsPage() {
const next = !platform.enabled;
setTogglingId(platform.id);
try {
await api.updateMessagingPlatform(platform.id, { enabled: next });
const result = await api.updateMessagingPlatform(platform.id, { enabled: next });
setPlatforms((prev) =>
prev.map((p) =>
p.id === platform.id
@@ -239,7 +245,8 @@ export default function ChannelsPage() {
: p,
),
);
setRestartNeeded(true);
if (result.hot_served) setTimeout(() => void load(), 4000);
else setRestartNeeded(true);
} catch (e) {
showToast(`Error: ${e}`, "error");
} finally {
@@ -430,9 +430,18 @@ down for any profile a running multiplexer or its own gateway already serves). A
multiplexer started as `hermes -p <name> gateway run` always ticks its own
profile's cron store as well.
One caveat: the served set is a **start-time snapshot**. A profile created while
the multiplexer is running is not picked up until `hermes gateway restart`
(profiles deleted at runtime are dropped from cron ticking automatically).
The served set is **live**. A profile created while the multiplexer is running
(`hermes profile create`, the dashboard, Desktop or the TUI) is served at once:
the creator pings the multiplexer over its control socket, and the multiplexer
also rescans `profiles/` every 30 seconds as a safety net. The new profile's
adapters are built the moment its `config.yaml`/`.env` carries a bot token
(creators usually create first, then add the token), `served_profiles` in the
default profile's `gateway_state.json` is updated, and `hermes -p <name> gateway
status` reports it as served — no restart, and the other profiles' adapters and
in-flight turns are untouched. Deleting a profile stops and unroutes its
adapters the same way. The one-credential-one-poller rule still applies: a
hot-added profile that reuses another profile's token is parked with a
`duplicate_credential` error, never started as a second poller.
### Routing shared-bot chats to profiles (`profile_routes`)
@@ -827,10 +836,10 @@ prefixed URL; nothing else about the key changes.
### Profiles created after the migration
The multiplexer snapshots the profile set at startup. `hermes profile create`
prints the reminder when a live multiplexer is detected: run
`hermes gateway restart` (from the default profile) and the new profile is
served.
A profile created while the multiplexer runs is served without a restart (see
above). `hermes profile create` confirms this when the live multiplexer picked the
profile up; it prints the `hermes gateway restart` reminder only when it could not
reach the multiplexer (for example, a gateway started from an older build).
### Rollback