feat(relay): egress typing indicators through the connector (op="typing") (#69721)

The base class spawns _keep_typing for every adapter — a 2s refresh loop
that runs for the whole turn, including while the stream consumer sends
the response — but RelayAdapter inherited BasePlatformAdapter's no-op
send_typing. The loop ran every turn and emitted nothing, so relay-fronted
chats (hosted Discord/Telegram/Signal/Slack) never showed 'is typing…'.

The rest of the pipe already existed on both sides: OutboundOp "typing"
is in the wire contract (contract_version 1), and every connector-side
sender implements it (Discord POST /channels/{id}/typing, Telegram
sendChatAction, Signal sendTyping, Slack assistant status).

This bridges the tick onto the existing outbound frame, mirroring send():
- _with_scope re-attaches the tenant discriminator (metadata.scope_id, or
  user_id for DMs) — the connector's routedEgressGuard wraps ALL ops, so
  an undiscriminated typing frame is declined like a bare send would be.
- the Phase 1.5 per-frame platform tag routes typing through the platform
  the chat lives on for multi-platform gateways.

Best-effort by design: transport errors are swallowed (typing is cosmetic;
the next tick retries), no transport is a silent no-op, and stop_typing
stays the base no-op (platform indicators self-expire). Additive within
contract_version 1 — an older connector returns an unsupported-op result
we ignore. Contract doc §4 updated (typing now carries metadata?).

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

gateway/relay/adapter.py

@@ -513,6 +513,46 @@ class RelayAdapter(BasePlatformAdapter):
             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:

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" ─────
 
 

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