feat(relay): Phase 1 parity — supported_ops discovery, wire identity fields, /handoff aliasing, provision displayName (#71300)

- CapabilityDescriptor.supported_ops + supports_op() with legacy-op-set
  fallback (additive, contract doc §2 updated)
- get_chat_info gated on op discovery (skip round trip on legacy connectors)
- _event_from_wire consumes user_display_name/user_handle (§3 fields
  previously dropped); native display-name parity, session-key stable
- /handoff <fronted-platform> works on relay-fronted gateways: CLI pre-check
  reads the GATEWAY_RELAY_PLATFORMS fronted set; watcher resolves through
  resolve_delivery_transport and replies via send_for_platform
- self-provision forwards displayName (env > skin branding, stock brand
  suppressed) — the primary name source for gg#171 attribution

docs/relay-connector-contract.md

@@ -52,6 +52,7 @@ JSON object. Source of truth: `gateway/relay/descriptor.py`.
 | `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. |
+| `supported_ops` | string[] | no | Op-level capability discovery: the outbound op names the connector's sender for this platform actually implements (e.g. `["send", "edit", "typing", "follow_up", "get_chat_info"]`). Absent/empty ⇒ the connector predates the field and the gateway assumes the legacy op set (`send`/`edit`/`typing`/`follow_up`); a NEW op is used only when explicitly advertised. |
 
 Most fields are a projection of the gateway's existing `PlatformEntry`; the
 runtime-only fields (`len_unit`, `supports_*`, `markdown_dialect`) come from the

gateway/relay/__init__.py

@@ -256,6 +256,44 @@ def relay_wake_url() -> Optional[str]:
     return value.rstrip("/") or None
 
 
+def relay_display_name() -> Optional[str]:
+    """The human-facing agent display name, forwarded at provision (Phase 1 parity).
+
+    The PRIMARY source for the connector's multi-agent reply-attribution prefix
+    (gateway-gateway #171): in a multi-agent scope the shared bot prepends
+    ``**<displayName>:** `` to this instance's replies. Gateway-asserted but
+    safely scoped exactly like ``relay_instance_id()`` / ``relay_wake_url()`` —
+    the tenant stays token-verified, so a dishonest gateway can only label its
+    OWN instance. Absent -> the connector stores null and attribution falls
+    back to the instance's linked-owner identity, else skips the prefix.
+
+    Env first (Docker/NAS stamps ``GATEWAY_RELAY_DISPLAY_NAME``), then the
+    skin's branded agent name (``get_branding("agent_name")`` — the same value
+    the CLI banner shows), so a self-hosted rename via skin config propagates
+    on the next boot's re-provision (the connector rotates on change, same as
+    a wake-url move).
+    """
+    value = os.environ.get("GATEWAY_RELAY_DISPLAY_NAME", "").strip()
+    if not value:
+        try:
+            from hermes_cli.skin_engine import get_active_skin  # late import: boot-safe
+
+            value = str(
+                get_active_skin().get_branding("agent_name", "") or ""
+            ).strip()
+        except Exception:  # noqa: BLE001 - branding absence must never crash boot
+            value = ""
+        # The stock brand name is IDENTICAL on every default install, so in a
+        # multi-agent scope it would prefix every reply "**Hermes Agent:**" —
+        # shadowing the connector's linked-owner fallback, which actually
+        # disambiguates. Only a deliberately customized name is forwarded.
+        if value == "Hermes Agent":
+            value = ""
+    # Mirror the connector's ingest sanitization (trim + 64-char cap) so what
+    # we send is what gets stored.
+    return value[:64] or None
+
+
 def _provision_url(relay_dial_url: str) -> str:
     """Map the ``ws(s)://…/relay`` dial URL to the ``http(s)://…/relay/provision`` POST URL."""
     raw = relay_dial_url.rstrip("/")
@@ -384,6 +422,7 @@ def _post_provision(
     route_keys: list[str],
     instance_id: Optional[str] = None,
     wake_url: Optional[str] = None,
+    display_name: Optional[str] = None,
     timeout: float = 15.0,
 ) -> dict:
     """POST to the connector's ``/relay/provision`` and return the JSON body.
@@ -413,6 +452,11 @@ def _post_provision(
     # stores null and simply can't wake this instance (buffering still works).
     if wake_url:
         body["wakeUrl"] = wake_url
+    # Same for the display name (Phase 1 parity, gg#171): omit when absent so
+    # the connector stores null and attribution falls back to the linked-owner
+    # identity.
+    if display_name:
+        body["displayName"] = display_name
     data = json.dumps(body).encode("utf-8")
     req = urllib.request.Request(
         provision_url,
@@ -591,6 +635,7 @@ def self_provision_relay() -> bool:
     route_keys = relay_route_keys()
     instance_id = relay_instance_id()
     wake_url = relay_wake_url()
+    display_name = relay_display_name()
 
     # Phase 1.5 (D-Q1.5c): provision EACH fronted platform under the SAME
     # gatewayId + the SAME (platform-less) per-gateway secret. The connector's
@@ -615,6 +660,7 @@ def self_provision_relay() -> bool:
                 route_keys=route_keys,
                 instance_id=instance_id,
                 wake_url=wake_url,
+                display_name=display_name,
             )
         except RuntimeError as exc:
             logger.warning(

gateway/relay/adapter.py

@@ -721,7 +721,12 @@ class RelayAdapter(BasePlatformAdapter):
 
     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:
+        # Gated on op-level capability discovery: a connector that doesn't
+        # advertise get_chat_info in supported_ops (including every legacy
+        # connector, where supported_ops is empty and the LEGACY_OPS set
+        # applies) would only return "unsupported op", so skip the round trip
+        # and answer with the same local fallback the error path produced.
+        if self._transport is None or not self.descriptor.supports_op("get_chat_info"):
             return {"name": chat_id, "type": "dm"}
         return await self._transport.get_chat_info(chat_id)
 

gateway/relay/descriptor.py

@@ -64,6 +64,32 @@ class CapabilityDescriptor:
     # "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
+    # Op-level capability discovery (Phase 1 parity): the outbound op names the
+    # connector's sender for this platform actually implements (e.g.
+    # ["send", "edit", "typing", "follow_up", "get_chat_info"]). Empty tuple =
+    # the connector predates the field; callers MUST treat that as "legacy op
+    # set" (send/edit/typing/follow_up) rather than "nothing supported", so an
+    # old connector keeps working unchanged. Additive within contract_version 1.
+    # Stored as a tuple so the frozen dataclass stays hashable/immutable.
+    supported_ops: tuple = ()
+
+    # The op set every connector supported before ``supported_ops`` existed.
+    # Used as the assumed capability set when a legacy connector sends no list.
+    LEGACY_OPS = ("send", "edit", "typing", "follow_up")
+
+    def supports_op(self, op: str) -> bool:
+        """Whether the connector advertises the outbound op ``op``.
+
+        Fail-open for legacy connectors: an empty ``supported_ops`` means the
+        connector predates op discovery, so assume the legacy op set (the four
+        ops every connector implemented before the field existed). A NEW op
+        (e.g. ``get_chat_info``) is therefore only True when explicitly
+        advertised — exactly the discovery semantics Phase 1 needs: the gateway
+        can probe capability without trying the op and parsing an error.
+        """
+        if not self.supported_ops:
+            return op in self.LEGACY_OPS
+        return op in self.supported_ops
 
     def to_json(self) -> str:
         """Serialize to a compact, stable JSON string for the handshake frame."""
@@ -93,6 +119,19 @@ class CapabilityDescriptor:
                     filtered["max_message_length"] = 4096
             except (TypeError, ValueError):
                 filtered["max_message_length"] = 4096
+        # Normalize supported_ops at the trust boundary: JSON carries a list;
+        # the frozen dataclass stores a tuple. Non-list/malformed values (or a
+        # list holding non-strings) degrade to () — the legacy-op-set fallback —
+        # rather than raising, matching the "malformed input never breaks the
+        # handshake" posture above.
+        if "supported_ops" in filtered:
+            raw_ops = filtered["supported_ops"]
+            if isinstance(raw_ops, (list, tuple)):
+                filtered["supported_ops"] = tuple(
+                    str(op) for op in raw_ops if isinstance(op, str) and op
+                )
+            else:
+                filtered["supported_ops"] = ()
         return cls(**filtered)
 
     @classmethod

gateway/relay/ws_transport.py

@@ -186,7 +186,17 @@ def _event_from_wire(raw: Dict[str, Any]) -> MessageEvent:
         chat_type=src.get("chat_type", "dm"),
         chat_name=src.get("chat_name"),
         user_id=src.get("user_id"),
-        user_name=src.get("user_name"),
+        # Native adapters surface the human-facing DISPLAY name as user_name
+        # (e.g. Discord `message.author.display_name`); the connector sends the
+        # raw platform username as user_name plus optional user_display_name /
+        # user_handle enrichments (contract §3). Prefer the display name for
+        # parity with native lanes — session keys derive from user_id, never
+        # user_name, so this is presentation-only and key-stable.
+        user_name=(
+            src.get("user_display_name")
+            or src.get("user_name")
+            or src.get("user_handle")
+        ),
         thread_id=src.get("thread_id"),
         chat_topic=src.get("chat_topic"),
         user_id_alt=src.get("user_id_alt"),

gateway/run.py

@@ -8793,12 +8793,18 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         except (ValueError, KeyError):
             raise RuntimeError(f"unknown platform '{platform_name}'")
 
-        # Adapter must be live
-        adapter = self.adapters.get(platform)
-        if not adapter:
+        # Adapter must be live. A relay-fronted gateway registers ONE adapter
+        # under Platform.RELAY that fronts N logical platforms — so a literal
+        # adapters.get(discord) misses even though "discord" is deliverable.
+        # resolve_delivery_transport is the shared alias-aware resolver (native
+        # adapter wins; relay eligible only when its authenticated transport
+        # advertises it fronts the logical platform).
+        transport = resolve_delivery_transport(platform, self.config, self.adapters)
+        if not transport:
             raise RuntimeError(
                 f"platform '{platform_name}' is not active in this gateway"
             )
+        adapter = transport.adapter
 
         # Home channel must be configured
         home = self.config.get_home_channel(platform)
@@ -8938,15 +8944,18 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
         # Send the agent's reply to the destination. Route to the new
         # thread if we created one; otherwise the configured home channel
-        # (which may itself carry a thread_id).
+        # (which may itself carry a thread_id). Send through the resolved
+        # transport (not adapter.send directly) so a relay-fronted logical
+        # platform is stamped on the outbound frame (send_for_platform).
         send_metadata: Dict[str, Any] = {}
         if effective_thread_id:
             send_metadata["thread_id"] = effective_thread_id
         try:
-            result = await adapter.send(
-                chat_id=str(home.chat_id),
-                content=response_text,
-                metadata=send_metadata or None,
+            result = await transport.send(
+                platform,
+                str(home.chat_id),
+                response_text,
+                send_metadata or None,
             )
         except Exception as exc:
             raise RuntimeError(f"adapter.send failed: {exc}") from exc

hermes_cli/cli_commands_mixin.py

@@ -569,8 +569,26 @@ class CLICommandsMixin:
 
         pcfg = gw_config.platforms.get(platform)
         if not pcfg or not pcfg.enabled:
-            _cprint(f"  Platform '{platform_name}' is not configured/enabled in the gateway.")
-            return True
+            # Relay aliasing: a relay-fronted gateway has no per-platform
+            # config block for the logical platform ("discord" etc.) — only a
+            # RELAY entry — yet /handoff discord is deliverable when the relay
+            # fronts it. The fronted set is deploy config
+            # (GATEWAY_RELAY_PLATFORMS), readable here without the live
+            # adapter; the gateway watcher re-checks against the authenticated
+            # transport (resolve_delivery_transport) before dispatch, so this
+            # is a UX pre-check, not the security gate.
+            relay_fronts = False
+            try:
+                from gateway.relay import relay_platform_identities
+                relay_cfg = gw_config.platforms.get(Platform.RELAY)
+                if relay_cfg and relay_cfg.enabled:
+                    fronted = {p for p, _ in relay_platform_identities()}
+                    relay_fronts = platform_name in fronted
+            except Exception:
+                relay_fronts = False
+            if not relay_fronts:
+                _cprint(f"  Platform '{platform_name}' is not configured/enabled in the gateway.")
+                return True
 
         home = gw_config.get_home_channel(platform)
         if not home or not home.chat_id:

tests/gateway/relay/test_descriptor.py

@@ -96,3 +96,45 @@ def test_module_is_marked_experimental():
     import gateway.relay.descriptor as m
 
     assert "EXPERIMENTAL" in (m.__doc__ or "")
+
+
+# ─────────────── supported_ops (op-level capability discovery, Phase 1) ───────────────
+
+def test_supported_ops_roundtrips_json():
+    d = _telegram_descriptor(supported_ops=("send", "edit", "typing"))
+    restored = CapabilityDescriptor.from_json(d.to_json())
+    assert restored.supported_ops == ("send", "edit", "typing")
+    assert restored == d
+
+
+def test_supports_op_advertised_list_is_authoritative():
+    d = _telegram_descriptor(supported_ops=("send", "typing", "get_chat_info"))
+    assert d.supports_op("send") is True
+    assert d.supports_op("get_chat_info") is True
+    # An advertised list EXCLUDES what it omits — even a legacy op.
+    assert d.supports_op("edit") is False
+
+
+def test_supports_op_legacy_connector_assumes_legacy_set():
+    """An empty supported_ops means the connector predates op discovery: the
+    legacy four ops are assumed supported (old connectors keep working), while
+    NEW ops are not (discovery semantics — never probe by trying)."""
+    d = _telegram_descriptor()  # no supported_ops
+    for op in ("send", "edit", "typing", "follow_up"):
+        assert d.supports_op(op) is True, op
+    assert d.supports_op("get_chat_info") is False
+
+
+def test_from_json_normalizes_malformed_supported_ops():
+    """Non-list shapes and non-string members degrade to the legacy fallback
+    (empty tuple), never raise — malformed input can't break the handshake."""
+    base = (
+        '{"contract_version": 1, "platform": "x", "label": "X", '
+        '"max_message_length": 2000, "supports_draft_streaming": false, '
+        '"supports_edit": true, "supports_threads": false, '
+        '"markdown_dialect": "plain", "len_unit": "chars", '
+    )
+    d = CapabilityDescriptor.from_json(base + '"supported_ops": "send"}')
+    assert d.supported_ops == ()
+    d = CapabilityDescriptor.from_json(base + '"supported_ops": ["send", 7, null, ""]}')
+    assert d.supported_ops == ("send",)

tests/gateway/relay/test_handoff_relay_aliasing.py

@@ -0,0 +1,123 @@
+"""Unit tests for /handoff platform aliasing over relay (Phase 1 parity).
+
+Symptom fixed: on a relay-fronted gateway, ``/handoff discord`` was rejected
+in TWO places even though "discord" is deliverable through the relay adapter:
+
+  1. The CLI pre-check (``cli_commands_mixin._handle_handoff_command``) looked
+     up ``gw_config.platforms.get(DISCORD)`` — only a RELAY entry exists.
+  2. The gateway watcher (``run.py _process_handoff``) did a literal
+     ``adapters.get(discord)`` — the adapter is registered under RELAY.
+
+Both now resolve fronted platforms through the same alias-aware machinery the
+delivery router uses (``resolve_delivery_transport`` — native adapter wins;
+relay eligible only when its authenticated transport advertises the logical
+platform). These tests cover the resolver semantics the watcher depends on,
+plus the CLI pre-check's env-derived fronted set.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.delivery import resolve_delivery_transport
+
+
+class _RelayStub:
+    """Minimal relay adapter double advertising a fronted-platform set."""
+
+    def __init__(self, fronted):
+        self._fronted = set(fronted)
+        self.sent = []
+
+    def fronts_platform(self, platform):
+        value = getattr(platform, "value", platform)
+        return str(value) in self._fronted
+
+    async def send_for_platform(self, logical_platform, chat_id, content, reply_to=None, metadata=None):
+        self.sent.append((getattr(logical_platform, "value", logical_platform), chat_id, content, metadata))
+        return type("R", (), {"success": True, "message_id": "m-1", "error": None})()
+
+    async def send(self, chat_id, content, reply_to=None, metadata=None):  # pragma: no cover
+        raise AssertionError("relay transport must use send_for_platform, not send")
+
+
+def _config_with(platforms):
+    cfg = GatewayConfig()
+    cfg.platforms = platforms
+    return cfg
+
+
+class TestHandoffRelayAliasing:
+    def test_fronted_platform_resolves_to_relay_adapter(self):
+        """The watcher's exact miss: adapters holds only RELAY, target is discord."""
+        relay = _RelayStub({"discord"})
+        cfg = _config_with({Platform.RELAY: PlatformConfig(enabled=True)})
+        transport = resolve_delivery_transport(
+            Platform.DISCORD, cfg, {Platform.RELAY: relay}
+        )
+        assert transport is not None
+        assert transport.adapter is relay
+        assert transport.is_relay is True
+
+    def test_unfronted_platform_still_fails(self):
+        """Aliasing must not let relay hijack a platform it does not front."""
+        relay = _RelayStub({"telegram"})
+        cfg = _config_with({Platform.RELAY: PlatformConfig(enabled=True)})
+        transport = resolve_delivery_transport(
+            Platform.DISCORD, cfg, {Platform.RELAY: relay}
+        )
+        assert transport is None
+
+    def test_native_adapter_wins_over_relay(self):
+        native = object()
+        relay = _RelayStub({"discord"})
+        cfg = _config_with(
+            {
+                Platform.DISCORD: PlatformConfig(enabled=True),
+                Platform.RELAY: PlatformConfig(enabled=True),
+            }
+        )
+        transport = resolve_delivery_transport(
+            Platform.DISCORD, cfg, {Platform.DISCORD: native, Platform.RELAY: relay}
+        )
+        assert transport is not None
+        assert transport.adapter is native
+        assert transport.is_relay is False
+
+    @pytest.mark.asyncio
+    async def test_transport_send_stamps_logical_platform(self):
+        """The handoff reply leg must go through send_for_platform so the
+        outbound frame carries the logical platform tag."""
+        relay = _RelayStub({"discord"})
+        cfg = _config_with({Platform.RELAY: PlatformConfig(enabled=True)})
+        transport = resolve_delivery_transport(
+            Platform.DISCORD, cfg, {Platform.RELAY: relay}
+        )
+        assert transport is not None
+        result = await transport.send(
+            Platform.DISCORD, "chan-1", "handed-off reply", {"thread_id": "t-9"}
+        )
+        assert result.success is True
+        assert relay.sent == [("discord", "chan-1", "handed-off reply", {"thread_id": "t-9"})]
+
+
+class TestCliHandoffFrontedSet:
+    """The CLI pre-check derives the fronted set from deploy env (no live adapter)."""
+
+    def test_fronted_set_from_env(self, monkeypatch):
+        monkeypatch.setenv("GATEWAY_RELAY_PLATFORMS", "discord,telegram")
+        monkeypatch.delenv("GATEWAY_RELAY_BOT_IDS", raising=False)
+        from gateway.relay import relay_platform_identities
+
+        fronted = {p for p, _ in relay_platform_identities()}
+        assert "discord" in fronted
+        assert "telegram" in fronted
+        assert "slack" not in fronted
+
+    def test_unconfigured_env_yields_generic_relay_only(self, monkeypatch):
+        monkeypatch.delenv("GATEWAY_RELAY_PLATFORMS", raising=False)
+        from gateway.relay import relay_platform_identities
+
+        fronted = {p for p, _ in relay_platform_identities()}
+        assert fronted == {"relay"}

tests/gateway/relay/test_relay_adapter.py

@@ -508,3 +508,58 @@ async def test_no_revocation_no_fatal():
     except asyncio.CancelledError:
         pass
     assert a.has_fatal_error is False
+
+
+# ─────────────── get_chat_info gated on supported_ops (Phase 1) ───────────────
+
+
+class _ChatInfoTransport:
+    """Transport stub that records whether get_chat_info was proxied."""
+
+    def __init__(self):
+        self.calls = []
+
+    def set_inbound_handler(self, h):  # noqa: D401
+        self._h = h
+
+    async def get_chat_info(self, chat_id):
+        self.calls.append(chat_id)
+        return {"name": "general", "type": "channel"}
+
+
+@pytest.mark.asyncio
+async def test_get_chat_info_proxied_when_advertised():
+    t = _ChatInfoTransport()
+    a = RelayAdapter(
+        PlatformConfig(),
+        make_desc(supported_ops=("send", "edit", "typing", "get_chat_info")),
+        transport=t,
+    )
+    info = await a.get_chat_info("chan-1")
+    assert info == {"name": "general", "type": "channel"}
+    assert t.calls == ["chan-1"]
+
+
+@pytest.mark.asyncio
+async def test_get_chat_info_local_fallback_for_legacy_connector():
+    """A legacy connector (no supported_ops) would only answer 'unsupported op'
+    — the adapter must skip the round trip and answer locally."""
+    t = _ChatInfoTransport()
+    a = RelayAdapter(PlatformConfig(), make_desc(), transport=t)
+    info = await a.get_chat_info("chan-1")
+    assert info == {"name": "chan-1", "type": "dm"}
+    assert t.calls == []
+
+
+@pytest.mark.asyncio
+async def test_get_chat_info_local_fallback_when_not_advertised():
+    """A connector that advertises ops but OMITS get_chat_info is authoritative."""
+    t = _ChatInfoTransport()
+    a = RelayAdapter(
+        PlatformConfig(),
+        make_desc(supported_ops=("send", "edit", "typing")),
+        transport=t,
+    )
+    info = await a.get_chat_info("chan-1")
+    assert info == {"name": "chan-1", "type": "dm"}
+    assert t.calls == []

tests/gateway/relay/test_relay_roundtrip.py

@@ -20,6 +20,8 @@ from gateway.session import SessionSource, build_session_key
 from gateway.relay.adapter import RelayAdapter
 from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
 
+from dataclasses import replace
+
 from tests.gateway.relay.stub_connector import StubConnector
 
 
@@ -112,6 +114,12 @@ async def test_outbound_send_round_trips(wired):
 @pytest.mark.asyncio
 async def test_get_chat_info_proxied_to_connector(wired):
     adapter, stub = wired
+    # Phase 1: the proxy is gated on op discovery — the connector must
+    # advertise get_chat_info in supported_ops (a legacy descriptor falls
+    # back to the local echo; see test_relay_adapter.py).
+    adapter._apply_descriptor(
+        replace(_discord_descriptor(), supported_ops=("send", "edit", "typing", "get_chat_info"))
+    )
     stub.chat_info["chan1"] = {"name": "general", "type": "group"}
     info = await adapter.get_chat_info("chan1")
     assert info == {"name": "general", "type": "group"}

tests/gateway/relay/test_self_provision.py

@@ -375,3 +375,107 @@ def test_connector_failure_is_non_fatal(monkeypatch):
     monkeypatch.setattr(relay, "_post_provision", _boom)
     assert relay.self_provision_relay() is False
     assert relay.relay_connection_auth() == (None, None)
+
+
+# ─────────────────────────── displayName (Phase 1 parity, gg#171) ───────────────────────────
+
+def test_relay_display_name_env_wins(monkeypatch):
+    monkeypatch.setenv("GATEWAY_RELAY_DISPLAY_NAME", "  Atlas  ")
+    assert relay.relay_display_name() == "Atlas"
+
+
+def test_relay_display_name_caps_at_64(monkeypatch):
+    monkeypatch.setenv("GATEWAY_RELAY_DISPLAY_NAME", "x" * 200)
+    assert relay.relay_display_name() == "x" * 64
+
+
+def test_relay_display_name_falls_back_to_skin_branding(monkeypatch):
+    monkeypatch.delenv("GATEWAY_RELAY_DISPLAY_NAME", raising=False)
+
+    class _Skin:
+        def get_branding(self, key, fallback=""):
+            return "Chatterbox" if key == "agent_name" else fallback
+
+    monkeypatch.setattr("hermes_cli.skin_engine.get_active_skin", lambda: _Skin())
+    assert relay.relay_display_name() == "Chatterbox"
+
+
+def test_relay_display_name_suppresses_stock_brand(monkeypatch):
+    """The default 'Hermes Agent' brand is identical on every install — forwarding
+    it would shadow the connector's linked-owner fallback (which actually
+    disambiguates) with a uniform label. Only customized names are forwarded."""
+    monkeypatch.delenv("GATEWAY_RELAY_DISPLAY_NAME", raising=False)
+
+    class _Skin:
+        def get_branding(self, key, fallback=""):
+            return "Hermes Agent" if key == "agent_name" else fallback
+
+    monkeypatch.setattr("hermes_cli.skin_engine.get_active_skin", lambda: _Skin())
+    assert relay.relay_display_name() is None
+
+
+def test_relay_display_name_branding_failure_is_non_fatal(monkeypatch):
+    monkeypatch.delenv("GATEWAY_RELAY_DISPLAY_NAME", raising=False)
+
+    def _boom():
+        raise RuntimeError("no skin engine")
+
+    monkeypatch.setattr("hermes_cli.skin_engine.get_active_skin", _boom)
+    assert relay.relay_display_name() is None
+
+
+def test_self_provision_forwards_display_name(monkeypatch):
+    _arm(monkeypatch)
+    monkeypatch.setenv("GATEWAY_RELAY_DISPLAY_NAME", "Atlas")
+    captured: dict = {}
+    monkeypatch.setattr(relay, "_post_provision", _stub_post(captured))
+
+    assert relay.self_provision_relay() is True
+    assert captured["display_name"] == "Atlas"
+
+
+def test_post_provision_body_includes_displayName_only_when_set(monkeypatch):
+    """`displayName` joins the body ONLY when a value is supplied — omitting it
+    lets the connector store null (attribution falls back to linked-owner)."""
+    import json
+
+    sent: dict = {}
+
+    class _Resp:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *a):
+            return False
+
+        def read(self):
+            return json.dumps({"secret": "a" * 64, "deliveryKey": "b" * 64, "tenant": "t", "gatewayId": "gw-1"}).encode()
+
+    def _fake_urlopen(req, timeout=None):  # noqa: ANN001
+        sent["body"] = json.loads(req.data.decode())
+        return _Resp()
+
+    monkeypatch.setattr("urllib.request.urlopen", _fake_urlopen)
+
+    relay._post_provision(
+        provision_url="https://c.example/relay/provision",
+        access_token="tok",
+        gateway_id="gw-1",
+        platform="discord",
+        bot_id="app",
+        gateway_endpoint=None,
+        route_keys=[],
+        display_name="Atlas",
+    )
+    assert sent["body"]["displayName"] == "Atlas"
+
+    relay._post_provision(
+        provision_url="https://c.example/relay/provision",
+        access_token="tok",
+        gateway_id="gw-1",
+        platform="discord",
+        bot_id="app",
+        gateway_endpoint=None,
+        route_keys=[],
+    )
+    assert "displayName" not in sent["body"]

tests/gateway/relay/test_wire_user_identity.py

@@ -0,0 +1,68 @@
+"""Unit tests for relay wire-field hygiene (Phase 1 parity).
+
+Covers _event_from_wire's consumption of the contract §3 user-identity
+enrichment fields the connector has always sent but the gateway used to drop:
+
+  - ``user_display_name`` — the human-facing display name (native parity: the
+    native Discord adapter surfaces ``message.author.display_name`` as
+    ``user_name``).
+  - ``user_handle`` — the raw platform handle, used only as a last resort.
+
+Precedence: user_display_name > user_name > user_handle. Session keys derive
+from user_id (never user_name), so this mapping is presentation-only and must
+remain key-stable — asserted here via build_session_key.
+
+Pure unit tests: no socket, no websockets dependency.
+"""
+
+from __future__ import annotations
+
+from gateway.relay.ws_transport import _event_from_wire
+from gateway.session import build_session_key
+
+
+def _wire_event(**src_overrides):
+    src = {
+        "platform": "discord",
+        "chat_id": "chan-1",
+        "chat_type": "group",
+        "user_id": "u-1",
+        "user_name": "rawusername",
+        "scope_id": "guild-1",
+    }
+    src.update(src_overrides)
+    return {"text": "hello", "message_type": "text", "source": src}
+
+
+class TestUserIdentityEnrichment:
+    def test_display_name_preferred_over_user_name(self):
+        ev = _event_from_wire(_wire_event(user_display_name="Ben Display"))
+        assert ev.source.user_name == "Ben Display"
+
+    def test_user_name_when_no_display_name(self):
+        ev = _event_from_wire(_wire_event())
+        assert ev.source.user_name == "rawusername"
+
+    def test_handle_is_last_resort(self):
+        ev = _event_from_wire(
+            _wire_event(user_name=None, user_handle="ben#1234")
+        )
+        assert ev.source.user_name == "ben#1234"
+
+    def test_all_absent_yields_none(self):
+        ev = _event_from_wire(_wire_event(user_name=None))
+        assert ev.source.user_name is None
+
+    def test_empty_display_name_falls_through(self):
+        """An empty-string enrichment must not shadow a real user_name."""
+        ev = _event_from_wire(_wire_event(user_display_name=""))
+        assert ev.source.user_name == "rawusername"
+
+    def test_session_key_is_stable_across_name_shapes(self):
+        """user_name is presentation-only: the same user_id keys the same
+        session whether or not the connector sent the enrichment fields."""
+        plain = _event_from_wire(_wire_event())
+        enriched = _event_from_wire(
+            _wire_event(user_display_name="Ben Display", user_handle="ben#1234")
+        )
+        assert build_session_key(plain.source) == build_session_key(enriched.source)