Gateway multiplex bot token resolution across profiles

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #65525 fix(gateway): multiplex primary gateways resolve bot tokens through profile secret scope (#64986 salvage)

  merged to main

  files: gateway/config.py, gateway/run.py, scripts/release.py, tests/gateway/test_64674_multiplex_primary_token_scope.py

  Show the diff

  • gateway/config.py

    @@ -440,6 +440,22 @@ def from_dict(cls, data: Dict[str, Any]) -> "ChannelOverride":
             )
     
     
    +# Canonical map of platforms whose primary credential is ``PlatformConfig.token``
    +# and the env var it loads from. Used for empty-token warnings at config
    +# validation and by the multiplex primary-startup credential gate in
    +# ``gateway.run`` (#64674). Platforms absent from this map authenticate some
    +# other way (session files, port-bound webhooks, api_key-only) and must never
    +# be skipped for a missing token.
    +PLATFORM_TOKEN_ENV_NAMES: dict["Platform", str] = {
    +    Platform.TELEGRAM: "TELEGRAM_BOT_TOKEN",
    +    Platform.DISCORD: "DISCORD_BOT_TOKEN",
    +    Platform.SLACK: "SLACK_BOT_TOKEN",
    +    Platform.MATTERMOST: "MATTERMOST_TOKEN",
    +    Platform.MATRIX: "MATRIX_ACCESS_TOKEN",
    +    Platform.WEIXIN: "WEIXIN_TOKEN",
    +}
    +
    +
     @dataclass
     class PlatformConfig:
         """Configuration for a single messaging platform."""
    @@ -1424,14 +1440,7 @@ def _validate_gateway_config(config: "GatewayConfig") -> None:
     
         # Warn about empty bot tokens — platforms that loaded an empty string
         # won't connect and the cause can be confusing without a log line.
    -    _token_env_names = {
    -        Platform.TELEGRAM: "TELEGRAM_BOT_TOKEN",
    -        Platform.DISCORD: "DISCORD_BOT_TOKEN",
    -        Platform.SLACK: "SLACK_BOT_TOKEN",
    -        Platform.MATTERMOST: "MATTERMOST_TOKEN",
    -        Platform.MATRIX: "MATRIX_ACCESS_TOKEN",
    -        Platform.WEIXIN: "WEIXIN_TOKEN",
    -    }
    +    _token_env_names = [redacted:secret-assignment]
         for platform, pconfig in config.platforms.items():
             if not pconfig.enabled:
                 continue

  • gateway/run.py

    @@ -1472,6 +1472,61 @@ def _profile_runtime_scope(profile_home: "Path"):
             reset_hermes_home_override(home_token)
     
     
    +def load_gateway_config_for_runner() -> "GatewayConfig":
    +    """Load gateway config for the process-level GatewayRunner.
    +
    +    When ``gateway.multiplex_profiles`` is off, this is identical to
    +    ``load_gateway_config()`` (legacy single-profile path).
    +
    +    When multiplexing is on, reload under the default/active profile's
    +    ``_profile_runtime_scope`` so platform tokens in that profile's ``.env``
    +    resolve through the secret scope — the same path secondary profiles use
    +    in ``_start_one_profile_adapters``. Without this, primary startup calls
    +    ``load_gateway_config()`` unscoped: ``_getenv`` falls through to
    +    ``os.environ``, which often has no ``TELEGRAM_BOT_TOKEN`` once the token
    +    lives only under ``profiles/<name>/.env`` (#64674).
    +
    +    Single-profile gateways never set ``multiplex_profiles``, so they keep the
    +    unscoped load and are unaffected.
    +    """
    +    cfg = load_gateway_config()
    +    if not getattr(cfg, "multiplex_profiles", False):
    +        return cfg
    +    try:
    +        home = get_hermes_home()
    +    except Exception:
    +        return cfg
    +    try:
    +        with _profile_runtime_scope(Path(home)):
    +            return load_gateway_config()
    +    except Exception:
    +        logger.debug(
    +            "multiplex default-scope config reload failed; using unscoped load",
    +            exc_info=True,
    +        )
    +        return cfg
    +
    +
    +def _platform_has_bot_credential(platform: "Platform", platform_config: "PlatformConfig") -> bool:
    +    """Return True when a token-authenticated platform has a usable bot credential.
    +
    +    Platforms that do not use ``PlatformConfig.token`` always return True so we
    +    never skip them here (Signal session paths, port-binding HTTP adapters, etc.).
    +    """
    +    from gateway.config import PLATFORM_TOKEN_ENV_NAMES
    +
    +    if platform not in PLATFORM_TOKEN_ENV_NAMES:
    +        return True
    +    token = [redacted:secret-assignment] "token", None) or ""
    +    if isinstance(token, str) and token.strip():
    +        return True
    +    # Some adapters also accept api_key as the primary credential.
    +    api_key = [redacted:secret-assignment] "api_key", None) or ""
    +    if isinstance(api_key, str) and api_key.strip():
    +        return True
    +    return False
    +
    +
     _DOCKER_VOLUME_SPEC_RE = re.compile(r"^(?P<host>.+):(?P<container>/[^:]+?)(?::(?P<options>[^:]+))?$")
     _DOCKER_MEDIA_OUTPUT_CONTAINER_PATHS = {"/output", "/outputs"}
     
    @@ -2858,7 +2913,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
     
         def __init__(self, config: Optional[GatewayConfig] = None):
             global _gateway_runner_ref
    -        self.config = config or load_gateway_config()
    +        # When multiplex_profiles is on, load under the default profile secret
    +        # scope so bot tokens in that profile's .env resolve the same way
    +        # secondary profiles do (#64674). Explicit config= injection (tests)
    +        # is left untouched.
    +        self.config = config if config is not None else load_gateway_config_for_runner()
             # Mark the process as a profile multiplexer when configured. This flips
             # agent.secret_scope.get_secret() to fail-closed on any unscoped
             # credential read, so a missed migration crashes loudly instead of
    @@ -7165,11 +7224,29 @@ async def start(self) -> bool:
             startup_retryable_errors: list[str] = []
             
             # Initialize and connect each configured platform
    +        _multiplex_on = bool(getattr(self.config, "multiplex_profiles", False))
    +        _multiplex_skipped_platforms: list[Platform] = []
             for platform, platform_config in self.config.platforms.items():
                 if await self._abort_startup_if_shutdown_requested():
                     return True
                 if not platform_config.enabled:
                     continue
    +            # Under multiplexing, a platform may be enabled on the default
    +            # profile's config.yaml while its bot token lives only in a
    +            # secondary profile's .env. 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)
    @@ -7314,6 +7391,25 @@ async def start(self) -> bool:
             except Exception as e:
                 logger.error("Secondary-profile adapter startup failed: %s", e, exc_info=True)
     
    +        # A platform we skipped on the primary for a missing credential was
    +        # supposed to be picked up by a secondary profile that owns the token.
    +        # 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:
    +            _served_by_secondary = any(
    +                _skipped in _profile_map
    +                for _profile_map in self._profile_adapters.values()
    +            )
    +            if not _served_by_secondary:
    +                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,
    +                )
    +
             if connected_count == 0:
                 if startup_nonretryable_errors:
                     reason = "; ".join(startup_nonretryable_errors)
    @@ -8008,6 +8104,17 @@ async def _platform_reconnect_watcher(self) -> None:
     
                     platform_config = info["config"]
                     attempt = info["attempts"] + 1
    +                # Empty-token primary configs can never reconnect; drop them so
    +                # multiplex setups where a secondary profile owns the bot do
    +                # not spin forever (#64674).
    +                if not _platform_has_bot_credential(platform, platform_config):
    +                    logger.warning(
    +                        "Reconnect %s: no bot credential on queued config, "
    +                        "removing from retry queue",
    +                        platform.value,
    +                    )
    +                    del self._failed_platforms[platform]
    +                    continue
                     logger.info(
                         "Reconnecting %s (attempt %d)...",
                         platform.value, attempt,

  • scripts/release.py

    @@ -45,6 +45,7 @@
     
     # Auto-extracted from noreply emails + manual overrides
     AUTHOR_MAP = {
    +    "[redacted:email]": "SAMBAS123",  # PR #64986 salvage (gateway: multiplex primary bot token scope)
         "[email protected]": "2751738943",  # PR #54785 salvage (tui: post-turn completion ownership routing)
         "[email protected]": "Burgunthy",  # PR #20096 salvage (gateway: profile-based routing for inbound messages)
         "[email protected]": "webtecnica",  # PR #63360 salvage (nous: restore inference-api base_url)

  • tests/gateway/test_64674_multiplex_primary_token_scope.py

    @@ -0,0 +1,216 @@
    +"""#64674 — multiplex primary gateway must not fail forever without bot tokens.
    +
    +When gateway.multiplex_profiles is on and TELEGRAM_BOT_TOKEN lives only in a
    +secondary profile's .env, the default-profile primary adapter used to start
    +with an empty token, log "No bot token configured", and queue an infinite
    +reconnect loop. Secondary profiles already load under _profile_runtime_scope;
    +this suite locks the complementary primary-path fixes:
    +
    +1. load_gateway_config_for_runner reloads under the default profile secret scope
    +   when multiplex is on (so default .env tokens resolve like secondary loads).
    +2. Primary startup skips token platforms that still have no credential under
    +   multiplex instead of connecting-and-failing forever.
    +3. The reconnect watcher drops empty-token queued configs.
    +"""
    +from __future__ import annotations
    +
    +from pathlib import Path
    +from types import SimpleNamespace
    +from unittest.mock import MagicMock
    +
    +import pytest
    +
    +from gateway.config import GatewayConfig, Platform, PlatformConfig
    +
    +
    [redacted:email](autouse=True)
    +def _reset_multiplex_flag():
    +    from agent import secret_scope as ss
    +
    +    ss.set_multiplex_active(False)
    +    yield
    +    ss.set_multiplex_active(False)
    +
    +
    +class TestLoadGatewayConfigForRunner:
    +    def test_unscoped_when_multiplex_off(self, tmp_path, monkeypatch):
    +        from gateway import run as run_mod
    +
    +        home = tmp_path / "home"
    +        home.mkdir()
    +        (home / ".env").write_text("TELEGRAM_BOT_TOKEN=[redacted:secret-assignment]", encoding="utf-8")
    +        (home / "config.yaml").write_text("gateway:\n  multiplex_profiles: false\n", encoding="utf-8")
    +        monkeypatch.setenv("HERMES_HOME", str(home))
    +        monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False)
    +
    +        # Without multiplex, dotenv is still loaded into os.environ by the
    +        # normal env loader in real gateways; here we only assert the helper
    +        # returns a non-multiplex config without requiring a scope.
    +        cfg = run_mod.load_gateway_config_for_runner()
    +        assert cfg.multiplex_profiles is False
    +
    +    def test_scoped_reload_picks_up_default_profile_token(self, tmp_path, monkeypatch):
    +        """Token only in default profile .env, not in process os.environ."""
    +        from gateway import run as run_mod
    +        import hermes_constants as hc
    +
    +        home = tmp_path / "home"
    +        home.mkdir()
    +        (home / ".env").write_text(
    +            "TELEGRAM_BOT_TOKEN=[redacted:secret-assignment]", encoding="utf-8"
    +        )
    +        (home / "config.yaml").write_text(
    +            "gateway:\n  multiplex_profiles: true\n", encoding="utf-8"
    +        )
    +        monkeypatch.setenv("HERMES_HOME", str(home))
    +        # Simulate a clean process env where the token was NOT exported and
    +        # was not bulk-loaded into os.environ (multiplex isolation path).
    +        monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False)
    +        # Point both hermes_constants and gateway.run at our temp home.
    +        monkeypatch.setattr(hc, "get_hermes_home", lambda: home)
    +        monkeypatch.setattr(run_mod, "get_hermes_home", lambda: home)
    +        monkeypatch.setattr(run_mod, "_hermes_home", home)
    +
    +        cfg = run_mod.load_gateway_config_for_runner()
    +        assert cfg.multiplex_profiles is True
    +        tg = cfg.platforms.get(Platform.TELEGRAM)
    +        assert tg is not None
    +        assert tg.token == "default-profile-token-123"
    +        assert tg.enabled is True
    +
    +
    +class TestPlatformHasBotCredential:
    +    def test_telegram_empty_token_false(self):
    +        from gateway.run import _platform_has_bot_credential
    +
    +        assert _platform_has_bot_credential(
    +            Platform.TELEGRAM, PlatformConfig(enabled=True, token="")
    +        ) is False
    +        assert _platform_has_bot_credential(
    +            Platform.TELEGRAM, PlatformConfig(enabled=True, token=None)
    +        ) is False
    +
    +    def test_telegram_with_token_true(self):
    +        from gateway.run import _platform_has_bot_credential
    +
    +        assert _platform_has_bot_credential(
    +            Platform.TELEGRAM, PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +        ) is True
    +
    +    def test_non_token_platform_always_true(self):
    +        from gateway.run import _platform_has_bot_credential
    +
    +        # SMS / webhook-style platforms are not gated by PlatformConfig.token.
    +        # Use a platform that exists but is outside the token set when possible.
    +        for plat in Platform:
    +            if plat in {
    +                Platform.TELEGRAM,
    +                Platform.DISCORD,
    +                Platform.SLACK,
    +                Platform.MATTERMOST,
    +                Platform.MATRIX,
    +                Platform.WEIXIN,
    +            }:
    +                continue
    +            assert _platform_has_bot_credential(
    +                plat, PlatformConfig(enabled=True, token=None)
    +            ) is True
    +            break
    +
    +
    +class TestPrimaryStartupSkipsEmptyTokenUnderMultiplex:
    +    @pytest.mark.asyncio
    +    async def test_skips_empty_telegram_when_multiplex_on(self, monkeypatch):
    +        from gateway.run import GatewayRunner
    +
    +        cfg = GatewayConfig(multiplex_profiles=True)
    +        cfg.platforms[Platform.TELEGRAM] = PlatformConfig(
    +            enabled=True, token=""  # empty — lives on secondary only
    +        )
    +
    +        runner = GatewayRunner.__new__(GatewayRunner)
    +        # Minimal init of attributes used by the start loop body we call.
    +        runner.config = cfg
    +        runner.adapters = {}
    +        runner._failed_platforms = {}
    +        runner._profile_adapters = {}
    +        runner._busy_text_mode = "off"
    +        runner.session_store = MagicMock()
    +        runner._shutdown_event = MagicMock()
    +        runner._running = True
    +
    +        created = []
    +
    +        def _fake_create(platform, platform_config):
    +            created.append(platform)
    +            return MagicMock()
    +
    +        runner._create_adapter = _fake_create  # type: ignore[method-assign]
    +        runner._abort_startup_if_shutdown_requested = MagicMock(return_value=False)  # type: ignore
    +        runner._update_platform_runtime_status = MagicMock()  # type: ignore
    +        runner._start_secondary_profile_adapters = MagicMock(return_value=0)  # type: ignore
    +        # Make the secondary call awaitable
    +        async def _sec():
    +            return 0
    +        runner._start_secondary_profile_adapters = _sec  # type: ignore
    +
    +        # We only want the primary platform loop; extract and run a thin
    +        # stand-in by invoking the real loop logic via a partial start is
    +        # heavy. Instead assert the skip helper path by simulating the
    +        # condition the start() loop uses.
    +        from gateway.run import _platform_has_bot_credential
    +
    +        skipped = []
    +        for platform, platform_config in cfg.platforms.items():
    +            if not platform_config.enabled:
    +                continue
    +            if cfg.multiplex_profiles and not _platform_has_bot_credential(
    +                platform, platform_config
    +            ):
    +                skipped.append(platform)
    +                continue
    +            created.append(platform)
    +
    +        assert skipped == [Platform.TELEGRAM]
    +        assert created == []
    +
    +    @pytest.mark.asyncio
    +    async def test_still_starts_when_token_present(self):
    +        from gateway.run import _platform_has_bot_credential
    +
    +        cfg = GatewayConfig(multiplex_profiles=True)
    +        cfg.platforms[Platform.TELEGRAM] = PlatformConfig(
    +            enabled=True, token="[redacted:secret-assignment]"
    +        )
    +        started = []
    +        for platform, platform_config in cfg.platforms.items():
    +            if not platform_config.enabled:
    +                continue
    +            if cfg.multiplex_profiles and not _platform_has_bot_credential(
    +                platform, platform_config
    +            ):
    +                continue
    +            started.append(platform)
    +        assert started == [Platform.TELEGRAM]
    +
    +
    +class TestReconnectDropsEmptyToken:
    +    @pytest.mark.asyncio
    +    async def test_empty_token_removed_from_queue(self):
    +        from gateway.run import GatewayRunner, _platform_has_bot_credential
    +        from gateway.config import Platform, PlatformConfig
    +
    +        # Unit-level: the branch condition the watcher uses.
    +        platform = Platform.TELEGRAM
    +        platform_config = PlatformConfig(enabled=True, token="")
    +        failed = {
    +            platform: {
    +                "config": platform_config,
    +                "attempts": 3,
    +                "next_retry": 0,
    +            }
    +        }
    +        assert not _platform_has_bot_credential(platform, platform_config)
    +        # Simulate watcher drop
    +        del failed[platform]
    +        assert failed == {}