mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 16:18:55 +00:00
fix(gateway): restore multiplex secondary adapters
This commit is contained in:
parent
7b5ba20547
commit
a7ffbbff72
5 changed files with 219 additions and 24 deletions
|
|
@ -8578,9 +8578,24 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||||
f"'{profile_name}'s config.yaml (configure it only on the "
|
f"'{profile_name}'s config.yaml (configure it only on the "
|
||||||
f"default profile)."
|
f"default profile)."
|
||||||
)
|
)
|
||||||
with _profile_runtime_scope(profile_home):
|
try:
|
||||||
adapter = self._create_adapter(platform, platform_config)
|
with _profile_runtime_scope(profile_home):
|
||||||
|
adapter = self._create_adapter(platform, platform_config)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(
|
||||||
|
"[MULTIPLEX] Profile '%s': _create_adapter('%s') raised %s",
|
||||||
|
profile_name,
|
||||||
|
platform.value,
|
||||||
|
e,
|
||||||
|
exc_info=True,
|
||||||
|
)
|
||||||
|
continue
|
||||||
if not adapter:
|
if not adapter:
|
||||||
|
logger.warning(
|
||||||
|
"[MULTIPLEX] Profile '%s': skipping platform '%s' - adapter creation returned None",
|
||||||
|
profile_name,
|
||||||
|
platform.value,
|
||||||
|
)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
# Same-token conflict detection — refuse a duplicate poll.
|
# Same-token conflict detection — refuse a duplicate poll.
|
||||||
|
|
@ -8610,6 +8625,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||||
adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
|
adapter.set_topic_recovery_fn(self._recover_telegram_topic_thread_id)
|
||||||
adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform))
|
adapter.set_authorization_check(self._make_adapter_auth_check(adapter.platform))
|
||||||
adapter._busy_text_mode = self._busy_text_mode
|
adapter._busy_text_mode = self._busy_text_mode
|
||||||
|
setattr(adapter, "_gateway_profile_label", profile_name)
|
||||||
|
if hasattr(adapter, "config") and hasattr(adapter.config, "extra") and isinstance(adapter.config.extra, dict):
|
||||||
|
adapter.config.extra.setdefault("_gateway_profile", profile_name)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
with _profile_runtime_scope(profile_home):
|
with _profile_runtime_scope(profile_home):
|
||||||
|
|
@ -8617,12 +8635,28 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||||
if success:
|
if success:
|
||||||
profile_map[platform] = adapter
|
profile_map[platform] = adapter
|
||||||
connected += 1
|
connected += 1
|
||||||
logger.info("✓ %s connected (profile: %s)", platform.value, profile_name)
|
logger.info(
|
||||||
|
"✓ %s connected (profile: %s, token_fp=%s)",
|
||||||
|
platform.value,
|
||||||
|
profile_name,
|
||||||
|
fp or "-",
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
logger.warning("✗ %s failed to connect (profile: %s)", platform.value, profile_name)
|
logger.warning(
|
||||||
|
"✗ %s failed to connect (profile: %s, token_fp=%s)",
|
||||||
|
platform.value,
|
||||||
|
profile_name,
|
||||||
|
fp or "-",
|
||||||
|
)
|
||||||
await self._safe_adapter_disconnect(adapter, platform)
|
await self._safe_adapter_disconnect(adapter, platform)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error("✗ %s error (profile: %s): %s", platform.value, profile_name, e)
|
logger.error(
|
||||||
|
"✗ %s error (profile: %s, token_fp=%s): %s",
|
||||||
|
platform.value,
|
||||||
|
profile_name,
|
||||||
|
fp or "-",
|
||||||
|
e,
|
||||||
|
)
|
||||||
await self._safe_adapter_disconnect(adapter, platform)
|
await self._safe_adapter_disconnect(adapter, platform)
|
||||||
return connected
|
return connected
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -40,12 +40,14 @@ logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def check_ha_requirements() -> bool:
|
def check_ha_requirements() -> bool:
|
||||||
"""Check if Home Assistant dependencies are available and configured."""
|
"""Check if Home Assistant runtime dependencies are available."""
|
||||||
if not AIOHTTP_AVAILABLE:
|
return bool(AIOHTTP_AVAILABLE)
|
||||||
return False
|
|
||||||
if not os.getenv("HASS_TOKEN"):
|
|
||||||
return False
|
def validate_ha_config(config: PlatformConfig) -> bool:
|
||||||
return True
|
"""Accept either a scoped config token or an env-provided HASS_TOKEN."""
|
||||||
|
token = getattr(config, "token", None) or os.getenv("HASS_TOKEN", "")
|
||||||
|
return bool(str(token or "").strip())
|
||||||
|
|
||||||
|
|
||||||
class HomeAssistantAdapter(BasePlatformAdapter):
|
class HomeAssistantAdapter(BasePlatformAdapter):
|
||||||
|
|
@ -560,6 +562,7 @@ def register(ctx) -> None:
|
||||||
label="Home Assistant",
|
label="Home Assistant",
|
||||||
adapter_factory=_build_adapter,
|
adapter_factory=_build_adapter,
|
||||||
check_fn=check_ha_requirements,
|
check_fn=check_ha_requirements,
|
||||||
|
validate_config=validate_ha_config,
|
||||||
is_connected=_is_connected,
|
is_connected=_is_connected,
|
||||||
required_env=["HASS_TOKEN"],
|
required_env=["HASS_TOKEN"],
|
||||||
install_hint="pip install aiohttp",
|
install_hint="pip install aiohttp",
|
||||||
|
|
|
||||||
|
|
@ -678,6 +678,42 @@ class TelegramAdapter(BasePlatformAdapter):
|
||||||
# blow the gateway's connect timeout (#46298).
|
# blow the gateway's connect timeout (#46298).
|
||||||
self._post_connect_task: Optional[asyncio.Task] = None
|
self._post_connect_task: Optional[asyncio.Task] = None
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _token_fingerprint(token: Optional[str]) -> Optional[str]:
|
||||||
|
"""Return a stable, log-safe fingerprint for a Telegram bot token."""
|
||||||
|
if not isinstance(token, str):
|
||||||
|
return None
|
||||||
|
token = token.strip()
|
||||||
|
if not token:
|
||||||
|
return None
|
||||||
|
import hashlib
|
||||||
|
|
||||||
|
return hashlib.sha256(("hermes-mux:" + token).encode("utf-8")).hexdigest()[:16]
|
||||||
|
|
||||||
|
def _gateway_profile_name(self) -> str:
|
||||||
|
"""Best-effort profile label for connect-path diagnostics."""
|
||||||
|
raw = getattr(self, "_gateway_profile_label", None)
|
||||||
|
if isinstance(raw, str) and raw.strip():
|
||||||
|
return raw.strip()
|
||||||
|
extra = getattr(self.config, "extra", None)
|
||||||
|
if isinstance(extra, dict):
|
||||||
|
for key in ("_gateway_profile", "profile"):
|
||||||
|
value = extra.get(key)
|
||||||
|
if isinstance(value, str) and value.strip():
|
||||||
|
return value.strip()
|
||||||
|
return "default"
|
||||||
|
|
||||||
|
def _connect_diagnostic_summary(self, token: Optional[str] = None) -> str:
|
||||||
|
"""Profile-attributed, masked token diagnostics for connect logs."""
|
||||||
|
if token is None:
|
||||||
|
token = getattr(self.config, "token", None)
|
||||||
|
token = token.strip() if isinstance(token, str) else ""
|
||||||
|
fingerprint = self._token_fingerprint(token) or "-"
|
||||||
|
return (
|
||||||
|
f"profile={self._gateway_profile_name()} "
|
||||||
|
f"token_len={len(token)} token_fp={fingerprint}"
|
||||||
|
)
|
||||||
|
|
||||||
def _mark_connected(self) -> None:
|
def _mark_connected(self) -> None:
|
||||||
self._drop_delayed_deliveries = False
|
self._drop_delayed_deliveries = False
|
||||||
super()._mark_connected()
|
super()._mark_connected()
|
||||||
|
|
@ -3016,23 +3052,28 @@ class TelegramAdapter(BasePlatformAdapter):
|
||||||
TELEGRAM_WEBHOOK_PORT Local listen port (default 8443)
|
TELEGRAM_WEBHOOK_PORT Local listen port (default 8443)
|
||||||
TELEGRAM_WEBHOOK_SECRET Secret token for update verification
|
TELEGRAM_WEBHOOK_SECRET Secret token for update verification
|
||||||
"""
|
"""
|
||||||
|
token = getattr(self.config, "token", None)
|
||||||
|
connect_diag = self._connect_diagnostic_summary(token)
|
||||||
if not TELEGRAM_AVAILABLE:
|
if not TELEGRAM_AVAILABLE:
|
||||||
logger.error(
|
logger.error(
|
||||||
"[%s] python-telegram-bot not installed. Run: pip install python-telegram-bot",
|
"[%s] python-telegram-bot not installed (%s). Run: pip install python-telegram-bot",
|
||||||
self.name,
|
self.name,
|
||||||
|
connect_diag,
|
||||||
)
|
)
|
||||||
return False
|
return False
|
||||||
|
|
||||||
if not self.config.token:
|
if not token:
|
||||||
logger.error("[%s] No bot token configured", self.name)
|
logger.error("[%s] No bot token configured (%s)", self.name, connect_diag)
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
logger.info("[%s] Telegram connect starting (%s)", self.name, connect_diag)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if not self._acquire_platform_lock('telegram-bot-token', self.config.token, 'Telegram bot token'):
|
if not self._acquire_platform_lock('telegram-bot-token', token, 'Telegram bot token'):
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# Build the application
|
# Build the application
|
||||||
builder = Application.builder().token(self.config.token)
|
builder = Application.builder().token(token)
|
||||||
custom_base_url = self.config.extra.get("base_url")
|
custom_base_url = self.config.extra.get("base_url")
|
||||||
if custom_base_url:
|
if custom_base_url:
|
||||||
builder = builder.base_url(custom_base_url)
|
builder = builder.base_url(custom_base_url)
|
||||||
|
|
@ -3210,8 +3251,9 @@ class TelegramAdapter(BasePlatformAdapter):
|
||||||
for _attempt in range(_max_connect):
|
for _attempt in range(_max_connect):
|
||||||
try:
|
try:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"[%s] Connecting to Telegram (attempt %d/%d)…",
|
"[%s] Connecting to Telegram (attempt %d/%d, %s)…",
|
||||||
self.name, _attempt + 1, _max_connect,
|
self.name, _attempt + 1, _max_connect,
|
||||||
|
connect_diag,
|
||||||
)
|
)
|
||||||
await _await_with_thread_deadline(
|
await _await_with_thread_deadline(
|
||||||
self._app.initialize(),
|
self._app.initialize(),
|
||||||
|
|
@ -3367,7 +3409,12 @@ class TelegramAdapter(BasePlatformAdapter):
|
||||||
|
|
||||||
self._mark_connected()
|
self._mark_connected()
|
||||||
mode = "webhook" if self._webhook_mode else "polling"
|
mode = "webhook" if self._webhook_mode else "polling"
|
||||||
logger.info("[%s] Connected to Telegram (%s mode)", self.name, mode)
|
logger.info(
|
||||||
|
"[%s] Connected to Telegram (%s mode, %s)",
|
||||||
|
self.name,
|
||||||
|
mode,
|
||||||
|
connect_diag,
|
||||||
|
)
|
||||||
|
|
||||||
# Start the persistent heartbeat loop in polling mode. Webhook mode
|
# Start the persistent heartbeat loop in polling mode. Webhook mode
|
||||||
# receives updates via incoming pushes — there is no long-poll
|
# receives updates via incoming pushes — there is no long-poll
|
||||||
|
|
@ -3396,7 +3443,12 @@ class TelegramAdapter(BasePlatformAdapter):
|
||||||
safe_error = _redact_telegram_error_text(e)
|
safe_error = _redact_telegram_error_text(e)
|
||||||
message = f"Telegram startup failed: {safe_error}"
|
message = f"Telegram startup failed: {safe_error}"
|
||||||
self._set_fatal_error("telegram_connect_error", message, retryable=True)
|
self._set_fatal_error("telegram_connect_error", message, retryable=True)
|
||||||
logger.error("[%s] Failed to connect to Telegram: %s", self.name, safe_error)
|
logger.error(
|
||||||
|
"[%s] Failed to connect to Telegram (%s): %s",
|
||||||
|
self.name,
|
||||||
|
connect_diag,
|
||||||
|
safe_error,
|
||||||
|
)
|
||||||
return False
|
return False
|
||||||
|
|
||||||
async def _set_status_indicator(self, online: bool) -> None:
|
async def _set_status_indicator(self, online: bool) -> None:
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,7 @@ from gateway.config import (
|
||||||
from plugins.platforms.homeassistant.adapter import (
|
from plugins.platforms.homeassistant.adapter import (
|
||||||
HomeAssistantAdapter,
|
HomeAssistantAdapter,
|
||||||
check_ha_requirements,
|
check_ha_requirements,
|
||||||
|
validate_ha_config,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -26,12 +27,8 @@ from plugins.platforms.homeassistant.adapter import (
|
||||||
|
|
||||||
|
|
||||||
class TestCheckRequirements:
|
class TestCheckRequirements:
|
||||||
def test_returns_false_without_token(self, monkeypatch):
|
def test_returns_true_when_aiohttp_present(self, monkeypatch):
|
||||||
monkeypatch.delenv("HASS_TOKEN", raising=False)
|
monkeypatch.delenv("HASS_TOKEN", raising=False)
|
||||||
assert check_ha_requirements() is False
|
|
||||||
|
|
||||||
def test_returns_true_with_token(self, monkeypatch):
|
|
||||||
monkeypatch.setenv("HASS_TOKEN", "test-token")
|
|
||||||
assert check_ha_requirements() is True
|
assert check_ha_requirements() is True
|
||||||
|
|
||||||
@patch("plugins.platforms.homeassistant.adapter.AIOHTTP_AVAILABLE", False)
|
@patch("plugins.platforms.homeassistant.adapter.AIOHTTP_AVAILABLE", False)
|
||||||
|
|
@ -40,6 +37,20 @@ class TestCheckRequirements:
|
||||||
assert check_ha_requirements() is False
|
assert check_ha_requirements() is False
|
||||||
|
|
||||||
|
|
||||||
|
class TestValidateConfig:
|
||||||
|
def test_returns_false_without_token_in_config_or_env(self, monkeypatch):
|
||||||
|
monkeypatch.delenv("HASS_TOKEN", raising=False)
|
||||||
|
assert validate_ha_config(PlatformConfig(enabled=True)) is False
|
||||||
|
|
||||||
|
def test_returns_true_with_token_in_env(self, monkeypatch):
|
||||||
|
monkeypatch.setenv("HASS_TOKEN", "test-token")
|
||||||
|
assert validate_ha_config(PlatformConfig(enabled=True)) is True
|
||||||
|
|
||||||
|
def test_returns_true_with_token_in_config(self, monkeypatch):
|
||||||
|
monkeypatch.delenv("HASS_TOKEN", raising=False)
|
||||||
|
assert validate_ha_config(PlatformConfig(enabled=True, token="cfg-token")) is True
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# _format_state_change - pure function, all domain branches
|
# _format_state_change - pure function, all domain branches
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,9 @@
|
||||||
"""Phase 3: secondary-profile adapter registry + same-token conflict detection."""
|
"""Phase 3: secondary-profile adapter registry + same-token conflict detection."""
|
||||||
|
import asyncio
|
||||||
|
from contextlib import nullcontext
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any, cast
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from gateway.run import GatewayRunner
|
from gateway.run import GatewayRunner
|
||||||
|
|
@ -10,6 +15,42 @@ class _FakeAdapter:
|
||||||
self.config = config
|
self.config = config
|
||||||
|
|
||||||
|
|
||||||
|
class _FakeTelegramAdapter(_FakeAdapter):
|
||||||
|
def __init__(self, token=None, config=None):
|
||||||
|
super().__init__(token=token, config=config)
|
||||||
|
self.platform: Any = None
|
||||||
|
self._busy_text_mode = None
|
||||||
|
self._gateway_profile_label = None
|
||||||
|
self.message_handler = None
|
||||||
|
self.fatal_error_handler = None
|
||||||
|
self.session_store = None
|
||||||
|
self.busy_session_handler = None
|
||||||
|
self.topic_recovery_fn = None
|
||||||
|
self.authorization_check = None
|
||||||
|
self.disconnect_called = False
|
||||||
|
|
||||||
|
def set_message_handler(self, handler):
|
||||||
|
self.message_handler = handler
|
||||||
|
|
||||||
|
def set_fatal_error_handler(self, handler):
|
||||||
|
self.fatal_error_handler = handler
|
||||||
|
|
||||||
|
def set_session_store(self, session_store):
|
||||||
|
self.session_store = session_store
|
||||||
|
|
||||||
|
def set_busy_session_handler(self, handler):
|
||||||
|
self.busy_session_handler = handler
|
||||||
|
|
||||||
|
def set_topic_recovery_fn(self, fn):
|
||||||
|
self.topic_recovery_fn = fn
|
||||||
|
|
||||||
|
def set_authorization_check(self, fn):
|
||||||
|
self.authorization_check = fn
|
||||||
|
|
||||||
|
async def disconnect(self):
|
||||||
|
self.disconnect_called = True
|
||||||
|
|
||||||
|
|
||||||
class TestCredentialFingerprint:
|
class TestCredentialFingerprint:
|
||||||
def test_none_without_token(self):
|
def test_none_without_token(self):
|
||||||
assert GatewayRunner._adapter_credential_fingerprint(_FakeAdapter()) is None
|
assert GatewayRunner._adapter_credential_fingerprint(_FakeAdapter()) is None
|
||||||
|
|
@ -187,6 +228,60 @@ class TestPortBindingHardError:
|
||||||
assert duplicate.disconnected is True
|
assert duplicate.disconnected is True
|
||||||
assert runner._profile_adapters["reviewer"] == {}
|
assert runner._profile_adapters["reviewer"] == {}
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_secondary_telegram_connect_logs_profile_and_token_fingerprint(self, monkeypatch, caplog):
|
||||||
|
"""Secondary-profile Telegram startup logs must attribute profile + token.
|
||||||
|
|
||||||
|
Regression guard for the multiplex diagnostic blind spot: when a
|
||||||
|
secondary Telegram adapter hangs or fails, the logs must still tell the
|
||||||
|
operator WHICH profile/token was in play, and the adapter must inherit
|
||||||
|
the profile label before connect() runs.
|
||||||
|
"""
|
||||||
|
import logging
|
||||||
|
from gateway.config import GatewayConfig, Platform, PlatformConfig
|
||||||
|
|
||||||
|
runner = GatewayRunner.__new__(GatewayRunner)
|
||||||
|
runner.config = GatewayConfig(multiplex_profiles=True)
|
||||||
|
runner._profile_adapters = {}
|
||||||
|
runner._busy_text_mode = "interrupt"
|
||||||
|
runner.session_store = cast(Any, object())
|
||||||
|
runner._handle_adapter_fatal_error = cast(Any, object())
|
||||||
|
runner._handle_active_session_busy_message = cast(Any, object())
|
||||||
|
runner._recover_telegram_topic_thread_id = cast(Any, object())
|
||||||
|
runner._make_profile_message_handler = cast(Any, lambda profile_name: f"handler:{profile_name}")
|
||||||
|
runner._make_adapter_auth_check = cast(Any, lambda platform: "auth-check")
|
||||||
|
|
||||||
|
reviewer_cfg = GatewayConfig(multiplex_profiles=True)
|
||||||
|
reviewer_cfg.platforms = {
|
||||||
|
Platform.TELEGRAM: PlatformConfig(enabled=True, token="reviewer-token-123"),
|
||||||
|
}
|
||||||
|
adapter = _FakeTelegramAdapter(config=reviewer_cfg.platforms[Platform.TELEGRAM])
|
||||||
|
adapter.platform = Platform.TELEGRAM
|
||||||
|
|
||||||
|
monkeypatch.setattr("gateway.config.load_gateway_config", lambda: reviewer_cfg)
|
||||||
|
monkeypatch.setattr("gateway.run._profile_runtime_scope", lambda _home: nullcontext())
|
||||||
|
monkeypatch.setattr(runner, "_create_adapter", lambda p, c: adapter)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
runner,
|
||||||
|
"_connect_adapter_with_timeout",
|
||||||
|
lambda a, p: asyncio.sleep(0, result=True),
|
||||||
|
)
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="gateway.run"):
|
||||||
|
connected = await runner._start_one_profile_adapters("reviewer", Path("/tmp/reviewer"), {})
|
||||||
|
|
||||||
|
assert connected == 1
|
||||||
|
assert adapter._gateway_profile_label == "reviewer"
|
||||||
|
assert adapter.config.extra.get("_gateway_profile") == "reviewer"
|
||||||
|
expected_fp = GatewayRunner._adapter_credential_fingerprint(adapter)
|
||||||
|
messages = [record.getMessage() for record in caplog.records]
|
||||||
|
assert any(
|
||||||
|
"✓ telegram connected" in msg
|
||||||
|
and "profile: reviewer" in msg
|
||||||
|
and f"token_fp={expected_fp}" in msg
|
||||||
|
for msg in messages
|
||||||
|
)
|
||||||
|
|
||||||
def test_port_binding_set_covers_known_listeners(self):
|
def test_port_binding_set_covers_known_listeners(self):
|
||||||
from gateway.run import _PORT_BINDING_PLATFORM_VALUES
|
from gateway.run import _PORT_BINDING_PLATFORM_VALUES
|
||||||
# Every adapter that binds a TCP port must be in the guard set.
|
# Every adapter that binds a TCP port must be in the guard set.
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue