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

@@ -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)
 

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