diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 935924b50b..f6c60920e5 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -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: diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index cba40be83d..02526f9eac 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -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) diff --git a/tests/gateway/test_queued_final_ledger.py b/tests/gateway/test_queued_final_ledger.py index b9c345d973..24dc9fbd8c 100644 --- a/tests/gateway/test_queued_final_ledger.py +++ b/tests/gateway/test_queued_final_ledger.py @@ -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()