hermes-agent/tests/gateway/relay/test_relay_per_platform_caps.py

"""Per-platform capability descriptors on the relay (multi-platform Phase 1.5).

The bug class: one relay adapter fronts N platforms on one WS, but the
capability surface (``MAX_MESSAGE_LENGTH`` / ``message_len_fn``) was a SCALAR
from whichever descriptor resolved the handshake — so a Discord chat on a
gateway whose primary identity was Telegram inherited Telegram's 4,096-char
cap and over-sent into Discord's 2,000-char API 400 (observed live: 2,543 and
2,641-char sends rejected).

Covers:
  - the transport accumulating one descriptor per platform (first = session
    default, later frames must NOT overwrite it),
  - the map resetting on a re-dial,
  - RelayAdapter.max_message_length_for_chat / message_len_fn_for_chat
    resolving from the chat's inbound platform,
  - fallback to the scalar descriptor for unknown chats / transports without
    the map,
  - the stream consumer's _raw_message_limit honoring the per-chat cap.
"""

from __future__ import annotations

import asyncio
import json
from typing import Any, Dict, List, Optional

import pytest

from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import MessageEvent, MessageType
from gateway.relay.adapter import RelayAdapter
from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
from gateway.session import SessionSource

from tests.gateway.relay.stub_connector import StubConnector


def _descriptor(platform: str, max_len: int, len_unit: str = "chars") -> CapabilityDescriptor:
    return CapabilityDescriptor(
        contract_version=CONTRACT_VERSION,
        platform=platform,
        label=platform.title(),
        max_message_length=max_len,
        supports_draft_streaming=False,
        supports_edit=True,
        supports_threads=False,
        markdown_dialect="plain",
        len_unit=len_unit,
    )


DISCORD = _descriptor("discord", 2000)
TELEGRAM = _descriptor("telegram", 4096, len_unit="utf16")


class MultiDescriptorStub(StubConnector):
    """StubConnector extended with the per-platform descriptor map."""

    def __init__(self, primary: CapabilityDescriptor, *others: CapabilityDescriptor) -> None:
        super().__init__(primary)
        self._by_platform = {d.platform: d for d in (primary, *others)}

    def descriptor_for_platform(self, platform: str) -> Optional[CapabilityDescriptor]:
        return self._by_platform.get(platform)


async def _push(stub: StubConnector, platform: Platform, chat_id: str) -> None:
    await stub.push_inbound(
        MessageEvent(
            text="hi",
            message_type=MessageType.TEXT,
            source=SessionSource(
                platform=platform, chat_id=chat_id, chat_type="dm", user_id="u-1"
            ),
        )
    )


# ───────────────────── transport descriptor accumulation ─────────────────────


def _make_transport():
    from gateway.relay.ws_transport import WebSocketRelayTransport

    return WebSocketRelayTransport(
        "wss://connector.example/relay",
        "telegram",
        "bot-9",
        identities=[("telegram", "bot-9"), ("discord", "app-1")],
    )


@pytest.mark.asyncio
async def test_transport_descriptor_map_resets_on_redial(monkeypatch):
    """A re-dial starts a fresh handshake generation: stale per-platform
    descriptors must not survive into the new connection."""
    t = _make_transport()
    loop = asyncio.get_running_loop()
    t._descriptor_ready = loop.create_future()
    await t._handle_frame(json.dumps({"type": "descriptor", "descriptor": DISCORD.__dict__}))
    assert t.descriptor_for_platform("discord") is not None

    # Simulate _dial_and_start's reset preamble without a real socket.
    class _FakeWs:
        async def close(self):  # pragma: no cover - not called
            pass

    sent: List[str] = []

    async def _fake_connect(url, **kwargs):
        return _FakeWs()

    async def _fake_send(payload):
        sent.append(payload)

    import gateway.relay.ws_transport as wst

    monkeypatch.setattr(wst, "websockets", type("M", (), {"connect": staticmethod(_fake_connect)}))
    monkeypatch.setattr(t, "_send", _fake_send)
    monkeypatch.setattr(
        t, "_read_loop", lambda: asyncio.sleep(0)
    )  # substitute a no-op coroutine factory
    await t._dial_and_start()

    assert t.descriptor_for_platform("discord") is None
    assert t._descriptor is None


# ───────────────────── adapter per-chat capability surface ─────────────────────


@pytest.mark.asyncio
async def test_adapter_resolves_per_chat_limits_from_inbound_platform():
    """The live bug shape: primary identity Telegram (4096), a Discord chat on
    the same adapter must get Discord's 2000 — not Telegram's scalar."""
    stub = MultiDescriptorStub(TELEGRAM, DISCORD)
    stub._identities = [("telegram", "bot-9"), ("discord", "app-1")]
    adapter = RelayAdapter(PlatformConfig(), TELEGRAM, transport=stub)
    await adapter.connect()

    await _push(stub, Platform.DISCORD, "dc-1")
    await _push(stub, Platform.TELEGRAM, "tg-1")

    # Scalar surface still the primary's (back-compat).
    assert adapter.MAX_MESSAGE_LENGTH == 4096
    # Per-chat: each chat resolves its own platform's cap.
    assert adapter.max_message_length_for_chat("dc-1") == 2000
    assert adapter.max_message_length_for_chat("tg-1") == 4096
    # Length unit follows the chat too: Telegram utf16, Discord codepoints.
    surrogate = "\U0001f600"  # 2 UTF-16 units, 1 codepoint
    assert adapter.message_len_fn_for_chat("tg-1")(surrogate) == 2
    assert adapter.message_len_fn_for_chat("dc-1")(surrogate) == 1


# ───────────────────── stream consumer integration ─────────────────────