From b0999c82f37759f13eef45714167abdaf5d2981a Mon Sep 17 00:00:00 2001 From: Ben Date: Mon, 8 Jun 2026 15:14:08 +1000 Subject: [PATCH] feat(relay): generic RelayAdapter advertising negotiated capabilities One BasePlatformAdapter subclass that reads its capability profile from a CapabilityDescriptor: MAX_MESSAGE_LENGTH attribute, message_len_fn (table-driven by len_unit: chars=len, utf16=Telegram-style code units), supports_draft_streaming. Implements the four abstract methods (connect/disconnect/send/get_chat_info) by delegating to an injected RelayTransport (full protocol lands in Task 1.2). Adds Platform.RELAY enum member. No per-platform gateway code. Phase 1, Task 1.1 of the gateway-relay plan. --- gateway/config.py | 1 + gateway/relay/adapter.py | 129 ++++++++++++++++++++++ tests/gateway/relay/test_relay_adapter.py | 77 +++++++++++++ 3 files changed, 207 insertions(+) create mode 100644 gateway/relay/adapter.py create mode 100644 tests/gateway/relay/test_relay_adapter.py diff --git a/gateway/config.py b/gateway/config.py index ae149f81d..eb92d833e 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -164,6 +164,7 @@ class Platform(Enum): BLUEBUBBLES = "bluebubbles" QQBOT = "qqbot" YUANBAO = "yuanbao" + RELAY = "relay" # generic relay adapter fronted by the connector (EXPERIMENTAL) @classmethod def _missing_(cls, value): """Accept unknown platform names only for known plugin adapters. diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py new file mode 100644 index 000000000..cbcea2800 --- /dev/null +++ b/gateway/relay/adapter.py @@ -0,0 +1,129 @@ +"""RelayAdapter — one generic gateway adapter fronted by the connector. EXPERIMENTAL. + +A single ``BasePlatformAdapter`` subclass that, at handshake, receives a +``CapabilityDescriptor`` from the connector telling it which platform it is +fronting and which capabilities to advertise to the ``GatewayStreamConsumer``. +It implements the four abstract methods (``connect`` / ``disconnect`` / ``send`` +/ ``get_chat_info``) plus the capability surface (``MAX_MESSAGE_LENGTH``, +``message_len_fn``, ``supports_draft_streaming``) by delegating wire I/O to an +injected transport and reading capabilities off the descriptor. + +There is NO per-platform gateway code: the connector is the only side that knows +"this chat_id maps to a Discord channel, send it via the Discord websocket." +The gateway sees an ordinary ``MessageEvent`` in and calls ``adapter.send`` out. + +EXPERIMENTAL: the transport protocol and descriptor schema may change without a +deprecation cycle until >=2 Class-1 platforms validate them. +""" + +from __future__ import annotations + +import logging +from typing import Any, Callable, Dict, Optional, Protocol, runtime_checkable + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import BasePlatformAdapter, SendResult +from gateway.relay.descriptor import CapabilityDescriptor + +logger = logging.getLogger(__name__) + + +def _utf16_len(text: str) -> int: + """Count UTF-16 code units (Telegram's length unit).""" + return len(text.encode("utf-16-le")) // 2 + + +# Table-driven length-unit selection from the descriptor's ``len_unit``. +_LEN_FNS: Dict[str, Callable[[str], int]] = { + "chars": len, + "utf16": _utf16_len, +} + + +@runtime_checkable +class RelayTransport(Protocol): + """Minimal transport contract the RelayAdapter delegates wire I/O to. + + The full protocol (inbound MessageEvent stream, interrupt channel) is + fleshed out in gateway/relay/transport.py (Task 1.2); the adapter only + needs these outbound + lifecycle calls to satisfy the abstract methods. + """ + + async def connect(self) -> bool: ... + + async def disconnect(self) -> None: ... + + async def send_outbound(self, action: Dict[str, Any]) -> Dict[str, Any]: ... + + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: ... + + +class RelayAdapter(BasePlatformAdapter): + """Generic relay adapter advertising a connector-negotiated capability profile.""" + + def __init__( + self, + config: PlatformConfig, + descriptor: CapabilityDescriptor, + transport: Optional[RelayTransport] = None, + ) -> None: + # The relay adapter fronts many platforms but presents as a single + # logical platform to the runner; Platform.RELAY identifies it. + super().__init__(config, Platform.RELAY) + self.descriptor = descriptor + self._transport = transport + # Capability surface read by stream_consumer (getattr(..., 4096)). + self.MAX_MESSAGE_LENGTH = descriptor.max_message_length + self.supports_code_blocks = descriptor.markdown_dialect not in ("", "plain") + + # ── capability surface (from descriptor) ───────────────────────────── + @property + def message_len_fn(self) -> Callable[[str], int]: + return _LEN_FNS.get(self.descriptor.len_unit, len) + + def supports_draft_streaming( + self, + chat_type: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> bool: + return self.descriptor.supports_draft_streaming + + # ── abstract methods (delegated to the transport) ──────────────────── + async def connect(self) -> bool: + if self._transport is None: + raise RuntimeError("RelayAdapter has no transport configured") + return await self._transport.connect() + + async def disconnect(self) -> None: + if self._transport is not None: + await self._transport.disconnect() + + async def send( + self, + chat_id: str, + content: str, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + if self._transport is None: + return SendResult(success=False, error="no transport") + result = await self._transport.send_outbound( + { + "op": "send", + "chat_id": chat_id, + "content": content, + "reply_to": reply_to, + "metadata": metadata or {}, + } + ) + return SendResult( + success=bool(result.get("success")), + message_id=result.get("message_id"), + error=result.get("error"), + ) + + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: + # Proxied to the connector (it owns the platform connection / cache). + if self._transport is None: + return {"name": chat_id, "type": "dm"} + return await self._transport.get_chat_info(chat_id) diff --git a/tests/gateway/relay/test_relay_adapter.py b/tests/gateway/relay/test_relay_adapter.py new file mode 100644 index 000000000..64d6aab2f --- /dev/null +++ b/tests/gateway/relay/test_relay_adapter.py @@ -0,0 +1,77 @@ +"""RelayAdapter capability-advertisement tests (relay Phase 1, Task 1.1).""" + +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.relay.adapter import RelayAdapter +from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor + + +def make_desc(**kw) -> CapabilityDescriptor: + base = dict( + contract_version=CONTRACT_VERSION, + platform="telegram", + label="Telegram", + max_message_length=4096, + supports_draft_streaming=False, + supports_edit=True, + supports_threads=True, + markdown_dialect="markdown_v2", + len_unit="utf16", + emoji="\u2708\ufe0f", + platform_hint="", + pii_safe=False, + ) + base.update(kw) + return CapabilityDescriptor(**base) + + +def _adapter(**desc_kw) -> RelayAdapter: + return RelayAdapter(PlatformConfig(), make_desc(**desc_kw)) + + +def test_relay_platform_member_exists(): + assert Platform("relay") is Platform.RELAY + + +def test_advertises_descriptor_max_length(): + a = _adapter(max_message_length=2000) + assert a.MAX_MESSAGE_LENGTH == 2000 + + +def test_supports_draft_streaming_follows_descriptor(): + assert _adapter(supports_draft_streaming=False).supports_draft_streaming() is False + assert _adapter(supports_draft_streaming=True).supports_draft_streaming() is True + + +def test_len_fn_utf16_counts_code_units(): + a = _adapter(len_unit="utf16") + # An astral-plane emoji is two UTF-16 code units. + assert a.message_len_fn("\U0001f600") == 2 + + +def test_len_fn_chars_uses_builtin_len(): + a = _adapter(len_unit="chars") + assert a.message_len_fn("\U0001f600") == 1 + + +def test_is_a_base_platform_adapter(): + # stream_consumer's isinstance(adapter, BasePlatformAdapter) guard must pass. + from gateway.platforms.base import BasePlatformAdapter + + assert isinstance(_adapter(), BasePlatformAdapter) + + +@pytest.mark.asyncio +async def test_connect_without_transport_raises(): + a = _adapter() + with pytest.raises(RuntimeError, match="no transport"): + await a.connect() + + +@pytest.mark.asyncio +async def test_send_without_transport_returns_failure(): + a = _adapter() + result = await a.send("chat1", "hello") + assert result.success is False + assert result.error == "no transport"