feat(relay): consume channel context from the connector (#64649)

Phase 3 of relay-channel-context (gateway/agent side, single PR). The
connector (gateway-gateway #122/#123/#124) now attaches read-only
surrounding channel/group context to an addressed relay turn; this wires
the gateway to consume it.

- descriptor.py: additive optional supports_context (default False) on
  CapabilityDescriptor. from_json already filters unknown keys, so this is
  back-compat both directions within contract_version 1.
- ws_transport.py: _event_from_wire maps the connector's read-only
  context[] array into the EXISTING MessageEvent.channel_context field via
  a new _render_relay_context() helper — reusing the same read-only
  injection path history-backfill uses (run.py prepends channel_context
  ahead of the trigger message). Never raises; absent/empty/malformed ->
  channel_context unset (byte-identical to today).
- docs/relay-connector-contract.md: document supports_context in the §2
  descriptor table (fixes the contract-doc conformance test) + the
  context/context_error inbound fields in §3.
- tests: descriptor default/round-trip/forward-compat; _render_relay_context
  rendering + malformed-safe; _event_from_wire context->channel_context
  mapping + the read-only invariant (trigger text untouched).

RELAY-ONLY: only gateway/relay/* + the shared MessageEvent consumption via
its existing channel_context field. No native adapter touched.

docs/relay-connector-contract.md

@@ -51,6 +51,7 @@ JSON object. Source of truth: `gateway/relay/descriptor.py`.
 | `emoji` | string | no | Display emoji (default 🔌). |
 | `platform_hint` | string | no | System-prompt platform hint. |
 | `pii_safe` | bool | no | Redact PII in session descriptions. |
+| `supports_context` | bool | no | Whether the connector can supply surrounding channel/group **context** for an addressed turn on this platform (Model A on-demand history fetch — Discord/Slack/Matrix; Model B passive buffer — Telegram/Signal/WhatsApp). Default false ⇒ no `context` is attached to inbound events. See §3. |
 
 Most fields are a projection of the gateway's existing `PlatformEntry`; the
 runtime-only fields (`len_unit`, `supports_*`, `markdown_dialect`) come from the
@@ -95,6 +96,24 @@ Frames (connector → gateway, over the WS):
 - `{"type":"interrupt_inbound", "session_key", "chat_id"}` (§5)
 - `{"type":"passthrough_forward", "forward": <PassthroughForward>, "bufferId"?}` (§5.1)
 
+**Channel context on inbound (design relay-channel-context).** When the source
+platform's descriptor advertised `supports_context` (§2) and the chat is
+multi-party (`chat_type` ∈ group/channel/thread/forum, never `dm`), the
+connector MAY attach two optional, additive fields to the inbound `MessageEvent`:
+
+- `context`: an array of read-only surrounding messages (same channel, oldest→
+  newest) — nearby non-addressed chatter the connector fetched (Model A) or
+  buffered (Model B). REFERENCE ONLY: it never triggers the agent (the trigger
+  decision was already made connector-side on the addressed event alone). The
+  gateway renders it into `MessageEvent.channel_context` (the same read-only
+  injection path history-backfill uses).
+- `context_error`: bool, true when the platform is context-capable but the
+  fetch/buffer failed and the connector fail-opened to an empty `context`
+  (observability marker; surfaced connector-side via the delivery span).
+
+Both absent ⇒ byte-identical to today. A connector that never sends them, or a
+`dm`, or a no-context platform, yields no `channel_context`.
+
 `PassthroughForward` is the wire form of a forwarded passthrough-plane request
 (Class-2/3 webhooks — Discord interactions, Twilio): `{platform, botId, method,
 path, headers: [[k,v],…], bodyB64}`. The body is base64-encoded so arbitrary

gateway/relay/descriptor.py

@@ -58,6 +58,12 @@ class CapabilityDescriptor:
     emoji: str = "\U0001f50c"  # 🔌 default (matches PlatformEntry default)
     platform_hint: str = ""
     pii_safe: bool = False
+    # Whether the connector can supply surrounding channel/group CONTEXT for an
+    # addressed turn (Model A pull / Model B buffer, per platform). Optional +
+    # defaults False so an older connector that never sends it is treated as
+    # "no context" — additive within contract_version 1. from_json filters
+    # unknown keys, so a connector sending this to an older gateway is safe too.
+    supports_context: bool = False
 
     def to_json(self) -> str:
         """Serialize to a compact, stable JSON string for the handshake frame."""

gateway/relay/ws_transport.py

@@ -34,7 +34,7 @@ import json
 import logging
 import uuid
 from dataclasses import dataclass
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional
 
 from gateway.platforms.base import MessageEvent, MessageType
 from gateway.session import SessionSource
@@ -91,6 +91,44 @@ def _ws_dial_url(url: str) -> str:
     return raw
 
 
+def _render_relay_context(context: Any) -> Optional[str]:
+    """Render the connector's read-only surrounding-context array into the string
+    ``MessageEvent.channel_context`` field.
+
+    The connector attaches ``context`` as a list of normalized message objects
+    (oldest→newest, same channel) for an addressed turn on a context-capable
+    platform (design relay-channel-context). We flatten each to a
+    ``<author>: <text>`` line so it rides the SAME read-only injection path that
+    history-backfill already uses (run.py prepends ``channel_context`` ahead of
+    the trigger message). This is REFERENCE context only — it never triggers the
+    agent; the trigger decision was already made connector-side on the addressed
+    event alone.
+
+    Returns None when there is no usable context (absent/empty list, or a
+    connector that doesn't send the field), so ``channel_context`` stays unset
+    and behaviour is byte-identical to today. Never raises — a malformed context
+    payload must not break inbound delivery of the (already-admitted) turn.
+    """
+    if not context or not isinstance(context, list):
+        return None
+    lines: List[str] = []
+    for item in context:
+        if not isinstance(item, dict):
+            continue
+        text = item.get("text")
+        if not text:
+            continue
+        src = item.get("source") or {}
+        author = ""
+        if isinstance(src, dict):
+            author = src.get("user_name") or src.get("user_id") or ""
+        lines.append(f"{author}: {text}" if author else str(text))
+    if not lines:
+        return None
+    body = "\n".join(lines)
+    return f"[Recent channel messages]\n{body}"
+
+
 def _event_from_wire(raw: Dict[str, Any]) -> MessageEvent:
     """Rebuild a MessageEvent from the connector's normalized inbound payload.
 
@@ -150,6 +188,15 @@ def _event_from_wire(raw: Dict[str, Any]) -> MessageEvent:
         message_id=raw.get("message_id"),
         reply_to_message_id=raw.get("reply_to_message_id"),
         media_urls=raw.get("media_urls") or [],
+        # Surrounding channel/group CONTEXT the connector attached for this
+        # addressed turn (design relay-channel-context): a read-only, oldest→
+        # newest list of nearby non-addressed messages (Model A pull / Model B
+        # buffer). Rendered into the existing ``channel_context`` field — the
+        # same read-only injection path history-backfill already uses
+        # (run.py prepends it ahead of the trigger message). Absent / empty on a
+        # connector that doesn't send it, a dm, or a no-context platform, so
+        # this is purely additive and byte-identical to today when unset.
+        channel_context=_render_relay_context(raw.get("context")),
     )
 
 

tests/gateway/relay/test_channel_context_consume.py

@@ -0,0 +1,143 @@
+"""Unit tests for relay channel-context consumption (design relay-channel-context).
+
+Covers:
+  - CapabilityDescriptor.supports_context: default False + JSON round-trip +
+    forward-compat (older gateway ignores it / newer connector sends it).
+  - _event_from_wire mapping the connector's read-only `context` array into the
+    existing MessageEvent.channel_context injection field, and leaving it unset
+    (byte-identical to today) when absent/empty/malformed.
+  - The trigger text is never affected by context (read-only invariant).
+
+Pure unit tests: no socket, no websockets dependency.
+"""
+
+from __future__ import annotations
+
+from gateway.relay.descriptor import CapabilityDescriptor
+from gateway.relay.ws_transport import _event_from_wire, _render_relay_context
+
+
+def _descriptor_kwargs(**overrides):
+    base = dict(
+        contract_version=1,
+        platform="discord",
+        label="Discord",
+        max_message_length=2000,
+        supports_draft_streaming=False,
+        supports_edit=True,
+        supports_threads=True,
+        markdown_dialect="discord",
+        len_unit="chars",
+    )
+    base.update(overrides)
+    return base
+
+
+class TestDescriptorSupportsContext:
+    def test_defaults_false(self):
+        d = CapabilityDescriptor(**_descriptor_kwargs())
+        assert d.supports_context is False
+
+    def test_round_trip_true(self):
+        d = CapabilityDescriptor(**_descriptor_kwargs(supports_context=True))
+        restored = CapabilityDescriptor.from_json(d.to_json())
+        assert restored.supports_context is True
+
+    def test_from_json_absent_defaults_false(self):
+        # An older connector that never sends the key -> default False.
+        payload = (
+            '{"contract_version":1,"platform":"discord","label":"Discord",'
+            '"max_message_length":2000,"supports_draft_streaming":false,'
+            '"supports_edit":true,"supports_threads":true,'
+            '"markdown_dialect":"discord","len_unit":"chars"}'
+        )
+        d = CapabilityDescriptor.from_json(payload)
+        assert d.supports_context is False
+
+    def test_from_json_ignores_unknown_keys(self):
+        # Forward-compat: a newer connector sending extra keys must not break.
+        payload = (
+            '{"contract_version":1,"platform":"discord","label":"Discord",'
+            '"max_message_length":2000,"supports_draft_streaming":false,'
+            '"supports_edit":true,"supports_threads":true,'
+            '"markdown_dialect":"discord","len_unit":"chars",'
+            '"supports_context":true,"some_future_field":123}'
+        )
+        d = CapabilityDescriptor.from_json(payload)
+        assert d.supports_context is True
+
+
+class TestRenderRelayContext:
+    def test_none_and_empty_return_none(self):
+        assert _render_relay_context(None) is None
+        assert _render_relay_context([]) is None
+        assert _render_relay_context("not a list") is None
+
+    def test_renders_author_and_text_oldest_first(self):
+        ctx = [
+            {"text": "first", "source": {"user_name": "alice"}},
+            {"text": "second", "source": {"user_name": "bob"}},
+        ]
+        out = _render_relay_context(ctx)
+        assert out is not None
+        assert "alice: first" in out
+        assert "bob: second" in out
+        # order preserved (oldest -> newest)
+        assert out.index("first") < out.index("second")
+
+    def test_falls_back_to_user_id_then_bare_text(self):
+        ctx = [
+            {"text": "has id", "source": {"user_id": "u123"}},
+            {"text": "no author", "source": {}},
+        ]
+        out = _render_relay_context(ctx)
+        assert "u123: has id" in out
+        assert "no author" in out
+
+    def test_skips_malformed_items_without_raising(self):
+        ctx = ["not a dict", {"no_text": True}, {"text": "", "source": {}}, 42]
+        # Nothing usable -> None, and definitely no exception.
+        assert _render_relay_context(ctx) is None
+
+
+class TestEventFromWireContext:
+    def _wire(self, **overrides):
+        base = {
+            "text": "@bot repeat what they said above",
+            "message_type": "text",
+            "source": {
+                "platform": "discord",
+                "chat_id": "chan-1",
+                "chat_type": "channel",
+                "user_id": "author-1",
+            },
+            "message_id": "m-100",
+        }
+        base.update(overrides)
+        return base
+
+    def test_context_maps_into_channel_context(self):
+        ev = _event_from_wire(
+            self._wire(
+                context=[
+                    {"text": "earlier", "source": {"user_name": "alice"}},
+                ]
+            )
+        )
+        assert ev.channel_context is not None
+        assert "alice: earlier" in ev.channel_context
+
+    def test_no_context_leaves_channel_context_unset(self):
+        ev = _event_from_wire(self._wire())
+        assert ev.channel_context is None
+
+    def test_empty_context_leaves_channel_context_unset(self):
+        ev = _event_from_wire(self._wire(context=[]))
+        assert ev.channel_context is None
+
+    def test_context_does_not_alter_trigger_text(self):
+        # Read-only invariant: the addressed text is untouched by context.
+        ev = _event_from_wire(
+            self._wire(context=[{"text": "noise", "source": {"user_name": "x"}}])
+        )
+        assert ev.text == "@bot repeat what they said above"