diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index 90d8d416be..9d20abfed4 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -317,12 +317,14 @@ class GatewayNotificationsMixin: self, response: str, source: SessionSource, adapter, metadata: Optional[Dict[str, Any]] = None, event_message_id: Optional[str] = None, text_already_delivered: bool = False, deliver_media: bool = True, stream_consumer=None, - session_key: Optional[str] = None, + session_key: Optional[str] = None, inbound_message_id: Optional[str] = None, ) -> None: """Deliver a queued response using the normal text+attachment split. ``session_key`` lets the text send record a delivery-ledger obligation like the normal final - send does (see ``_send_queued_final_text``); without it the send stays unledgered.""" + send does, keyed on ``inbound_message_id`` (the raw inbound id, distinct from the + ``event_message_id`` reply anchor); see ``_send_queued_final_text``. Without a key the send + stays unledgered.""" from gateway.run import _strip_response_attachments_for_direct_send if not text_already_delivered: text_content = _strip_response_attachments_for_direct_send(response, adapter) @@ -363,7 +365,8 @@ class GatewayNotificationsMixin: logger.debug("Queued-lane reconcile edit failed (%s); falling back to send.", _qe) if not _reconciled: await self._send_queued_final_text( - adapter, source, text_content, metadata, event_message_id, session_key) + adapter, source, text_content, metadata, event_message_id, session_key, + inbound_message_id) # Failed turns deliver their (normalized failure) text but must not upload attachments as if # they succeeded — mirrors the ``not agent_result.get("failed")`` completed-turn guard. if not deliver_media: @@ -376,27 +379,34 @@ class GatewayNotificationsMixin: async def _send_queued_final_text( self, adapter, source: SessionSource, text_content: str, metadata: Optional[Dict[str, Any]], event_message_id: Optional[str], session_key: Optional[str], + inbound_message_id: Optional[str] = None, ): - """Send a queued-lane final through the adapter's ledger-bracketed final send. + """Send a queued-lane final under the delivery-ledger bracket the normal final send uses. - This lane used to call ``adapter.send`` bare and discard the result. A final refused here - (flood control, a transport that just died) was gone for good: no delivery-ledger row was ever - written, so neither the boot sweep nor the runtime redelivery could see it, and the queued - follow-up ran as if the answer had landed. The normal final send records the obligation before - the send and finalizes it from the result; this routes the queued lane through that same method - so both lanes recover the same way. The event carries the inbound message id, so the obligation - id matches the one the normal lane would compute for the same turn (idempotent re-record), and - the reply is marked notify-worthy like every other user-visible final. Adapters without the - base contract (relay doubles, third-party plugins) and sends without a session key keep the - plain send. Returns the SendResult, or None when the adapter's send returned nothing - (queued-final-ledger).""" - bracketed = getattr(adapter, "_send_final_text", None) - if session_key and callable(bracketed): - results: list = [] - await bracketed( - MessageEvent(text="", source=source, message_id=event_message_id), session_key, - text_content, _mark_notify_metadata(metadata), False, 0, results.append) - result = results[-1] if results else None + This lane used to call ``adapter.send`` bare and discard the result, so a final refused here + (flood control, a transport that had just died) left no ledger row and was gone for good. + Now the obligation is recorded before the send and finalized from the result, the same + record / send-with-retry / finalize sequence as ``_send_final_text``, so both lanes recover + the same way. The ledger id is keyed on the raw inbound message id, as the normal lane's is; + ``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. Returns the + SendResult, or None when the adapter's send returned nothing.""" + record = getattr(adapter, "_record_delivery_obligation", None) + finalize = getattr(adapter, "_finalize_delivery_obligation", None) + transport_for = getattr(adapter, "_final_delivery_adapter", None) + if session_key and callable(record) and callable(finalize) and callable(transport_for): + delivery_adapter = transport_for(source) + ledger_message_id = ( + inbound_message_id if inbound_message_id is not None else event_message_id) + ledger_event = MessageEvent(text="", source=source, message_id=ledger_message_id) + obligation_id = await record( + ledger_event, session_key, text_content, delivery_adapter, False) + result = await delivery_adapter._send_with_retry( + chat_id=source.chat_id, content=text_content, reply_to=event_message_id, + metadata=_mark_notify_metadata(metadata)) + if obligation_id is not None: + await finalize(obligation_id, result, ledger_event, delivery_adapter) else: result = await adapter.send(source.chat_id, text_content, metadata=metadata) if not getattr(result, "success", False): diff --git a/gateway/run_turn.py b/gateway/run_turn.py index 25ab2612fd..7b1a84c78e 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -3383,8 +3383,9 @@ class GatewayTurnMixin: metadata=turn_ctx._status_thread_metadata, event_message_id=turn_ctx.event_message_id, text_already_delivered=_already_streamed, deliver_media=not _delivery_result.get("failed"), stream_consumer=_sc, - # The text send records a delivery-ledger obligation under this key (queued-final-ledger). - session_key=session_key, + # The text send records a delivery-ledger obligation under this key, keyed on + # the raw inbound id (the anchor above is only the reply target). + session_key=session_key, inbound_message_id=turn_ctx.inbound_message_id, ) except Exception as e: logger.warning("Failed to send first response before queued message: %s", e) diff --git a/tests/gateway/test_queued_final_ledger.py b/tests/gateway/test_queued_final_ledger.py index bea5513e39..cf0751ff3e 100644 --- a/tests/gateway/test_queued_final_ledger.py +++ b/tests/gateway/test_queued_final_ledger.py @@ -9,12 +9,14 @@ sweep nor the runtime redelivery could see it, and the follow-up ran as if the a had landed. On 5 Sep 2026 two long replies were lost this way within an hour, both refused by Telegram flood control while a delegation batch arrived the same second. -Now the queued lane routes its text send through the adapter's ``_send_final_text``, -the same ledger-bracketed method the normal lane uses: the obligation is recorded -before the send under the id the normal lane would compute for the same turn, the -result finalizes it, and the reply is marked notify-worthy like every other final. -The reconcile-by-edit path is unchanged. Adapters without the base contract and -sends without a session key keep the plain send. +Now the queued lane runs the same bracket as the normal lane: the obligation is +recorded before the send, the send goes through the adapter's retrying transport, and +the result finalizes the row. The obligation id is keyed on the raw inbound message id +exactly as the normal lane keys it; the reply anchor is only the reply target and is +None wherever replies are not used (Telegram forum topics), so it cannot identify the +turn. The reply is marked notify-worthy like every other final. The reconcile-by-edit +path is unchanged. Adapters without the base contract and sends without a session key +keep the plain send. """ from __future__ import annotations @@ -31,6 +33,7 @@ from gateway.config import Platform, PlatformConfig from gateway.platforms.base import BasePlatformAdapter, SendResult SESSION_KEY = "agent:main:telegram:dm:5230977008" +TOPIC_SESSION_KEY = "agent:main:telegram:group:-1001:topic:7" CHAT = "5230977008" INBOUND_ID = "5301" TEXT = "**Yes.** I would make room for one island trip and one Balkan trip." @@ -46,17 +49,19 @@ def _fresh_db(tmp_path, monkeypatch): yield +_COLUMNS = ("obligation_id", "session_key", "state", "attempts", "last_error", "content", "chat_id", + "platform") + + def _rows(): with dl._connect() as conn: - cur = conn.execute( - "SELECT obligation_id, session_key, state, attempts, last_error, content, chat_id, platform" - " FROM delivery_obligations") - return [dict(zip(("obligation_id", "session_key", "state", "attempts", "last_error", "content", - "chat_id", "platform"), r)) for r in cur.fetchall()] + cur = conn.execute(f"SELECT {', '.join(_COLUMNS)} FROM delivery_obligations") + return [dict(zip(_COLUMNS, r)) for r in cur.fetchall()] -def _source(): - return SimpleNamespace(platform=Platform.TELEGRAM, chat_id=CHAT, thread_id=None, chat_type="dm") +def _source(*, chat_id=CHAT, thread_id=None, chat_type="dm"): + return SimpleNamespace(platform=Platform.TELEGRAM, chat_id=chat_id, thread_id=thread_id, + chat_type=chat_type) def _telegram_adapter(send_result=None): @@ -69,7 +74,8 @@ def _telegram_adapter(send_result=None): runner._schedule_flood_redelivery = MagicMock(return_value=187.0) adapter.gateway_runner = runner adapter.send = AsyncMock(return_value=send_result or SendResult(success=True, message_id="900")) - adapter.edit_message = AsyncMock(return_value=SendResult(success=False, error="message to edit not found")) + adapter.edit_message = AsyncMock( + return_value=SendResult(success=False, error="message to edit not found")) return adapter @@ -86,13 +92,14 @@ def _runner(): return object.__new__(GatewayRunner) -async def _deliver(adapter, *, session_key=SESSION_KEY, stream_consumer=None, metadata=None, text=TEXT): +async def _deliver(adapter, *, session_key=SESSION_KEY, stream_consumer=None, metadata=None, + text=TEXT, source=None, anchor=INBOUND_ID, inbound_id=INBOUND_ID): from gateway.run import GatewayRunner await GatewayRunner._deliver_queued_first_response( - _runner(), text, source=_source(), adapter=adapter, metadata=metadata, event_message_id=INBOUND_ID, - text_already_delivered=False, deliver_media=False, stream_consumer=stream_consumer, - session_key=session_key) + _runner(), text, source=source or _source(), adapter=adapter, metadata=metadata, + event_message_id=anchor, text_already_delivered=False, deliver_media=False, + stream_consumer=stream_consumer, session_key=session_key, inbound_message_id=inbound_id) # --------------------------------------------------------------------------- @@ -114,6 +121,23 @@ async def test_a_delivered_queued_final_is_recorded_and_marked_delivered(): adapter.send.assert_awaited_once() +@pytest.mark.asyncio +async def test_the_row_is_already_attempting_while_the_send_is_in_flight(): + """Record BEFORE the send: a crash mid-send must leave a row the next boot can redeliver.""" + adapter = _telegram_adapter() + seen: list = [] + + async def _send(**kwargs): + seen.append([(r["state"], r["content"]) for r in _rows()]) + return SendResult(success=True, message_id="900") + + adapter.send = AsyncMock(side_effect=_send) + + await _deliver(adapter) + + assert seen == [[("attempting", TEXT)]] + + @pytest.mark.asyncio async def test_the_obligation_id_is_the_one_the_normal_lane_would_use(): """Same turn, same text: re-recording from either lane is idempotent, never a second row.""" @@ -124,9 +148,29 @@ async def test_the_obligation_id_is_the_one_the_normal_lane_would_use(): assert _rows()[0]["obligation_id"] == dl.compute_obligation_id(SESSION_KEY, INBOUND_ID, TEXT) +@pytest.mark.asyncio +async def test_forum_topic_turns_are_identified_by_their_inbound_id_not_the_reply_anchor(): + """Telegram forum topics route by topic metadata and never reply, so the anchor is None for + every message in the topic. Two turns answering with the same text must still get two rows, + or the second would overwrite the first's outstanding obligation and retry state.""" + adapter = _telegram_adapter() + topic = _source(chat_id="-1001", thread_id="7", chat_type="supergroup") + + await _deliver(adapter, session_key=TOPIC_SESSION_KEY, source=topic, anchor=None, + inbound_id="5301") + await _deliver(adapter, session_key=TOPIC_SESSION_KEY, source=topic, anchor=None, + inbound_id="5302") + + ids = sorted(r["obligation_id"] for r in _rows()) + assert ids == sorted([dl.compute_obligation_id(TOPIC_SESSION_KEY, "5301", TEXT), + dl.compute_obligation_id(TOPIC_SESSION_KEY, "5302", TEXT)]) + assert [c.kwargs["reply_to"] for c in adapter.send.await_args_list] == [None, None] + + @pytest.mark.asyncio async def test_a_flood_refused_queued_final_stays_in_the_ledger_as_failed(caplog): - """The row survives the refusal, so the sweeps (and a flood timer, where present) can redeliver.""" + """The row survives the refusal, so the sweeps (and a flood timer, where present) can redeliver + it.""" adapter = _telegram_adapter(SendResult(success=False, error="flood_control:185.0")) with caplog.at_level(logging.WARNING, logger="gateway.run"): @@ -136,13 +180,14 @@ async def test_a_flood_refused_queued_final_stays_in_the_ledger_as_failed(caplog assert len(rows) == 1 assert rows[0]["state"] == "failed" assert rows[0]["last_error"] == "flood_control:185.0" - assert any("Queued-lane final send" in r.getMessage() and "flood_control:185.0" in r.getMessage() - for r in caplog.records) + assert any("Queued-lane final send" in r.getMessage() + and "flood_control:185.0" in r.getMessage() for r in caplog.records) @pytest.mark.asyncio async def test_any_other_refusal_is_recorded_as_failed_too(): - adapter = _telegram_adapter(SendResult(success=False, error="Forbidden: bot was blocked by the user")) + adapter = _telegram_adapter( + SendResult(success=False, error="Forbidden: bot was blocked by the user")) await _deliver(adapter) @@ -228,11 +273,11 @@ async def test_a_plain_send_failure_is_still_logged(caplog): # --------------------------------------------------------------------------- -# The call site hands the session key over. +# The call site hands the session key and the raw inbound id over. # --------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_the_queued_lane_receives_the_session_key_from_the_turn(): +async def test_the_queued_lane_receives_the_session_key_and_inbound_id_from_the_turn(): from gateway.run import GatewayRunner runner = _runner() @@ -240,9 +285,12 @@ async def test_the_queued_lane_receives_the_session_key_from_the_turn(): runner._is_intentional_silence = MagicMock(return_value=False) runner._pop_post_delivery_callback = MagicMock(return_value=None) runner._deliver_queued_first_response = AsyncMock() + # A forum-topic turn: no reply anchor, but a raw inbound id the ledger must be keyed on. turn_ctx = SimpleNamespace( - session_key=SESSION_KEY, stream_consumer_holder=[None], source=_source(), - _status_thread_metadata={"thread_id": None}, event_message_id=INBOUND_ID, run_generation=3) + session_key=TOPIC_SESSION_KEY, stream_consumer_holder=[None], + source=_source(chat_id="-1001", thread_id="7", chat_type="supergroup"), + _status_thread_metadata={"thread_id": "7"}, event_message_id=None, + inbound_message_id=INBOUND_ID, run_generation=3) adapter = _telegram_adapter() await GatewayRunner._run_agent_deliver_first_response( @@ -250,6 +298,7 @@ async def test_the_queued_lane_receives_the_session_key_from_the_turn(): runner._deliver_queued_first_response.assert_awaited_once() kwargs = runner._deliver_queued_first_response.await_args.kwargs - assert kwargs["session_key"] == SESSION_KEY - assert kwargs["event_message_id"] == INBOUND_ID + assert kwargs["session_key"] == TOPIC_SESSION_KEY + assert kwargs["inbound_message_id"] == INBOUND_ID + assert kwargs["event_message_id"] is None assert kwargs["text_already_delivered"] is False