fix(telegram): preserve reply context across message batches

gateway/platforms/base.py

@@ -3480,7 +3480,7 @@ def _is_queue_text_debounce_candidate(self, event: MessageEvent) -> bool:
         return result
 
     def _can_merge_text_debounce_events(self, existing: MessageEvent, event: MessageEvent) -> bool:
-        """Return True when two text debounce events came from the same sender."""
+        """Return True when sender and full reply context are identical."""
 
         def _identity(candidate: MessageEvent) -> tuple[str, ...] | None:
             source = getattr(candidate, "source", None)
@@ -3493,8 +3493,22 @@ def _identity(candidate: MessageEvent) -> tuple[str, ...] | None:
             if getattr(source, "chat_type", None) in {"dm", "private"} and getattr(source, "chat_id", None):
                 return (platform, "dm", str(source.chat_id))
             return None
+        def _reply_context(candidate: MessageEvent) -> tuple[Any, ...]:
+            return (
+                candidate.reply_to_message_id,
+                candidate.reply_to_text,
+                candidate.reply_to_author_id,
+                candidate.reply_to_author_name,
+                bool(candidate.reply_to_is_own_message),
+            )
+
         existing_sender = _identity(existing)
-        return existing_sender is not None and existing_sender == _identity(event)
+        incoming_sender = _identity(event)
+        return (
+            existing_sender is not None
+            and existing_sender == incoming_sender
+            and _reply_context(existing) == _reply_context(event)
+        )
 
     def _text_debounce_delay(self, session_key: str) -> float:
         """Return bounded busy-text debounce delay for ``session_key``."""
@@ -3527,11 +3541,10 @@ async def _queue_text_debounce(self, session_key: str, event: MessageEvent) -> N
             if event.text:
                 state.event.text = _append_text(state.event.text, event.text)
             latest_message_id = getattr(event, "message_id", None)
-            latest_anchor = latest_message_id or getattr(event, "reply_to_message_id", None)
             if latest_message_id is not None:
+                # Responses should anchor to the latest inbound message, while
+                # reply_to_* remains the user's original quote/author context.
                 state.event.message_id = str(latest_message_id)
-            if latest_anchor is not None and hasattr(state.event, "reply_to_message_id"):
-                state.event.reply_to_message_id = str(latest_anchor)
             state.last_ts = now
         state.cancel_timer()
         delay = self._text_debounce_delay(session_key)

plugins/platforms/telegram/adapter.py

@@ -6036,11 +6036,67 @@ def _text_batch_key(self, event: MessageEvent) -> str:
         self._apply_topic_recovery(event)
         return super()._text_batch_key(event)
 
+    @staticmethod
+    def _text_batch_reply_context(event: MessageEvent) -> tuple:
+        """Return every reply field whose meaning would spread across a batch."""
+        return (
+            event.reply_to_message_id,
+            event.reply_to_text,
+            event.reply_to_author_id,
+            event.reply_to_author_name,
+            bool(event.reply_to_is_own_message),
+        )
+
+    @classmethod
+    def _text_batch_has_reply_context(cls, event: MessageEvent) -> bool:
+        return any(
+            value not in (None, "", False)
+            for value in cls._text_batch_reply_context(event)
+        )
+
+    def _text_batch_context_compatible(
+        self,
+        existing: MessageEvent,
+        incoming: MessageEvent,
+    ) -> bool:
+        """Return whether coalescing preserves the reply/quote semantics.
+
+        Telegram may attach reply metadata only to the first near-limit chunk
+        of a client-split long message. A following metadata-free chunk can
+        inherit that first chunk's reply context.
+        """
+        if self._text_batch_reply_context(existing) == self._text_batch_reply_context(
+            incoming
+        ):
+            return True
+
+        existing_last_len = getattr(
+            existing,
+            "_last_chunk_len",
+            len(existing.text or ""),
+        )
+        return (
+            existing_last_len >= self._SPLIT_THRESHOLD
+            and not self._text_batch_has_reply_context(incoming)
+        )
+
     def _enqueue_text_event(self, event: MessageEvent) -> None:
         """Buffer a text chunk, or hold it while delayed delivery must be dropped."""
         if self._should_drop_delayed_delivery():
             self._hold_inbound_event(event, where="text-enqueue")
             return
+        key = self._text_batch_key(event)
+        existing = self._pending_text_batches.get(key)
+        if existing is not None and not self._text_batch_context_compatible(existing, event):
+            prior_task = self._pending_text_batch_tasks.pop(key, None)
+            if prior_task and not prior_task.done():
+                prior_task.cancel()
+            self._pending_text_batches.pop(key, None)
+            logger.info(
+                "[Telegram] Flushing text batch %s before incompatible reply context",
+                key,
+            )
+            self._hold_inbound_event(existing, where="text-reply-context-boundary")
         super()._enqueue_text_event(event)
 
     async def _flush_buffered(self, pending: dict, tasks: dict, key: str, delay: float, where: str, log_fn=None) -> None:

tests/gateway/test_active_session_text_merge.py

@@ -46,6 +46,11 @@ def _make_event(
     user_id: str = "u1",
     user_name: str | None = None,
     thread_id: str | None = None,
+    reply_to_message_id: str | None = None,
+    reply_to_text: str | None = None,
+    reply_to_author_id: str | None = None,
+    reply_to_author_name: str | None = None,
+    reply_to_is_own_message: bool = False,
 ) -> MessageEvent:
     source = SessionSource(
         platform=Platform.TELEGRAM,
@@ -60,6 +65,11 @@ def _make_event(
         message_type=MessageType.TEXT,
         source=source,
         message_id=f"msg-{text[:8]}",
+        reply_to_message_id=reply_to_message_id,
+        reply_to_text=reply_to_text,
+        reply_to_author_id=reply_to_author_id,
+        reply_to_author_name=reply_to_author_name,
+        reply_to_is_own_message=reply_to_is_own_message,
     )
 
 
@@ -261,3 +271,85 @@ def test_command_messages_bypass_debounce_even_in_queue_mode():
     assert not adapter._is_queue_text_debounce_candidate(_make_event("/stop"))
 
 
+@pytest.mark.asyncio
+async def test_queue_debounce_preserves_same_reply_context():
+    adapter = _make_adapter()
+    first = _make_event(
+        "one",
+        reply_to_message_id="reply-1",
+        reply_to_text="quoted",
+        reply_to_author_id="author-1",
+        reply_to_author_name="Author One",
+        reply_to_is_own_message=True,
+    )
+    session_key = build_session_key(first.source)
+    adapter._active_sessions[session_key] = asyncio.Event()
+
+    await adapter.handle_message(first)
+    await adapter.handle_message(
+        _make_event(
+            "two",
+            reply_to_message_id="reply-1",
+            reply_to_text="quoted",
+            reply_to_author_id="author-1",
+            reply_to_author_name="Author One",
+            reply_to_is_own_message=True,
+        )
+    )
+
+    merged = _debounced_event(adapter, session_key)
+    assert merged.text == "one\ntwo"
+    assert merged.message_id == "msg-two"
+    assert (
+        merged.reply_to_message_id,
+        merged.reply_to_text,
+        merged.reply_to_author_id,
+        merged.reply_to_author_name,
+        merged.reply_to_is_own_message,
+    ) == ("reply-1", "quoted", "author-1", "Author One", True)
+    adapter._discard_text_debounce(session_key)
+
+
+@pytest.mark.asyncio
+async def test_queue_debounce_splits_incompatible_reply_contexts():
+    adapter = _make_adapter()
+    first = _make_event(
+        "one",
+        reply_to_message_id="reply-1",
+        reply_to_text="first quote",
+        reply_to_author_id="author-1",
+        reply_to_author_name="Author One",
+    )
+    session_key = build_session_key(first.source)
+    adapter._active_sessions[session_key] = asyncio.Event()
+
+    await adapter.handle_message(first)
+    await adapter.handle_message(
+        _make_event(
+            "two",
+            reply_to_message_id="reply-2",
+            reply_to_text="second quote",
+            reply_to_author_id="author-2",
+            reply_to_author_name="Author Two",
+        )
+    )
+
+    pending = adapter._pending_messages[session_key]
+    queued = _debounced_event(adapter, session_key)
+    assert pending.text == "one"
+    assert (
+        pending.reply_to_message_id,
+        pending.reply_to_text,
+        pending.reply_to_author_id,
+        pending.reply_to_author_name,
+    ) == ("reply-1", "first quote", "author-1", "Author One")
+    assert queued.text == "two"
+    assert (
+        queued.reply_to_message_id,
+        queued.reply_to_text,
+        queued.reply_to_author_id,
+        queued.reply_to_author_name,
+    ) == ("reply-2", "second quote", "author-2", "Author Two")
+    adapter._discard_text_debounce(session_key)
+
+