mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
fix(relay): per-platform capability descriptors for multi-platform gateways (#70717)
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 — and the transport's read loop OVERWROTE it on every descriptor frame (last-writer-wins). A Discord chat on a gateway whose applied descriptor was Telegram's inherited the 4,096-char cap and over-sent into Discord's 2,000-char API 400 (observed live: 2,543/2,641-char replies silently lost while inbound kept working). - ws_transport: accumulate one descriptor per platform in _descriptors_by_platform (exposed via descriptor_for_platform); the FIRST descriptor of a connection generation stays the session default instead of last-writer-wins; the map resets on re-dial. - BasePlatformAdapter: new max_message_length_for_chat / message_len_fn_for_chat hooks defaulting to the scalar surface (native single-platform adapters unchanged). - RelayAdapter: overrides resolve the chat's platform from _platform_by_chat (the same map per-frame egress uses) and look up that platform's negotiated descriptor; falls back to the scalar for unknown chats/transports. - stream_consumer (streaming budget, _raw_message_limit, fallback-continuation chunking) + run.py tool-progress limit now resolve per-chat. Tests: tests/gateway/relay/test_relay_per_platform_caps.py (7) — verified fail-without/pass-with (all 7 fail with the fix stashed). Relay + stream consumer suites green (213 + 223).
This commit is contained in:
Ben Barclay
GitHub
parent
91d69c4ca3
commit
96996a55bf
6 changed files with 379 additions and 6 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
252
tests/gateway/relay/test_relay_per_platform_caps.py
Normal file
252
tests/gateway/relay/test_relay_per_platform_caps.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Add a link
Reference in a new issue