From af09c4033e10b2b9ed41d62ed2caf7409ecdd244 Mon Sep 17 00:00:00 2001 From: Alexander Russell Date: Sat, 5 Sep 2026 21:06:58 +0300 Subject: [PATCH] fix(gateway): key the queued-lane obligation on the raw inbound id The queued lane's ledger bracket used the reply anchor as the obligation's message reference. The anchor is None wherever replies are not used (Telegram forum topics, Slack reaction handoffs), so two turns in one topic answering with the same text shared an obligation id and the second record overwrote the first's outstanding row; the id also differed from the normal lane's. The lane now runs the adapter's record / send-with-retry / finalize sequence itself, keyed on turn_ctx.inbound_message_id like the normal lane, while the anchor stays the reply target. Tests cover the forum-topic identity, the row being `attempting` while the send is in flight, and the call site passing both ids. --- gateway/run_notifications.py | 54 ++++++----- gateway/run_turn.py | 5 +- tests/gateway/test_queued_final_ledger.py | 105 ++++++++++++++++------ 3 files changed, 112 insertions(+), 52 deletions(-) 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