feat(relay): delete_message over the additive delete op — fresh-final preview cleanup

Companion to the connector's delete op (gateway-gateway 119a228). The
fresh-final unfurl route re-posts the completed reply and previously left
the sealed streamed preview behind (double delivery). delete_message now
emits op=delete when the negotiated descriptor advertises it; without the
advertisement it returns False with zero wire traffic, degrading to the
old leave-the-preview behavior against older connectors.

Consumer-level test drives placeholder -> stamped fresh final -> delete
of the original preview id.

gateway/relay/adapter.py

@@ -2298,6 +2298,44 @@ class RelayAdapter(BasePlatformAdapter):
             error=result.get("error"),
         )
 
+    async def delete_message(
+        self,
+        chat_id: str,
+        message_id: str,
+    ) -> bool:
+        """Delete a relayed message through the connector-owned platform API.
+
+        Consumer: the stream consumer's fresh-final cleanup — on the Slack
+        unfurl force-on route the completed reply is re-delivered as a new
+        stamped post and the sealed streamed preview must go away, or the
+        user sees the answer twice.
+
+        Gated on the negotiated descriptor advertising the ``delete`` op
+        (additive within contract_version 1): older connectors never receive
+        an op they can't dispatch, and this returns False so the consumer's
+        best-effort cleanup degrades to leaving the preview in place —
+        exactly the pre-delete behavior.
+        """
+        if self._transport is None:
+            return False
+        desc = self._descriptor_for_chat(str(chat_id))
+        if "delete" not in (desc.supported_ops or ()):
+            return False
+        try:
+            result = await self._transport.send_outbound(
+                {
+                    "op": "delete",
+                    "chat_id": chat_id,
+                    "message_id": message_id,
+                    "metadata": self._with_scope(chat_id, {}),
+                },
+                platform=self._platform_by_chat.get(str(chat_id)),
+            )
+        except Exception:
+            logger.debug("relay delete_message failed", exc_info=True)
+            return False
+        return bool(result.get("success"))
+
     async def send_typing(self, chat_id: str, metadata=None) -> None:
         """Egress a typing indicator through the connector.
 

tests/gateway/relay/test_relay_slack_unfurl.py

@@ -410,3 +410,67 @@ class TestConsumerRoutesForceOnFinalAsFreshSend:
         assert ops[0] == "send"
         assert transport.actions[0]["metadata"]["unfurl_links"] is False
         assert ops[-1] == "edit", f"ops={ops}"
+
+
+class TestDeleteOpForFreshFinalCleanup:
+    """Relay delete_message: emitted only when the negotiated descriptor
+    advertises the additive `delete` op; older connectors degrade to the
+    leave-the-preview-behind behavior (return False, no wire traffic)."""
+
+    @pytest.mark.asyncio
+    async def test_delete_emitted_when_advertised(self):
+        transport = _RecordingTransport()
+        a = RelayAdapter(
+            PlatformConfig(extra={"slack": {"unfurl_links": True}}),
+            make_desc(
+                platform="slack",
+                supported_ops=("send", "edit", "delete"),
+            ),
+            transport=transport,
+        )
+        ok = await a.delete_message("D1", "1700000000.000200")
+        assert ok is True
+        assert transport.actions[-1]["op"] == "delete"
+        assert transport.actions[-1]["message_id"] == "1700000000.000200"
+
+    @pytest.mark.asyncio
+    async def test_delete_refused_when_not_advertised(self):
+        transport = _RecordingTransport()
+        a = RelayAdapter(
+            PlatformConfig(extra={"slack": {"unfurl_links": True}}),
+            make_desc(platform="slack", supported_ops=("send", "edit")),
+            transport=transport,
+        )
+        ok = await a.delete_message("D1", "1700000000.000200")
+        assert ok is False
+        assert transport.actions == []  # no wire traffic for old connectors
+
+    @pytest.mark.asyncio
+    async def test_consumer_fresh_final_deletes_preview_when_supported(self):
+        from gateway.stream_consumer import (
+            GatewayStreamConsumer,
+            StreamConsumerConfig,
+        )
+
+        transport = _RecordingTransport()
+        a = RelayAdapter(
+            PlatformConfig(extra={"slack": {"unfurl_links": True, "unfurl_media": True}}),
+            make_desc(
+                platform="slack",
+                supports_edit=True,
+                supported_ops=("send", "edit", "delete"),
+            ),
+            transport=transport,
+        )
+        consumer = GatewayStreamConsumer(
+            adapter=a, chat_id="D1", config=StreamConsumerConfig()
+        )
+        await consumer._send_or_edit("Working on it…")
+        await consumer._send_or_edit("see https://studiotwin.ai", finalize=True)
+
+        ops = [x.get("op") for x in transport.actions]
+        # send (placeholder) ... send (fresh final) ... delete (preview)
+        assert ops[-1] == "delete", f"ops={ops}"
+        deleted = transport.actions[-1]["message_id"]
+        # The deleted message must be the FIRST send's id (m1), not the final.
+        assert deleted == "m1"