fix(gateway): the ephemeral delete goes to the adapter that sent the final

Follow-up to #106316. send_final_ledgered resolved the live adapter internally and _send_final_text resolved it a second time for _schedule_ephemeral_delete; a reconnect between the two sent the delete to a transport that never owned result.message_id (the ownership rule _final_delivery_adapter documents). The bracket now returns (result, adapter). The queued lane carried the ledger identity through MessageEvent.message_id while the PR added ledger_message_id for exactly that; it now uses the typed field, and the ledger read is getattr-tolerant of duck-typed events (a missing attribute was swallowed as "ledger skipped").
2026-09-09 12:59:09 +05:30
parent 91433c8466
commit b530a482d8
3 changed files with 45 additions and 13 deletions
@@ -3654,7 +3654,9 @@ class BasePlatformAdapter(ABC):
            source = event.source
            # ``ledger_message_id`` wins when set: a queued chain's final answers the last message
            # of the chain, not the event that opened it (see ``MessageEvent.ledger_message_id``).
-            _ledger_id = event.ledger_message_id if event.ledger_message_id is not None else getattr(event, "message_id", "")
+            _ledger_id = getattr(event, "ledger_message_id", None)
+            if _ledger_id is None:
+                _ledger_id = getattr(event, "message_id", "")
            obligation_id = compute_obligation_id(
                session_key, str(_ledger_id or ""), text_content)
            await asyncio.to_thread(
@@ -3760,12 +3762,14 @@ class BasePlatformAdapter(ABC):
    async def send_final_ledgered(
        self, event: MessageEvent, session_key: str, text_content: str, metadata: Dict[str, Any], *,
        reply_to: Optional[str], is_ephemeral_response: bool = False,
-    ) -> "SendResult":
+    ) -> "tuple[SendResult, BasePlatformAdapter]":
        """The delivery-ledger bracket every final text goes through, on the CURRENT transport
        (a reconnect may have replaced this adapter): record the obligation before the send,
        send with retry, finalize from the result — so a refused final (flood control, a dead
        transport) leaves a ledger row the boot sweep / runtime redelivery can act on. ``event``
-        supplies the source and the ledger identity (``ledger_message_id`` or ``message_id``)."""
+        supplies the source and the ledger identity (``ledger_message_id`` or ``message_id``).
+        Returns the result with the adapter that sent it: that adapter owns ``result.message_id``
+        (an ephemeral delete must go to the same transport)."""
        delivery_adapter = self._final_delivery_adapter(event.source)
        logger.info("[%s] Sending response (%d chars) to %s", delivery_adapter.name,
                    len(text_content), event.source.chat_id)
@@ -3775,19 +3779,18 @@ class BasePlatformAdapter(ABC):
            chat_id=event.source.chat_id, content=text_content, reply_to=reply_to, metadata=metadata)
        if obligation_id is not None:
            await self._finalize_delivery_obligation(obligation_id, result, event, delivery_adapter)
-        return result
+        return result, delivery_adapter

    async def _send_final_text(
        self, event: MessageEvent, session_key: str, text_content: str, metadata: Dict[str, Any],
        is_ephemeral_response: bool, ephemeral_ttl: int, record_delivery: Callable) -> None:
        """Normal-lane final: the ledger bracket plus the message-id owner's ephemeral delete."""
-        result = await self.send_final_ledgered(
+        result, delivery_adapter = await self.send_final_ledgered(
            event, session_key, text_content, metadata,
            reply_to=_reply_anchor_for_event(event), is_ephemeral_response=is_ephemeral_response)
        record_delivery(result)
        if ephemeral_ttl and ephemeral_ttl > 0 and result.success and result.message_id:
-            self._final_delivery_adapter(event.source)._schedule_ephemeral_delete(
-                event.source.chat_id, result.message_id, ephemeral_ttl)
+            delivery_adapter._schedule_ephemeral_delete(event.source.chat_id, result.message_id, ephemeral_ttl)

    async def _notify_turn_error(self, event: MessageEvent, e: BaseException) -> Optional[dict]:
        """Tell the user a turn failed rather than leaving radio silence (last resort:
@@ -386,13 +386,12 @@ class GatewayNotificationsMixin:
        result, so a final refused here (flood control, a transport that had just died) left no
        ledger row and was gone for good. The ledger identity is the raw inbound message id;
        ``event_message_id`` is only the reply anchor, which is None wherever replies are not used
-        (Telegram forum topics, Slack reaction handoffs) and so cannot identify the turn. Adapters
-        without the base contract and sends without a session key keep the plain send."""
+        (Telegram forum topics, Slack reaction handoffs) and so cannot identify the turn; with no
+        inbound id the ledger falls back to the event's own (empty) message id. Adapters without
+        the base contract and sends without a session key keep the plain send."""
        if session_key and isinstance(adapter, BasePlatformAdapter):
-            ledger_message_id = (
-                inbound_message_id if inbound_message_id is not None else event_message_id)
-            result = await adapter.send_final_ledgered(
-                MessageEvent(text="", source=source, message_id=ledger_message_id),
+            result, _ = await adapter.send_final_ledgered(
+                MessageEvent(text="", source=source, ledger_message_id=inbound_message_id),
                session_key, text_content, _mark_notify_metadata(metadata), reply_to=event_message_id)
        else:
            result = await adapter.send(source.chat_id, text_content, metadata=metadata)
@@ -305,3 +305,33 @@ async def test_a_deeper_chain_keeps_the_innermost_inbound_id():
        response="resp", result={"interrupted": True, "messages": []}, stream_task=None)

    assert merged["queued_terminal_inbound_id"] == "6003"
+
+
+# ---------------------------------------------------------------------------
+# The adapter that sent the final owns its message id: the ephemeral delete goes there.
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_ephemeral_delete_targets_the_adapter_that_sent_the_final(tmp_path, monkeypatch):
+    """A reconnect between the send and the delete swaps the runner's live adapter; the delete
+    must still go to the transport that produced ``result.message_id``."""
+    from gateway.platforms.event import MessageEvent
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    sender = _telegram_adapter()
+    replacement = _telegram_adapter()
+    sender._schedule_ephemeral_delete = MagicMock()
+    replacement._schedule_ephemeral_delete = MagicMock()
+    live = {"adapter": sender}
+    sender.gateway_runner._adapter_for_source = MagicMock(side_effect=lambda _s: live["adapter"])
+
+    async def swap_then_send(*args, **kwargs):
+        live["adapter"] = replacement  # reconnect lands while the send is in flight
+        return SendResult(success=True, message_id="900")
+
+    sender.send = AsyncMock(side_effect=swap_then_send)
+    event = MessageEvent(text="hi", source=_source(), message_id=INBOUND_ID)
+    await sender._send_final_text(event, SESSION_KEY, TEXT, {}, False, 30, lambda _r: None)
+
+    sender._schedule_ephemeral_delete.assert_called_once_with(CHAT, "900", 30)
+    replacement._schedule_ephemeral_delete.assert_not_called()