Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 17 additions & 2 deletions gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -9037,6 +9037,8 @@ def _promote_queued_event(
session_key: str,
adapter: Any,
pending_event: Optional["MessageEvent"],
*,
defer_if_recursion_limit: bool = False,
) -> Optional["MessageEvent"]:
"""Promote the next overflow item after the slot was drained.

Expand All @@ -9053,6 +9055,14 @@ def _promote_queued_event(
overflow = _q_state.conversation.queued_events if _q_state else None
if not overflow:
return pending_event
# The recursion guard below may refuse to process ``pending_event``.
# Do not pop the overflow head before that guard: if it is staged in
# the slot and the guard then requeues ``pending_event`` through the
# single-slot merge helper, the staged event is overwritten and lost.
# Leaving the overflow untouched lets the FIFO enqueue path restore
# the already-dequeued event ahead of the tail.
if defer_if_recursion_limit and pending_event is not None:
return pending_event
next_queued = overflow.pop(0)
if pending_event is None:
return next_queued
Expand Down Expand Up @@ -29538,7 +29548,12 @@ def _run_sync_with_timeout_lifecycle():
# occupied for the full FIFO chain, which (a) preserves
# order, and (b) causes any mid-chain /queue to correctly
# route to overflow rather than jumping the queue.
pending_event = self._promote_queued_event(session_key, adapter, pending_event)
pending_event = self._promote_queued_event(
session_key,
adapter,
pending_event,
defer_if_recursion_limit=_interrupt_depth >= self._MAX_INTERRUPT_DEPTH,
)
if result.get("interrupted") and not pending_event and result.get("interrupt_message"):
interrupt_message = result.get("interrupt_message")
if _is_control_interrupt_message(interrupt_message):
Expand Down Expand Up @@ -29634,7 +29649,7 @@ def _run_sync_with_timeout_lifecycle():
)
adapter = self._adapter_for_source(source)
if adapter and pending_event:
merge_pending_message_event(adapter._pending_messages, session_key, pending_event)
self._enqueue_fifo(session_key, pending_event, adapter)
elif adapter and hasattr(adapter, 'queue_message'):
adapter.queue_message(session_key, pending)
return result_holder[0] or {"final_response": response, "messages": history}
Expand Down
40 changes: 40 additions & 0 deletions tests/gateway/test_queue_consumption.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
"""

import asyncio
from types import SimpleNamespace
from unittest.mock import MagicMock


Expand Down Expand Up @@ -169,6 +170,45 @@ def test_promote_stages_overflow_when_slot_already_populated(self):
# gets the next-in-line item.
assert adapter._pending_messages[session_key].text == "Q2"

def test_recursion_cap_requeues_dequeued_event_before_overflow(self):
"""The recursion guard must preserve the complete FIFO chain."""
from gateway.run import GatewayRunner

runner = GatewayRunner.__new__(GatewayRunner)
state = SimpleNamespace(conversation=SimpleNamespace(queued_events=[]))
runner._peek_session_state = lambda _key: state
runner._session_state = lambda _key: state
adapter = _StubAdapter()
session_key = "telegram:user:recursion-cap"

for text in ("Q1", "Q2", "Q3", "Q4"):
runner._enqueue_fifo(
session_key,
MessageEvent(
text=text,
message_type=MessageType.TEXT,
source=MagicMock(),
message_id=f"q-{text}",
),
adapter,
)

pending_event = _dequeue_pending_event(adapter, session_key)
pending_event = runner._promote_queued_event(
session_key,
adapter,
pending_event,
defer_if_recursion_limit=True,
)
runner._enqueue_fifo(session_key, pending_event, adapter)

assert adapter._pending_messages[session_key].text == "Q1"
assert [event.text for event in state.conversation.queued_events] == [
"Q2",
"Q3",
"Q4",
]


class TestBusyInputModeQueueFifo:
"""Regression coverage for issue #28503.
Expand Down
Loading