diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 5d513458941..e55c91b4dd1 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2693,6 +2693,30 @@ class BasePlatformAdapter(ABC): """ return len + def max_message_length_for_chat(self, chat_id: str) -> int: + """Per-chat max message length, in ``message_len_fn_for_chat`` units. + + Default: the adapter-scalar ``MAX_MESSAGE_LENGTH`` (4096 when absent) — + for a native adapter every chat lives on the same platform so the + scalar is already correct. The relay adapter overrides this: one relay + adapter fronts N platforms with different caps (Discord 2000 vs + Telegram 4096 vs Slack 39000), and the right cap depends on which + platform the chat's inbound arrived from. + """ + try: + return int(getattr(self, "MAX_MESSAGE_LENGTH", 4096) or 4096) + except (TypeError, ValueError): + return 4096 + + def message_len_fn_for_chat(self, chat_id: str) -> Callable[[str], int]: + """Per-chat length function (companion to max_message_length_for_chat). + + Default: the adapter-wide ``message_len_fn``. The relay adapter + overrides it so a Telegram-fronted chat measures UTF-16 units while a + Discord-fronted chat on the same adapter measures codepoints. + """ + return self.message_len_fn + @property def enforces_own_access_policy(self) -> bool: """Whether this adapter gates inbound access before dispatch. diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index b149539cee5..281602922fa 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -20,7 +20,7 @@ from __future__ import annotations import asyncio import logging -from typing import Any, Callable, Dict, Optional +from typing import Any, Callable, Dict, Optional, cast from gateway.config import Platform, PlatformConfig from gateway.platforms.base import BasePlatformAdapter, MessageEvent, SendResult @@ -123,6 +123,42 @@ class RelayAdapter(BasePlatformAdapter): def message_len_fn(self) -> Callable[[str], int]: return _LEN_FNS.get(self.descriptor.len_unit, len) + # ── per-chat capability resolution (Phase 1.5 multi-platform) ───────── + def _descriptor_for_chat(self, chat_id: str) -> CapabilityDescriptor: + """The capability descriptor governing a specific chat. + + A multi-platform gateway fronts N platforms on ONE adapter, but the + scalar `descriptor`/`MAX_MESSAGE_LENGTH` surface can only carry one + platform's profile (the primary identity's). Platform caps genuinely + differ — Discord 2000 / Telegram 4096 / Slack 39000 — so applying the + primary's cap to every chat either fragments needlessly (small primary) + or over-sends into a platform 400 (large primary; the live bug: 2,543 + and 2,641-char sends rejected by Discord). Resolve the chat's platform + from what we saw inbound (`_platform_by_chat`, the same map per-frame + egress uses) and look up that platform's negotiated descriptor on the + transport. Falls back to the scalar descriptor when the chat's platform + is unknown (never saw inbound) or the transport predates the map. + """ + platform = self._platform_by_chat.get(str(chat_id)) + if platform and self._transport is not None: + resolve = getattr(self._transport, "descriptor_for_platform", None) + if callable(resolve): + try: + per_platform = cast( + Optional[CapabilityDescriptor], resolve(platform) + ) + except Exception: # noqa: BLE001 - capability lookup must never break a send + per_platform = None + if per_platform is not None: + return per_platform + return self.descriptor + + def max_message_length_for_chat(self, chat_id: str) -> int: + return self._descriptor_for_chat(chat_id).max_message_length + + def message_len_fn_for_chat(self, chat_id: str) -> Callable[[str], int]: + return _LEN_FNS.get(self._descriptor_for_chat(chat_id).len_unit, len) + def supports_draft_streaming( self, chat_type: Optional[str] = None, diff --git a/gateway/relay/ws_transport.py b/gateway/relay/ws_transport.py index 62fe683c8fc..f19b39405c0 100644 --- a/gateway/relay/ws_transport.py +++ b/gateway/relay/ws_transport.py @@ -404,6 +404,11 @@ class WebSocketRelayTransport: self._reader: Optional[asyncio.Task[None]] = None self._inbound: Optional[InboundHandler] = None self._descriptor: Optional[CapabilityDescriptor] = None + # Phase 1.5 multi-platform: descriptors keyed by the underlying platform + # (one per hello'd identity). `_descriptor` above stays the FIRST + # (primary-identity) descriptor for back-compat; this map is the + # per-platform capability surface read via `descriptor_for_platform`. + self._descriptors_by_platform: Dict[str, CapabilityDescriptor] = {} self._descriptor_ready: asyncio.Future[CapabilityDescriptor] | None = None # requestId -> future awaiting the matching outbound_result. self._pending: Dict[str, asyncio.Future[Dict[str, Any]]] = {} @@ -433,8 +438,10 @@ class WebSocketRelayTransport: loop = asyncio.get_running_loop() self._descriptor_ready = loop.create_future() # A fresh handshake is coming; clear any stale descriptor so handshake() - # awaits the new one (matters on a re-dial). + # awaits the new one (matters on a re-dial). The per-platform map resets + # with it — a reconnected connector re-sends one descriptor per hello. self._descriptor = None + self._descriptors_by_platform = {} # scale-to-zero (D12): a successful (re-)dial ends any dormant state — we # are live again, so a subsequent UNEXPECTED close should reconnect on the # normal fast backoff, not the dormant cadence. @@ -520,6 +527,18 @@ class WebSocketRelayTransport: raise RuntimeError("handshake() called before connect()") return await asyncio.wait_for(self._descriptor_ready, timeout=self._connect_timeout_s) + def descriptor_for_platform(self, platform: str) -> Optional[CapabilityDescriptor]: + """The negotiated descriptor for one fronted platform, or None. + + Phase 1.5 multi-platform: the connector replies one descriptor per + hello'd identity; they accumulate here keyed by the descriptor's own + ``platform`` field. Callers (RelayAdapter) use this to resolve PER-CHAT + capabilities — e.g. Discord's 2000-char max_message_length vs + Telegram's 4096 — instead of applying the primary identity's scalar + descriptor to every platform this gateway fronts. + """ + return self._descriptors_by_platform.get(platform) + @property def auth_revoked(self) -> bool: """True once the connector closed the socket with 4401 AFTER a prior @@ -795,7 +814,19 @@ class WebSocketRelayTransport: ftype = frame.get("type") if ftype == "descriptor": descriptor = CapabilityDescriptor.from_json(json.dumps(frame.get("descriptor", {}))) - self._descriptor = descriptor + # Phase 1.5 multi-platform: one descriptor frame arrives per hello'd + # identity. Accumulate them keyed by the descriptor's own platform so + # the adapter can resolve PER-CHAT capabilities (e.g. Discord's 2000 + # vs Telegram's 4096 max_message_length) instead of collapsing N + # platforms onto whichever descriptor arrived last. + if descriptor.platform: + self._descriptors_by_platform[descriptor.platform] = descriptor + # The FIRST descriptor of this connection generation is the session + # default (the primary identity's) — later arrivals must NOT + # overwrite it, or the scalar capability surface silently becomes + # last-writer-wins across platforms. + if self._descriptor is None: + self._descriptor = descriptor # Phase 7 Unit 7d-B: a received descriptor means the WS upgrade auth # passed and the connector accepted us — record that we've handshaked # at least once, so a LATER 4401 close is read as a revocation diff --git a/gateway/run.py b/gateway/run.py index cb5aa02490e..37dfd9d92cd 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -21023,6 +21023,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _raw_progress_limit = int(getattr(adapter, "MAX_MESSAGE_LENGTH", 4000) or 4000) except Exception: _raw_progress_limit = 4000 + # Per-chat resolution (relay adapter fronting N platforms): the cap + # and length unit follow the chat's underlying platform. Native + # adapters return their scalar/property unchanged. + if isinstance(adapter, BasePlatformAdapter): + try: + _raw_progress_limit = int( + adapter.max_message_length_for_chat(source.chat_id) or 4000 + ) + _progress_len_fn = adapter.message_len_fn_for_chat(source.chat_id) + except Exception: + pass # Leave a little room for platform quirks / formatting. For tiny # test adapters keep the limit usable instead of clamping to 500+. _PROGRESS_TEXT_LIMIT = max( diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index e60e9117031..7d3e31cf153 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -675,10 +675,13 @@ class GatewayStreamConsumer: # Platform message length limit — leave room for cursor + formatting. # Use the adapter's length function (e.g. utf16_len for Telegram) so # overflow detection matches what the platform actually enforces. + # Both resolve PER-CHAT (max_message_length_for_chat): a relay adapter + # fronting N platforms has different caps per chat (Discord 2000 vs + # Telegram 4096); native adapters return their scalar unchanged. # Gate on isinstance(BasePlatformAdapter) so test MagicMocks (whose # auto-attributes return mock objects, not callables) fall back to len. _len_fn: "Callable[[str], int]" = ( - self.adapter.message_len_fn + self.adapter.message_len_fn_for_chat(self.chat_id) if isinstance(self.adapter, _BasePlatformAdapter) else len ) @@ -1359,6 +1362,15 @@ class GatewayStreamConsumer: if isinstance(self.adapter, _BasePlatformAdapter) else len ) + # Per-chat resolution (relay adapter fronting N platforms): the cap and + # length unit follow the chat's underlying platform, not the adapter + # scalar. Native adapters return their scalar/property unchanged. + if isinstance(self.adapter, _BasePlatformAdapter): + try: + raw_limit = self.adapter.max_message_length_for_chat(self.chat_id) + _len_fn = self.adapter.message_len_fn_for_chat(self.chat_id) + except Exception as e: + logger.debug("per-chat limit resolution failed: %s", e) safe_limit = max(500, raw_limit - 100) chunks = self._split_text_chunks(continuation, safe_limit, len_fn=_len_fn) @@ -1744,8 +1756,11 @@ class GatewayStreamConsumer: """Per-message length budget (in the adapter's ``message_len_fn`` units) before the consumer splits an overflowing reply. - Adapters with a richer send/draft path (e.g. Telegram rich messages) - can raise this above ``MAX_MESSAGE_LENGTH`` via + Resolved PER-CHAT via ``max_message_length_for_chat`` — a relay adapter + fronting N platforms has a different cap per chat (Discord 2000 vs + Telegram 4096 vs Slack 39000); native adapters return their scalar + ``MAX_MESSAGE_LENGTH`` unchanged. Adapters with a richer send/draft + path (e.g. Telegram rich messages) can raise this above the base via ``streaming_overflow_limit`` so a reply that fits one rich message isn't fragmented at the legacy edit limit. Falls back to ``MAX_MESSAGE_LENGTH`` (4096 default) for everyone else. @@ -1754,6 +1769,10 @@ class GatewayStreamConsumer: # isinstance gate: MagicMock adapters return mock objects (truthy, not # ints) for arbitrary attribute access — keep them on the base limit. if isinstance(self.adapter, _BasePlatformAdapter): + try: + base = self.adapter.max_message_length_for_chat(self.chat_id) + except Exception as e: + logger.debug("max_message_length_for_chat failed: %s", e) try: cap = self.adapter.streaming_overflow_limit() except Exception as e: diff --git a/tests/gateway/relay/test_relay_per_platform_caps.py b/tests/gateway/relay/test_relay_per_platform_caps.py new file mode 100644 index 00000000000..a27c3bfbe8a --- /dev/null +++ b/tests/gateway/relay/test_relay_per_platform_caps.py @@ -0,0 +1,252 @@ +"""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_accumulates_descriptors_first_wins_as_default(): + """One descriptor frame per hello: the map holds each platform's, and the + scalar `_descriptor` (the handshake result / session default) stays the + FIRST one — the regression was last-writer-wins across platforms.""" + t = _make_transport() + loop = asyncio.get_running_loop() + t._descriptor_ready = loop.create_future() + + frame = {"type": "descriptor", "descriptor": TELEGRAM.__dict__} + await t._handle_frame(json.dumps(frame)) + frame2 = {"type": "descriptor", "descriptor": DISCORD.__dict__} + await t._handle_frame(json.dumps(frame2)) + + # Per-platform map has both. + assert t.descriptor_for_platform("telegram").max_message_length == 4096 + assert t.descriptor_for_platform("discord").max_message_length == 2000 + assert t.descriptor_for_platform("slack") is None + # The session default is the FIRST (primary identity) — NOT overwritten. + assert t._descriptor.platform == "telegram" + assert (await t.handshake()).platform == "telegram" + + +@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 + + +@pytest.mark.asyncio +async def test_adapter_unknown_chat_falls_back_to_scalar_descriptor(): + stub = MultiDescriptorStub(TELEGRAM, DISCORD) + adapter = RelayAdapter(PlatformConfig(), TELEGRAM, transport=stub) + await adapter.connect() + # Never saw inbound for this chat — platform unknown -> scalar descriptor. + assert adapter.max_message_length_for_chat("never-seen") == 4096 + + +@pytest.mark.asyncio +async def test_adapter_transport_without_map_falls_back_to_scalar(): + """A plain StubConnector (no descriptor_for_platform) — e.g. an older or + test transport — must keep the scalar behavior, not raise.""" + stub = StubConnector(TELEGRAM) + adapter = RelayAdapter(PlatformConfig(), TELEGRAM, transport=stub) + await adapter.connect() + await _push(stub, Platform.DISCORD, "dc-1") + assert adapter.max_message_length_for_chat("dc-1") == 4096 + + +def test_native_adapter_defaults_scalar(): + """BasePlatformAdapter's default per-chat hooks mirror the scalar surface + (native adapters are single-platform; nothing changes for them).""" + from gateway.platforms.base import BasePlatformAdapter + + class _Native(BasePlatformAdapter): + MAX_MESSAGE_LENGTH = 1234 + + def __init__(self): # bypass Base __init__ plumbing + pass + + async def connect(self, *, is_reconnect: bool = False) -> bool: # pragma: no cover + return True + + async def disconnect(self) -> None: # pragma: no cover + pass + + async def send(self, chat_id, content, reply_to=None, metadata=None): # pragma: no cover + raise NotImplementedError + + async def get_chat_info(self, chat_id): # pragma: no cover + return {} + + a = _Native() + assert a.max_message_length_for_chat("any") == 1234 + assert a.message_len_fn_for_chat("any") is a.message_len_fn + + +# ───────────────────── stream consumer integration ───────────────────── + + +@pytest.mark.asyncio +async def test_stream_consumer_raw_limit_uses_per_chat_cap(): + """_raw_message_limit resolves the CHAT's platform cap on a relay adapter: + a Discord chat splits at 2000 even when the scalar descriptor says 4096. + Without the fix this returned 4096 and a 2,543-char reply reached Discord + whole -> HTTP 400.""" + from gateway.stream_consumer import GatewayStreamConsumer + + 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") + + dc = GatewayStreamConsumer.__new__(GatewayStreamConsumer) + dc.adapter = adapter + dc.chat_id = "dc-1" + assert dc._raw_message_limit() == 2000 + + tg = GatewayStreamConsumer.__new__(GatewayStreamConsumer) + tg.adapter = adapter + tg.chat_id = "tg-1" + assert tg._raw_message_limit() == 4096