diff --git a/docs/relay-connector-contract.md b/docs/relay-connector-contract.md index 1c9c01ed1e10..c7a5f931582b 100644 --- a/docs/relay-connector-contract.md +++ b/docs/relay-connector-contract.md @@ -391,7 +391,7 @@ The gateway calls the transport with action dicts. Source of truth: | --- | --- | --- | | `send` | `chat_id`, `content`, `reply_to?`, `metadata?` | `{success: bool, message_id?, error?}` | | `edit` | `chat_id`, `message_id`, `content`, `metadata?` | `{success: bool, error?}` | -| `typing` | `chat_id` | `{success: bool}` | +| `typing` | `chat_id`, `metadata?` | `{success: bool}` | | `follow_up` | `session_key`, `kind`, `content`, `metadata?` | `{success: bool, message_id?, error?}` | `get_chat_info(chat_id)` is a separate proxied call returning at least diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index cf0646d1b239..f58d7e2c8a52 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -513,6 +513,46 @@ async def send( error=result.get("error"), ) + async def send_typing(self, chat_id: str, metadata=None) -> None: + """Egress a typing indicator through the connector. + + The base class spawns ``_keep_typing`` for every adapter (a 2s refresh + loop for the life of the turn), but the relay adapter inherited the + base no-op ``send_typing`` — so hosted/relay chats never showed + "is typing…" even though the wire contract (``OutboundOp "typing"``) + and every connector-side sender (Discord ``POST /channels/{id}/typing``, + Telegram ``sendChatAction``, Signal ``sendTyping``, Slack assistant + status) already implement it. This bridges the loop's tick onto the + existing outbound frame. + + Two details are load-bearing, mirroring ``send()``: + - ``_with_scope``: the connector's egress guard wraps ALL ops + (routedEgressGuard), so a typing frame without a resolvable tenant + discriminator (metadata.scope_id, or user_id for DMs) is declined + exactly like a bare send would be. + - the per-frame ``platform`` tag (Phase 1.5): a multi-platform + gateway must egress typing through the platform the chat lives on. + + Best-effort: failures are swallowed (``_keep_typing`` already treats + send_typing errors as non-fatal, and an older connector that rejects + the op just returns an unsuccessful result we ignore). Each call is + one-shot — Discord/Telegram indicators self-expire, so there is no + state to clean up and the base no-op ``stop_typing`` stays correct. + """ + if self._transport is None: + return + try: + await self._transport.send_outbound( + { + "op": "typing", + "chat_id": chat_id, + "metadata": self._with_scope(chat_id, metadata), + }, + platform=self._platform_by_chat.get(str(chat_id)), + ) + except Exception: # noqa: BLE001 - typing is cosmetic, never breaks a turn + logger.debug("relay send_typing failed for %s", chat_id, exc_info=True) + 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: diff --git a/tests/gateway/relay/test_relay_adapter.py b/tests/gateway/relay/test_relay_adapter.py index dba0d0edfa23..707f0d8891bf 100644 --- a/tests/gateway/relay/test_relay_adapter.py +++ b/tests/gateway/relay/test_relay_adapter.py @@ -287,6 +287,89 @@ async def test_scoped_reply_without_inbound_author_carries_scope_only(): assert "user_id" not in t.sent["metadata"] +# ── typing indicator over the relay (op="typing") ──────────────────────────── + + +@pytest.mark.asyncio +async def test_send_typing_emits_typing_op_with_scope(): + """The base class's ``_keep_typing`` refresh loop calls ``send_typing`` every + ~2s for the life of a turn, but RelayAdapter inherited the base no-op — so + hosted/relay Discord chats never showed "is typing…" even though the wire + contract and the connector's senders already implement the ``typing`` op. + The adapter must bridge the tick onto an outbound frame carrying the same + tenant discriminator a send would (the connector's routedEgressGuard wraps + ALL ops; an undiscriminated typing frame is declined).""" + t = _CaptureTransport() + a = RelayAdapter(PlatformConfig(), make_desc(platform="discord"), transport=t) + a._capture_scope(_make_event(chat_id="chan-1", scope_id="scope-9")) + + await a.send_typing("chan-1") + + assert t.sent["op"] == "typing" + assert t.sent["chat_id"] == "chan-1" + assert t.sent["metadata"].get("scope_id") == "scope-9" + + +@pytest.mark.asyncio +async def test_send_typing_dm_carries_user_id(): + """A DM typing frame has no scope, so it must carry the authentic author + user_id (the connector resolves the tenant via the recipient's author + binding, same as a DM send).""" + t = _CaptureTransport() + a = RelayAdapter(PlatformConfig(), make_desc(platform="discord"), transport=t) + a._capture_scope(_make_dm_event(chat_id="dm-1", user_id="user-42")) + + await a.send_typing("dm-1") + + assert t.sent["op"] == "typing" + assert t.sent["metadata"].get("user_id") == "user-42" + assert "scope_id" not in t.sent["metadata"] + + +@pytest.mark.asyncio +async def test_send_typing_tags_egress_platform(): + """Phase 1.5: a multi-platform gateway must egress typing through the + platform the chat lives on, exactly like send() — the underlying platform + learned from the inbound event tags the frame.""" + from gateway.platforms.base import MessageEvent, MessageType + from gateway.session import SessionSource + + t = _CaptureTransport() + a = RelayAdapter(PlatformConfig(), make_desc(platform="discord"), transport=t) + src = SessionSource( + platform=Platform.DISCORD, + chat_id="chan-2", + chat_type="channel", + scope_id="scope-1", + ) + a._capture_scope(MessageEvent(text="hi", source=src, message_type=MessageType.TEXT)) + + await a.send_typing("chan-2") + + assert t.sent_platform == "discord" + + +@pytest.mark.asyncio +async def test_send_typing_without_transport_is_noop(): + """No transport ⇒ silent no-op (typing is cosmetic; must never raise into + the _keep_typing loop).""" + a = _adapter() + await a.send_typing("chan-1") # must not raise + + +@pytest.mark.asyncio +async def test_send_typing_swallows_transport_errors(): + """A transport failure (WS down mid-turn) must not propagate — _keep_typing + treats errors as non-fatal, and the next 2s tick retries anyway.""" + + class _FailingTransport(_CaptureTransport): + async def send_outbound(self, action, *, platform=None): + raise RuntimeError("ws down") + + a = RelayAdapter(PlatformConfig(), make_desc(platform="discord"), transport=_FailingTransport()) + await a.send_typing("chan-1") # must not raise + + # ── Phase 7 Unit 7d-B: terminal auth revocation → clean "relay disabled" ───── diff --git a/tests/gateway/relay/test_relay_roundtrip.py b/tests/gateway/relay/test_relay_roundtrip.py index 7509ba6c1037..c80571786aff 100644 --- a/tests/gateway/relay/test_relay_roundtrip.py +++ b/tests/gateway/relay/test_relay_roundtrip.py @@ -117,6 +117,42 @@ async def test_get_chat_info_proxied_to_connector(wired): assert info == {"name": "general", "type": "group"} +@pytest.mark.asyncio +async def test_keep_typing_loop_emits_typing_frames_with_scope(wired): + """E2E through the REAL base-class refresh loop: the same ``_keep_typing`` + task ``_process_message_background`` spawns for every turn must produce + ``op="typing"`` frames on the relay transport, carrying the tenant + discriminator captured from the inbound event (the connector's egress + guard declines undiscriminated frames). Regression: RelayAdapter inherited + the base no-op send_typing, so this loop ran all turn and emitted nothing — + no \"is typing…\" on any relay-fronted platform.""" + import asyncio + + adapter, stub = wired + await adapter.connect() + # Inbound captures chan1 -> guildA (scope) exactly as a real turn would. + adapter._capture_scope(_discord_event("guildA", "chan1", "userX", "hello")) + + stop = asyncio.Event() + task = asyncio.create_task( + adapter._keep_typing("chan1", interval=0.05, stop_event=stop) + ) + await asyncio.sleep(0.12) # >= 2 ticks + stop.set() + await asyncio.wait_for(task, timeout=1.0) + + typing_frames = [f for f in stub.sent if f.get("op") == "typing"] + assert len(typing_frames) >= 2, f"expected repeated typing frames, got {stub.sent}" + for frame in typing_frames: + assert frame["chat_id"] == "chan1" + assert frame["metadata"].get("scope_id") == "guildA" + # Phase 1.5: each frame is tagged with the underlying platform for egress. + typing_platforms = [ + p for f, p in zip(stub.sent, stub.sent_platforms) if f.get("op") == "typing" + ] + assert all(p == "discord" for p in typing_platforms) + + async def _async_capture(sink, event): sink.append(event) return None