Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/relay-connector-contract.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
40 changes: 40 additions & 0 deletions gateway/relay/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
83 changes: 83 additions & 0 deletions tests/gateway/relay/test_relay_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -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" ─────


Expand Down
36 changes: 36 additions & 0 deletions tests/gateway/relay/test_relay_roundtrip.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading