fix(relay): drop replayed inbounds — bounded dedupe on (chat_id, message_id)

Live-canary finding #3 (Alice, staging): the relay inbound leg is at-least-once. On WS re-handshake the connector replays its durable per-instance buffer; a long multi-tool turn (60-100s) straddling a quiet socket drop got its ORIGINAL inbound replayed after the turn finished, re-running the entire turn — the user saw the final answer posted 2-5x (each a separate execution, hence slightly different texts). Receipts: same msg text at history=0 in back-to-back sessions 121647/121840, no Slack-side retry on the connector (envelope dedupe never fired). Consumer-side idempotency: bounded FIFO seen-set (512) keyed by platform message identity; events without a message_id never dedupe (fail-open — dropping a real message is worse than rerunning one). No wire change; contract v1 untouched. Transplanted-from: victor-fork/feat/relay-slack-live-cards@73ce04ae75 (extracted for the rc.4 relay-fixes train; tests moved to a standalone file with no live-cards dependencies)
2026-08-15 12:30:13 +00:00
parent 534f8d4fb0
commit b5f95d0890
2 changed files with 160 additions and 0 deletions
@@ -95,6 +95,9 @@ class RelayAdapter(BasePlatformAdapter):
        # feedback off SendResult — see send()). Consumed by the gateway's
        # semantic thread-rename lane; bounded like the sibling caches.
        self._auto_thread_by_chat: Dict[str, Tuple[str, str]] = {}
+        # Bounded FIFO seen-set for inbound replay dedupe (finding #3);
+        # dict preserves insertion order, giving cheap oldest-first eviction.
+        self._seen_inbound: Dict[str, None] = {}
        # chat_id -> event fired when the entry above lands, so a consumer that
        # arrives before the send can wait for it instead of polling. See
        # wait_for_auto_thread_info.
@@ -361,6 +364,24 @@ class RelayAdapter(BasePlatformAdapter):

    async def _on_inbound(self, event) -> None:
        """Bridge a connector-delivered MessageEvent into the normal adapter path."""
+        # Inbound replay dedupe (live-canary finding #3, Alice staging): the
+        # relay leg is at-least-once — on WS re-handshake the connector
+        # replays its durable per-instance buffer, and a long multi-tool turn
+        # (60-100s) straddling a quiet socket drop gets its ORIGINAL inbound
+        # replayed after the turn completes, re-running the whole turn (user
+        # saw the final answer 2-5x). Platform message identity (chat_id +
+        # message_id/ts) is stable across replays, so a bounded seen-set
+        # drops them. Consumer-side idempotency; no wire change.
+        dedupe_key = self._inbound_dedupe_key(event)
+        if dedupe_key is not None:
+            if dedupe_key in self._seen_inbound:
+                logger.info(
+                    "relay inbound dropped as replay (dedupe key=%s)", dedupe_key
+                )
+                return
+            self._seen_inbound[dedupe_key] = None
+            while len(self._seen_inbound) > self._SEEN_INBOUND_MAX:
+                self._seen_inbound.pop(next(iter(self._seen_inbound)))
        self._capture_scope(event)
        self._stamp_slack_session_thread(event)
        # Phase 3: a structured prompt answer resolves its waiting primitive
@@ -372,6 +393,22 @@ class RelayAdapter(BasePlatformAdapter):
        await self._localize_inbound_media(event)
        await self.handle_message(event)

+    _SEEN_INBOUND_MAX = 512
+
+    def _inbound_dedupe_key(self, event) -> Optional[str]:
+        """Stable replay identity: (chat, platform message id).
+
+        Returns None when the event carries no platform message id (synthetic
+        events, some prompt responses) — those never dedupe, fail-open by
+        design: dropping a real user message is strictly worse than rerunning
+        one, so only dedupe when identity is certain.
+        """
+        message_id = getattr(event, "message_id", None)
+        chat_id = getattr(event, "chat_id", None)
+        if not message_id or not chat_id:
+            return None
+        return f"{chat_id}:{message_id}"
+
    def _relay_slack_extra(self) -> Dict[str, Any]:
        """The Slack-behavior subset of the RELAY platform config.

@@ -0,0 +1,123 @@
+"""Inbound replay dedupe on the relay adapter (transplanted from the
+live-cards branch for the rc.4 relay-fixes train).
+
+Live-canary finding #3 (Alice, staging): the relay inbound leg is
+at-least-once. On WS re-handshake the connector replays its durable
+per-instance buffer; a long multi-tool turn straddling a quiet socket drop
+got its ORIGINAL inbound replayed after the turn finished, re-running the
+entire turn — the user saw the final answer posted 2-5x. Platform message
+identity (chat_id + message_id/ts) is stable across replays, so a bounded
+seen-set drops them. Fail-open: events without a message_id never dedupe
+(dropping a real message is strictly worse than rerunning one).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from gateway.config import PlatformConfig
+from gateway.relay.adapter import RelayAdapter
+from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
+from tests.gateway.relay.stub_connector import StubConnector
+
+
+def make_desc(**kw) -> CapabilityDescriptor:
+    base = dict(
+        contract_version=CONTRACT_VERSION,
+        platform="slack",
+        label="Slack",
+        max_message_length=39000,
+        supports_draft_streaming=True,
+        supports_edit=True,
+        supports_threads=True,
+        markdown_dialect="slack",
+        len_unit="chars",
+        emoji="\U0001f4ac",
+        platform_hint="",
+        pii_safe=False,
+        supported_ops=("send", "edit", "typing"),
+    )
+    base.update(kw)
+    return CapabilityDescriptor(**base)
+
+
+def _connected_adapter(**desc_kw):
+    desc = make_desc(**desc_kw)
+    stub = StubConnector(desc)
+    adapter = RelayAdapter(PlatformConfig(), desc, transport=stub)
+    return adapter, stub
+
+
+@pytest.fixture()
+def loop():
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    yield loop
+    loop.close()
+
+
+def _record(bucket, event):
+    async def _coro():
+        bucket.append(event)
+    return _coro()
+
+
+async def _false_coro():
+    return False
+
+
+async def _none_coro():
+    return None
+
+
+class TestInboundReplayDedupe:
+    """Finding #3 (live canary): connector replay of the original inbound
+    after a WS re-handshake must not re-run the turn."""
+
+    def _event(self, message_id="1700.100", chat_id="C1", text="hi"):
+        return SimpleNamespace(
+            message_id=message_id, chat_id=chat_id, text=text, media=None
+        )
+
+    def _tap(self, adapter, handled):
+        adapter.handle_message = lambda e: _record(handled, e)
+        adapter._consume_prompt_response = lambda e: _false_coro()
+        adapter._localize_inbound_media = lambda e: _none_coro()
+
+    def test_replayed_inbound_dropped(self, loop):
+        adapter, _ = _connected_adapter()
+        handled = []
+        self._tap(adapter, handled)
+        e = self._event()
+        loop.run_until_complete(adapter._on_inbound(e))
+        loop.run_until_complete(adapter._on_inbound(e))  # replay
+        assert len(handled) == 1
+
+    def test_distinct_messages_both_handled(self, loop):
+        adapter, _ = _connected_adapter()
+        handled = []
+        self._tap(adapter, handled)
+        loop.run_until_complete(adapter._on_inbound(self._event("1700.100")))
+        loop.run_until_complete(adapter._on_inbound(self._event("1700.200")))
+        assert len(handled) == 2
+
+    def test_missing_message_id_fails_open(self, loop):
+        adapter, _ = _connected_adapter()
+        handled = []
+        self._tap(adapter, handled)
+        e = self._event(message_id=None)
+        loop.run_until_complete(adapter._on_inbound(e))
+        loop.run_until_complete(adapter._on_inbound(e))
+        assert len(handled) == 2  # never dedupe without identity
+
+    def test_seen_set_bounded(self, loop):
+        adapter, _ = _connected_adapter()
+        adapter.handle_message = lambda e: _none_coro()
+        adapter._consume_prompt_response = lambda e: _false_coro()
+        adapter._localize_inbound_media = lambda e: _none_coro()
+        for i in range(600):
+            loop.run_until_complete(adapter._on_inbound(self._event(f"ts.{i}")))
+        assert len(adapter._seen_inbound) <= adapter._SEEN_INBOUND_MAX