hermes-agent/tests/gateway/test_startup_connect_parallel.py

"""Regression tests for parallel platform connect at gateway startup (#83791).

The old ``GatewayRunner.start()`` loop awaited each platform's connect()
(including its own timeout) in turn. A single slow/failing platform (e.g.
Telegram behind a dead proxy) therefore delayed every later platform's
connect by a full timeout window, cascading one platform's failure onto
WeChat/QQ/etc. These tests prove the connects now run concurrently.

Why event-order, not wall-clock timings
---------------------------------------
An earlier version of this test recorded ``time.monotonic()`` around each
connect() and asserted ``slow_start < fast_end``. That assertion is true in
BOTH the serial and the parallel world, so it proved nothing:

  serial:   slow_start=0, slow_end=0.300, fast_start=0.300, fast_end=0.300
            -> 0 < 0.300  (passes, but it's serial!)
  parallel: slow_start=0, fast_start=0,    fast_end=0.001, slow_end=0.300
            -> 0 < 0.001  (passes)

The only assertion that distinguishes them is ``fast_end`` occurring *before*
``slow_end`` (true only when the two connects overlap). We record the
connect start/end events in arrival order, which is fully independent of clock
resolution -- ``time.monotonic()`` has only ~15 ms resolution on Windows
(GetTickCount64), so parallel connects can land on the same tick and defeat any
wall-clock comparison. Event ordering cannot be defeated by a coarse clock.
"""

import asyncio

import pytest

from gateway.config import GatewayConfig, Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter
from gateway.run import GatewayRunner


class _OrderRecorder:
    """Collects connect start/end events in arrival order (clock-agnostic)."""

    events: list = []

    @classmethod
    def reset(cls) -> None:
        cls.events = []

    @classmethod
    def index_of(cls, platform_value: str, kind: str) -> int:
        for i, (name, evt) in enumerate(cls.events):
            if name == platform_value and evt == kind:
                return i
        return -1


class _TimingAdapter(BasePlatformAdapter):
    """Adapter whose ``connect()`` records an event and sleeps.

    Used to prove the startup connect loop launches every platform's
    connect() concurrently rather than serially.
    """

    def __init__(self, platform: Platform, sleep: float):
        super().__init__(PlatformConfig(enabled=True, token="***"), platform)
        self._sleep = sleep

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        _OrderRecorder.events.append((self.platform.value, "start"))
        await asyncio.sleep(self._sleep)
        _OrderRecorder.events.append((self.platform.value, "end"))
        return True

    async def disconnect(self) -> None:
        self._mark_disconnected()

    async def send(self, chat_id, content, reply_to=None, metadata=None):
        raise NotImplementedError

    async def get_chat_info(self, chat_id):
        return {"id": chat_id}


@pytest.mark.asyncio
async def test_startup_connects_platforms_concurrently(monkeypatch, tmp_path):
    """A slow platform must not block a later platform at startup (#83791).

    "slow" (Telegram) is listed first so a serial loop would fully block
    "fast" (Discord). We prove the connect calls overlap by recording the
    order in which connects finish: under a serial loop the slow platform's
    connect ends *before* the fast one even begins, so the fast platform's
    end can never precede the slow platform's end. Only parallel execution
    puts ``fast_end`` before ``slow_end``.
    """
    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    _OrderRecorder.reset()

    config = GatewayConfig(
        platforms={
            Platform.TELEGRAM: PlatformConfig(enabled=True, token="***"),
            Platform.DISCORD: PlatformConfig(enabled=True, token="***"),
        },
        sessions_dir=tmp_path / "sessions",
    )
    runner = GatewayRunner(config)

    def _make_adapter(platform, platform_config):
        sleep = 0.3 if platform is Platform.TELEGRAM else 0.0
        return _TimingAdapter(platform, sleep)

    monkeypatch.setattr(runner, "_create_adapter", _make_adapter)
    # Keep the rest of startup lightweight / non-fatal.
    monkeypatch.setattr(runner, "_start_secondary_profile_adapters", lambda: 0)

    await runner.start()

    events = _OrderRecorder.events
    assert events, "no connect() event was recorded"

    fast_end = _OrderRecorder.index_of(Platform.DISCORD.value, "end")
    slow_end = _OrderRecorder.index_of(Platform.TELEGRAM.value, "end")
    assert fast_end != -1 and slow_end != -1, f"missing end events: {events}"

    # Overlap proof: the fast platform finished before the slow one did,
    # which is only possible if the two connects ran at the same time.
    assert fast_end < slow_end, (
        f"connects did not overlap (serial loop?): events={events}"
    )
    # Both platforms should be registered once startup settles.
    assert Platform.TELEGRAM in runner.adapters
    assert Platform.DISCORD in runner.adapters


@pytest.mark.asyncio
async def test_startup_one_failing_platform_does_not_block_others(monkeypatch, tmp_path):
    """A failing/slow platform must not prevent others from connecting (#83791).

    Mirrors the reported Windows symptom: Telegram (dead proxy) must not keep
    WeChat/QQ offline. Here Telegram fails (returns False after a sleep) while
    Discord connects successfully and is registered.
    """

    class _FailingSlowAdapter(BasePlatformAdapter):
        def __init__(self):
            super().__init__(PlatformConfig(enabled=True, token="***"), Platform.TELEGRAM)

        async def connect(self, *, is_reconnect: bool = False) -> bool:
            await asyncio.sleep(0.3)
            self._set_fatal_error("telegram_proxy_dead", "proxy unreachable", retryable=True)
            return False

        async def disconnect(self) -> None:
            self._mark_disconnected()

        async def send(self, chat_id, content, reply_to=None, metadata=None):
            raise NotImplementedError

        async def get_chat_info(self, chat_id):
            return {"id": chat_id}

    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    _OrderRecorder.reset()

    config = GatewayConfig(
        platforms={
            Platform.TELEGRAM: PlatformConfig(enabled=True, token="***"),
            Platform.DISCORD: PlatformConfig(enabled=True, token="***"),
        },
        sessions_dir=tmp_path / "sessions",
    )
    runner = GatewayRunner(config)

    def _make_adapter(platform, platform_config):
        if platform is Platform.TELEGRAM:
            return _FailingSlowAdapter()
        return _TimingAdapter(platform, 0.0)

    monkeypatch.setattr(runner, "_create_adapter", _make_adapter)
    monkeypatch.setattr(runner, "_start_secondary_profile_adapters", lambda: 0)

    await runner.start()

    # The healthy platform connected and is registered despite Telegram failing.
    assert Platform.DISCORD in runner.adapters
    # The failed platform is queued for retry, not silently dropped.
    assert Platform.TELEGRAM in runner._failed_platforms


class TestTelegramColdStartCap:
    """The initial (pre-`running`) Telegram connect uses a capped budget (#85993).

    The full 180s Telegram connect budget (#67498) still applies to reconnect
    watcher retries; only the cold-start attempt awaited before the gateway
    reaches `running` is capped, so an unreachable Telegram can't hold every
    other platform's serving state hostage for 3 minutes.
    """

    def _runner(self, tmp_path):
        config = GatewayConfig(
            platforms={}, sessions_dir=tmp_path / "sessions"
        )
        return GatewayRunner(config)

    def test_initial_telegram_budget_is_capped(self, tmp_path, monkeypatch):
        monkeypatch.delenv("HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT", raising=False)
        runner = self._runner(tmp_path)
        initial = runner._platform_connect_timeout_secs(
            Platform.TELEGRAM, initial=True
        )
        full = runner._platform_connect_timeout_secs(Platform.TELEGRAM)
        assert initial < full, (
            "cold-start Telegram budget must be shorter than the reconnect "
            f"budget (initial={initial}, full={full})"
        )
        assert full == 180.0  # #67498 reconnect budget unchanged
        assert initial <= 60.0  # gateway reaches `running` within a minute

    def test_other_platforms_unchanged(self, tmp_path, monkeypatch):
        monkeypatch.delenv("HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT", raising=False)
        runner = self._runner(tmp_path)
        assert runner._platform_connect_timeout_secs(
            Platform.DISCORD, initial=True
        ) == runner._platform_connect_timeout_secs(Platform.DISCORD)

    def test_env_override_applies_to_initial(self, tmp_path, monkeypatch):
        monkeypatch.setenv("HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT", "12")
        runner = self._runner(tmp_path)
        assert runner._platform_connect_timeout_secs(
            Platform.TELEGRAM, initial=True
        ) == 12.0

    @pytest.mark.asyncio
    async def test_initial_connect_times_out_at_cap_and_queues_retry(
        self, tmp_path, monkeypatch
    ):
        """A wedged Telegram connect is abandoned at the capped budget and the
        platform lands in the reconnect queue instead of blocking startup."""
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        monkeypatch.delenv("HERMES_GATEWAY_PLATFORM_CONNECT_TIMEOUT", raising=False)

        class _WedgedAdapter(BasePlatformAdapter):
            def __init__(self):
                super().__init__(
                    PlatformConfig(enabled=True, token="***"), Platform.TELEGRAM
                )

            async def connect(self, *, is_reconnect: bool = False) -> bool:
                await asyncio.sleep(3600)
                return True

            async def disconnect(self) -> None:
                self._mark_disconnected()

            async def send(self, chat_id, content, reply_to=None, metadata=None):
                raise NotImplementedError

            async def get_chat_info(self, chat_id):
                return {"id": chat_id}

        config = GatewayConfig(
            platforms={
                Platform.TELEGRAM: PlatformConfig(enabled=True, token="***"),
                Platform.DISCORD: PlatformConfig(enabled=True, token="***"),
            },
            sessions_dir=tmp_path / "sessions",
        )
        runner = GatewayRunner(config)

        # Shrink the capped budget so the test is fast; the assertion is that
        # the INITIAL path (initial=True) is the one that fires, not the 180s
        # reconnect budget.
        import gateway.run as gateway_run

        monkeypatch.setattr(
            gateway_run, "_TELEGRAM_INITIAL_CONNECT_TIMEOUT_SECS_DEFAULT", 0.2
        )

        def _make_adapter(platform, platform_config):
            if platform is Platform.TELEGRAM:
                return _WedgedAdapter()
            return _TimingAdapter(platform, 0.0)

        monkeypatch.setattr(runner, "_create_adapter", _make_adapter)
        monkeypatch.setattr(runner, "_start_secondary_profile_adapters", lambda: 0)

        await asyncio.wait_for(runner.start(), timeout=30)

        # Discord served; Telegram queued for the watcher's full-budget retry.
        assert Platform.DISCORD in runner.adapters
        assert Platform.TELEGRAM not in runner.adapters
        assert Platform.TELEGRAM in runner._failed_platforms