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 <victor@rocketfueldev.com>

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

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(

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:

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"),

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

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)

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