From 305a3c74246118e23ccaf42018dab60419c6683a Mon Sep 17 00:00:00 2001 From: Ben Barclay Date: Thu, 23 Jul 2026 12:51:13 +1000 Subject: [PATCH] fix(relay): restore streaming delivery, Slack command parity, and status clearing (salvage of #69716) (#69747) * fix(gateway): restore relay streaming delivery * fix(relay): route Slack parent commands before session gates * fix(relay): clear Slack typing status after turns --------- Co-authored-by: Victor Kyriazakos --- docs/relay-connector-contract.md | 12 ++- gateway/authz_mixin.py | 13 +++ gateway/relay/adapter.py | 69 +++++++++++++++- gateway/relay/ws_transport.py | 46 ++++++++++- tests/gateway/relay/test_relay_adapter.py | 82 +++++++++++++++++++ tests/gateway/test_multiplex_profile_authz.py | 48 +++++++++++ .../test_slack_relay_parent_command.py | 53 ++++++++++++ 7 files changed, 319 insertions(+), 4 deletions(-) create mode 100644 tests/gateway/test_slack_relay_parent_command.py diff --git a/docs/relay-connector-contract.md b/docs/relay-connector-contract.md index c7a5f931582..6560139c2ae 100644 --- a/docs/relay-connector-contract.md +++ b/docs/relay-connector-contract.md @@ -391,13 +391,23 @@ 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`, `metadata?` | `{success: bool}` | +| `typing` | `chat_id`, `content?`, `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 `{name, type}`. Media actions follow the same envelope shape (deferred to a later contract revision; additive). +**`typing` `content?` (Slack status clear).** A `typing` frame normally omits +`content` — the connector renders its platform's active indicator ("is +typing…" Assistant status on Slack, one-shot typing elsewhere). An **empty +string** `content` is an explicit *clear* request: on Slack the connector sets +the Assistant thread status to `""`, dismissing it. The gateway emits the +clear only for Slack (persistent status); one-shot platforms never receive it. +Additive within `contract_version` 1, but note the deploy order: a connector +predating gateway-gateway #154 ignores `content` and would *set* "is typing…" +on a clear frame — deploy the connector first. + **`follow_up` (A2 capability action).** Some inbound payloads carry a credential that acts on the **shared** bot identity (e.g. a Discord interaction follow-up token). Per §6 the connector strips that at the edge and binds it in its diff --git a/gateway/authz_mixin.py b/gateway/authz_mixin.py index 1e4b73fc4b6..45f7f32cf2a 100644 --- a/gateway/authz_mixin.py +++ b/gateway/authz_mixin.py @@ -95,6 +95,19 @@ class GatewayAuthorizationMixin: transport_adapter = self._registered_transport_adapter(source) if transport_adapter is not None: return transport_adapter + # Relay ingress deliberately keeps the underlying platform on the + # source so session keys and display policy remain Slack/Discord/etc. + # Delivery still has to use the one live RelayAdapter that owns the + # authenticated connector socket. Looking up the underlying platform + # here silently disables streaming, typing, and tool progress when a + # managed gateway does not also run that platform's native adapter. + if getattr(source, "delivered_via_upstream_relay", False) is True: + # One process-level RelayAdapter owns the connector socket for all + # multiplexed profiles. Secondary profiles intentionally do not + # register their own relay adapters, so profile-aware lookup would + # fail and suppress streamed delivery for those profiles. + adapters = getattr(self, "adapters", None) or {} + return adapters.get(Platform.RELAY) # ``getattr`` guards test fixtures that build a bare source via # SimpleNamespace and omit ``profile`` (see AGENTS.md pitfall #17). return self._authorization_adapter( diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index f58d7e2c8a5..e743f26b290 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -513,6 +513,34 @@ class RelayAdapter(BasePlatformAdapter): error=result.get("error"), ) + async def edit_message( + self, + chat_id: str, + message_id: str, + content: str, + *, + finalize: bool = False, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Edit a relayed message through the connector-owned platform API.""" + if self._transport is None: + return SendResult(success=False, error="no transport") + result = await self._transport.send_outbound( + { + "op": "edit", + "chat_id": chat_id, + "message_id": message_id, + "content": content, + "metadata": self._with_scope(chat_id, metadata), + }, + platform=self._platform_by_chat.get(str(chat_id)), + ) + return SendResult( + success=bool(result.get("success")), + message_id=result.get("message_id") or message_id, + error=result.get("error"), + ) + async def send_typing(self, chat_id: str, metadata=None) -> None: """Egress a typing indicator through the connector. @@ -536,8 +564,9 @@ class RelayAdapter(BasePlatformAdapter): 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. + one-shot — Discord/Telegram indicators self-expire and need no cleanup; + Slack Assistant status persists, so ``stop_typing`` below sends an + explicit clear for Slack only. """ if self._transport is None: return @@ -553,6 +582,42 @@ class RelayAdapter(BasePlatformAdapter): 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 stop_typing( + self, + chat_id: str, + metadata: Optional[Dict[str, Any]] = None, + ) -> None: + """Forward an explicit typing/status clear to the connector. + + Slack Assistant status persists until explicitly cleared (empty + ``content`` on the ``typing`` op). Other relay senders expose only + one-shot typing heartbeats; sending an empty heartbeat there would + incorrectly re-trigger typing at completion, so this is Slack-gated. + + NOTE (deploy order): a connector older than gateway-gateway #154 + hardcodes ``status: "is typing…"`` for the typing op, so an empty + clear frame would SET the status instead of clearing it. Deploy the + connector first. Best-effort like ``send_typing``: status clearing is + cosmetic and must never break turn completion. + """ + if self._transport is None: + return + platform = self._platform_by_chat.get(str(chat_id)) + if platform != Platform.SLACK.value: + return + try: + await self._transport.send_outbound( + { + "op": "typing", + "chat_id": chat_id, + "content": "", + "metadata": self._with_scope(chat_id, metadata), + }, + platform=platform, + ) + except Exception: # noqa: BLE001 - status clear is cosmetic, never breaks a turn + logger.debug("relay stop_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/gateway/relay/ws_transport.py b/gateway/relay/ws_transport.py index 33aaaa9dbcd..aaba3ee1252 100644 --- a/gateway/relay/ws_transport.py +++ b/gateway/relay/ws_transport.py @@ -129,6 +129,42 @@ def _render_relay_context(context: Any) -> Optional[str]: return f"[Recent channel messages]\n{body}" +def _normalize_slack_parent_command( + text: str, + message_type: MessageType, +) -> tuple[str, MessageType]: + """Mirror native Slack ``/hermes`` routing for authenticated relay text.""" + stripped = text.strip() + parent_parts = stripped.split(maxsplit=1) + if not parent_parts or parent_parts[0] != "/hermes": + return text, message_type + + from hermes_cli.commands import slack_subcommand_map + + payload = parent_parts[1].strip() if len(parent_parts) > 1 else "" + subcommand_map = slack_subcommand_map() + subcommand_map["compact"] = "/compress" + payload_parts = payload.split() if payload else [] + first_word = payload_parts[0] if payload_parts else "" + + if first_word in subcommand_map: + rest = payload[len(first_word) :].strip() + normalized = ( + f"{subcommand_map[first_word]} {rest}".strip() + if rest + else subcommand_map[first_word] + ) + elif payload: + normalized = payload + else: + normalized = "/help" + + normalized_type = ( + MessageType.COMMAND if normalized.startswith("/") else MessageType.TEXT + ) + return normalized, normalized_type + + def _event_from_wire(raw: Dict[str, Any]) -> MessageEvent: """Rebuild a MessageEvent from the connector's normalized inbound payload. @@ -181,8 +217,16 @@ def _event_from_wire(raw: Dict[str, Any]) -> MessageEvent: except ValueError: msg_type = MessageType.TEXT + text = raw.get("text", "") + if platform_enum == Platform.SLACK: + # Team Gateway carries Slack slash text over the authenticated message + # relay, bypassing Hermes' native Slack command callback. Normalize at + # the wire boundary so adapter-level active-session gates see the real + # gateway command rather than the legacy `hermes` parent name. + text, msg_type = _normalize_slack_parent_command(text, msg_type) + return MessageEvent( - text=raw.get("text", ""), + text=text, message_type=msg_type, source=source, message_id=raw.get("message_id"), diff --git a/tests/gateway/relay/test_relay_adapter.py b/tests/gateway/relay/test_relay_adapter.py index 707f0d8891b..64e3b60fe25 100644 --- a/tests/gateway/relay/test_relay_adapter.py +++ b/tests/gateway/relay/test_relay_adapter.py @@ -287,6 +287,88 @@ async def test_scoped_reply_without_inbound_author_carries_scope_only(): assert "user_id" not in t.sent["metadata"] +@pytest.mark.asyncio +async def test_edit_message_forwards_relay_action_with_routing_context(): + t = _CaptureTransport() + a = RelayAdapter(PlatformConfig(), make_desc(platform="slack"), transport=t) + event = _make_event(chat_id="channel-1", scope_id="workspace-1") + event.source.platform = Platform.SLACK + a._capture_scope(event) + + result = await a.edit_message( + "channel-1", + "message-1", + "streamed answer", + metadata={"thread_id": "thread-1"}, + ) + + assert result.success is True + assert t.sent == { + "op": "edit", + "chat_id": "channel-1", + "message_id": "message-1", + "content": "streamed answer", + "metadata": { + "thread_id": "thread-1", + "scope_id": "workspace-1", + }, + } + assert t.sent_platform == "slack" + + +@pytest.mark.asyncio +async def test_stop_typing_forwards_explicit_clear_with_routing_context(): + t = _CaptureTransport() + a = RelayAdapter(PlatformConfig(), make_desc(platform="slack"), transport=t) + event = _make_event(chat_id="channel-1", scope_id="workspace-1") + event.source.platform = Platform.SLACK + a._capture_scope(event) + + await a.stop_typing("channel-1", metadata={"thread_id": "thread-1"}) + + assert t.sent == { + "op": "typing", + "chat_id": "channel-1", + "content": "", + "metadata": { + "thread_id": "thread-1", + "scope_id": "workspace-1", + }, + } + assert t.sent_platform == "slack" + + +@pytest.mark.asyncio +async def test_stop_typing_is_noop_for_non_slack_relay_platforms(): + t = _CaptureTransport() + a = RelayAdapter(PlatformConfig(), make_desc(platform="telegram"), transport=t) + event = _make_event(chat_id="channel-1", scope_id="workspace-1") + event.source.platform = Platform.TELEGRAM + a._capture_scope(event) + + await a.stop_typing("channel-1", metadata={"thread_id": "thread-1"}) + + assert t.sent is None + assert t.sent_platform is None + + +@pytest.mark.asyncio +async def test_stop_typing_swallows_transport_errors(): + """A WS drop at end-of-turn must not propagate out of stop_typing — status + clearing is cosmetic and turn completion must never fail on it.""" + + class _FailingTransport(_CaptureTransport): + async def send_outbound(self, action, *, platform=None): + raise RuntimeError("ws down") + + a = RelayAdapter(PlatformConfig(), make_desc(platform="slack"), transport=_FailingTransport()) + event = _make_event(chat_id="channel-1", scope_id="workspace-1") + event.source.platform = Platform.SLACK + a._capture_scope(event) + + await a.stop_typing("channel-1") # must not raise + + # ── typing indicator over the relay (op="typing") ──────────────────────────── diff --git a/tests/gateway/test_multiplex_profile_authz.py b/tests/gateway/test_multiplex_profile_authz.py index 4289fcb1064..0620b14e0cc 100644 --- a/tests/gateway/test_multiplex_profile_authz.py +++ b/tests/gateway/test_multiplex_profile_authz.py @@ -153,6 +153,54 @@ def test_chat_routed_source_keeps_receiving_shared_adapter(monkeypatch): assert runner._is_user_authorized(source) is True +def test_adapter_for_relay_delivered_source_uses_relay_transport(monkeypatch): + """A relayed Slack event keeps Slack session semantics but replies over relay.""" + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + slack_adapter = SimpleNamespace(send=AsyncMock()) + relay_adapter = SimpleNamespace(send=AsyncMock()) + runner.adapters = { + Platform.SLACK: slack_adapter, + Platform.RELAY: relay_adapter, + } + runner._profile_adapters = {} + + source = SessionSource( + platform=Platform.SLACK, + user_id="U123", + chat_id="C123", + chat_type="channel", + profile="coder", + delivered_via_upstream_relay=True, + ) + + assert runner._adapter_for_source(source) is relay_adapter + + +def test_adapter_for_direct_source_keeps_native_platform_adapter(monkeypatch): + """The relay routing rule must not affect direct Slack connector delivery.""" + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + slack_adapter = SimpleNamespace(send=AsyncMock()) + relay_adapter = SimpleNamespace(send=AsyncMock()) + runner.adapters = { + Platform.SLACK: slack_adapter, + Platform.RELAY: relay_adapter, + } + runner._profile_adapters = {} + + source = SessionSource( + platform=Platform.SLACK, + user_id="U123", + chat_id="C123", + chat_type="channel", + ) + + assert runner._adapter_for_source(source) is slack_adapter + + def test_secondary_allowlist_dm_behavior_ignores_unauthorized(monkeypatch): """Unauthorized-DM behavior must read the secondary adapter's dm_policy.""" runner, _default_adapter, secondary_adapter = _make_multiplex_runner(monkeypatch) diff --git a/tests/gateway/test_slack_relay_parent_command.py b/tests/gateway/test_slack_relay_parent_command.py new file mode 100644 index 00000000000..260e72c735a --- /dev/null +++ b/tests/gateway/test_slack_relay_parent_command.py @@ -0,0 +1,53 @@ +import pytest + +from gateway.config import Platform +from gateway.platforms.base import MessageType +from gateway.relay.ws_transport import _event_from_wire + + +def _wire(text: str, *, platform: str = "slack") -> dict: + return { + "text": text, + "message_type": "command", + "source": { + "platform": platform, + "chat_id": "D123", + "chat_type": "dm", + "user_id": "U123", + }, + } + + +@pytest.mark.parametrize( + ("wire_text", "expected"), + [ + ("/hermes sethome", "/sethome"), + ("/hermes\tsethome", "/sethome"), + ( + "/hermes model gpt-5.6 --provider openai", + "/model gpt-5.6 --provider openai", + ), + ("/hermes", "/help"), + ], +) +def test_slack_relay_parent_becomes_gateway_command(wire_text: str, expected: str): + event = _event_from_wire(_wire(wire_text)) + + assert event.text == expected + assert event.message_type == MessageType.COMMAND + assert event.source.platform == Platform.SLACK + assert event.source.delivered_via_upstream_relay is True + + +def test_slack_relay_parent_freeform_text_matches_native_adapter(): + event = _event_from_wire(_wire("/hermes explain this")) + + assert event.text == "explain this" + assert event.message_type == MessageType.TEXT + + +def test_non_slack_relay_message_is_not_rewritten(): + event = _event_from_wire(_wire("/hermes sethome", platform="discord")) + + assert event.text == "/hermes sethome" + assert event.message_type == MessageType.COMMAND