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

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

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)

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"