fix(gateway): restore multiplex secondary adapters

This commit is contained in:
Jay Stothard 2026-07-12 15:11:26 +00:00
parent 7b5ba20547
commit a7ffbbff72
5 changed files with 219 additions and 24 deletions

View file

@ -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

View file

@ -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",

View file

@ -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:

View file

@ -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
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------

View file

@ -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.