From ec4eecd864925fa195a9cf3651aff5d52c72849f Mon Sep 17 00:00:00 2001 From: Alexander Russell Date: Sat, 5 Sep 2026 21:28:07 +0300 Subject: [PATCH] fix(gateway): carry the raw inbound id through a chained queued turn Round-2 review. _run_agent_deliver_first_response passed the turn's inbound id to the queued lane, but the recursive _run_agent for a chained follow-up did not, so the chained turn ran with inbound_message_id=None. In a Telegram forum topic (no reply anchor) two chained follow-ups answering with the same text would then key their queued-final obligations on None and collide. The recursive call now carries pending_event's raw message id, with a test on the chained path. --- gateway/run_turn.py | 9 ++++-- tests/gateway/test_queued_final_ledger.py | 34 +++++++++++++++++++++++ 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/gateway/run_turn.py b/gateway/run_turn.py index 7b1a84c78e..4eb8f1e0bf 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -3439,6 +3439,10 @@ class GatewayTurnMixin: next_source, next_message, next_session_key = source, pending, session_key # message_type is carried into the recursive call so queued voice turns can stream TTS. next_message_id = next_channel_prompt = next_message_type = None + # The raw inbound id keys the delivery-ledger obligation for the follow-up's own final send, + # distinct from the reply anchor above (None in forum topics). Carry it or two chained + # topic turns with the same text would collide on one obligation id (queued-final-ledger). + next_inbound_id = None # See #60671. if pending_event is not None: next_source = getattr(pending_event, "source", None) or source @@ -3463,6 +3467,7 @@ class GatewayTurnMixin: if next_message is None: return result next_message_id = self._reply_anchor_for_event(pending_event) + next_inbound_id = str(pending_event.message_id) if getattr(pending_event, "message_id", None) else None next_channel_prompt = getattr(pending_event, "channel_prompt", None) next_message_type = getattr(pending_event, "message_type", None) @@ -3498,8 +3503,8 @@ class GatewayTurnMixin: message=next_message, context_prompt=turn_ctx.context_prompt, history=updated_history, source=next_source, session_id=session_id, session_key=next_session_key, run_generation=run_generation, _interrupt_depth=_interrupt_depth + 1, - event_message_id=next_message_id, channel_prompt=next_channel_prompt, - message_type=next_message_type, + event_message_id=next_message_id, inbound_message_id=next_inbound_id, + channel_prompt=next_channel_prompt, message_type=next_message_type, ) return _preserve_queued_followup_history_offset(result, followup_result) diff --git a/tests/gateway/test_queued_final_ledger.py b/tests/gateway/test_queued_final_ledger.py index cf0751ff3e..9f52459fd4 100644 --- a/tests/gateway/test_queued_final_ledger.py +++ b/tests/gateway/test_queued_final_ledger.py @@ -302,3 +302,37 @@ async def test_the_queued_lane_receives_the_session_key_and_inbound_id_from_the_ assert kwargs["inbound_message_id"] == INBOUND_ID assert kwargs["event_message_id"] is None assert kwargs["text_already_delivered"] is False + + +@pytest.mark.asyncio +async def test_a_chained_queued_turn_carries_its_own_inbound_id(): + """A queued follow-up that itself queues another follow-up (recursive _run_agent) must pass the + inbound id on, or the chained turn's own queued final would key on None in a forum topic and + collide with a sibling. The recursive call carries pending_event's raw message id.""" + from gateway.run import GatewayRunner + + runner = _runner() + runner._MAX_INTERRUPT_DEPTH = 8 + runner._run_agent = AsyncMock(return_value={"final_response": "done", "messages": []}) + runner._run_agent_deliver_first_response = AsyncMock() + runner._is_goal_continuation_event = MagicMock(return_value=False) + runner._session_key_for_source = MagicMock(return_value=TOPIC_SESSION_KEY) + runner._prepare_profile_scoped_inbound_message_text = AsyncMock(return_value="the follow-up") + runner._reply_anchor_for_event = MagicMock(return_value=None) # forum topic: no anchor + runner._adapter_for_source = MagicMock(return_value=None) + runner._refresh_agent_cache_message_count = AsyncMock() + topic = _source(chat_id="-1001", thread_id="7", chat_type="supergroup") + turn_ctx = SimpleNamespace( + source=topic, session_id="sid", session_key=TOPIC_SESSION_KEY, run_generation=1, + _interrupt_depth=0, history=[], _status_thread_metadata={"thread_id": "7"}, + context_prompt=None, result_holder=[None]) + pending_event = SimpleNamespace( + source=topic, message_id="6002", channel_prompt=None, message_type=None) + + await GatewayRunner._run_agent_queued_followup( + runner, turn_ctx, adapter=None, pending="hi again", pending_event=pending_event, + response="resp", result={"interrupted": True, "messages": []}, stream_task=None) + + runner._run_agent.assert_awaited_once() + assert runner._run_agent.await_args.kwargs["inbound_message_id"] == "6002" + assert runner._run_agent.await_args.kwargs["event_message_id"] is None