diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 04f0c1deb50ed..5bcda6b32e088 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -674,29 +674,29 @@ class SendResult: def merge_pending_message_event( - pending_messages: Dict[str, MessageEvent], + pending_messages: Dict[str, list[MessageEvent]], session_key: str, event: MessageEvent, ) -> None: - """Store or merge a pending event for a session. + """Append or merge a pending event into the FIFO queue for a session. Photo bursts/albums often arrive as multiple near-simultaneous PHOTO - events. Merge those into the existing queued event so the next turn sees - the whole burst, while non-photo follow-ups still replace the pending - event normally. + events. Merge those into the *last* queued event so the next turn sees + the whole burst, while non-photo follow-ups are appended as separate + entries in the FIFO queue. """ - existing = pending_messages.get(session_key) + queue = pending_messages.setdefault(session_key, []) if ( - existing - and getattr(existing, "message_type", None) == MessageType.PHOTO + queue + and getattr(queue[-1], "message_type", None) == MessageType.PHOTO and event.message_type == MessageType.PHOTO ): - existing.media_urls.extend(event.media_urls) - existing.media_types.extend(event.media_types) + queue[-1].media_urls.extend(event.media_urls) + queue[-1].media_types.extend(event.media_types) if event.text: - existing.text = BasePlatformAdapter._merge_caption(existing.text, event.text) + queue[-1].text = BasePlatformAdapter._merge_caption(queue[-1].text, event.text) return - pending_messages[session_key] = event + queue.append(event) # Error substrings that indicate a transient *connection* failure worth retrying. @@ -747,7 +747,7 @@ def __init__(self, config: PlatformConfig, platform: Platform): # Track active message handlers per session for interrupt support # Key: session_key (e.g., chat_id), Value: (event, asyncio.Event for interrupt) self._active_sessions: Dict[str, asyncio.Event] = {} - self._pending_messages: Dict[str, MessageEvent] = {} + self._pending_messages: Dict[str, list[MessageEvent]] = {} # Background message-processing tasks spawned by handle_message(). # Gateway shutdown cancels these so an old gateway instance doesn't keep # working on a task after --replace or manual restarts. @@ -1463,7 +1463,7 @@ async def handle_message(self, event: MessageEvent) -> None: # Default behavior for non-photo follow-ups: interrupt the running agent logger.debug("[%s] New message while session %s is active — triggering interrupt", self.name, session_key) - self._pending_messages[session_key] = event + self._pending_messages.setdefault(session_key, []).append(event) # Signal the interrupt (the processing task checks this) self._active_sessions[session_key].set() return # Don't process now - will be handled after current task finishes @@ -1721,9 +1721,12 @@ def _record_delivery(result): ) # Check if there's a pending message that was queued during our processing - if session_key in self._pending_messages: - pending_event = self._pending_messages.pop(session_key) - logger.debug("[%s] Processing queued message from interrupt", self.name) + if session_key in self._pending_messages and self._pending_messages[session_key]: + pending_event = self._pending_messages[session_key].pop(0) + # Clean up empty queue + if not self._pending_messages[session_key]: + del self._pending_messages[session_key] + logger.debug("[%s] Processing queued message from interrupt (%d remaining in queue)", self.name, len(self._pending_messages.get(session_key, []))) # Clean up current session before processing pending if session_key in self._active_sessions: del self._active_sessions[session_key] @@ -1802,8 +1805,14 @@ def has_pending_interrupt(self, session_key: str) -> bool: return session_key in self._active_sessions and self._active_sessions[session_key].is_set() def get_pending_message(self, session_key: str) -> Optional[MessageEvent]: - """Get and clear any pending message for a session.""" - return self._pending_messages.pop(session_key, None) + """Pop the first pending message from the FIFO queue for a session.""" + queue = self._pending_messages.get(session_key) + if not queue: + return None + event = queue.pop(0) + if not queue: + del self._pending_messages[session_key] + return event def build_source( self, diff --git a/gateway/run.py b/gateway/run.py index df69a498c1530..1242221b59ef2 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2366,7 +2366,9 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: # Force-clean: remove the session lock regardless of agent state adapter = self.adapters.get(source.platform) if adapter and hasattr(adapter, 'get_pending_message'): - adapter.get_pending_message(_quick_key) # consume and discard + # Drain the entire FIFO queue + while adapter.get_pending_message(_quick_key): + pass self._pending_messages.pop(_quick_key, None) if _quick_key in self._running_agents: del self._running_agents[_quick_key] @@ -2387,7 +2389,9 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: # Clear any pending messages so the old text doesn't replay adapter = self.adapters.get(source.platform) if adapter and hasattr(adapter, 'get_pending_message'): - adapter.get_pending_message(_quick_key) # consume and discard + # Drain the entire FIFO queue + while adapter.get_pending_message(_quick_key): + pass self._pending_messages.pop(_quick_key, None) # Clean up the running agent entry so the reset handler # doesn't think an agent is still active. @@ -2409,7 +2413,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: source=event.source, message_id=event.message_id, ) - adapter._pending_messages[_quick_key] = queued_event + adapter._pending_messages.setdefault(_quick_key, []).append(queued_event) return "Queued for the next turn." # /model must not be used while the agent is running. @@ -2450,7 +2454,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: # agent starts. adapter = self.adapters.get(source.platform) if adapter: - adapter._pending_messages[_quick_key] = event + adapter._pending_messages.setdefault(_quick_key, []).append(event) return None if self._draining: if self._queue_during_drain_enabled():