fix(gateway): keep overflow stream chunks editable

gateway/stream_consumer.py

@@ -744,39 +744,76 @@ async def run(self) -> None:
                         and self._message_id is None
                     ):
                         # No existing message to edit (first message or after a
-                        # segment break).  Use truncate_message — the same
-                        # helper the non-streaming path uses — to split with
-                        # proper word/code-fence boundaries and chunk
-                        # indicators like "(1/2)".
-                        chunks = self.adapter.truncate_message(
-                            self._accumulated, _safe_limit, len_fn=_len_fn,
+                        # segment break).  Seal only the overflowing head chunks
+                        # as fixed messages, then keep the trailing chunk in
+                        # _accumulated so the normal send/edit path below makes
+                        # it the active preview.  That lets chunk 2, 3, ... keep
+                        # updating in-place as later streamed deltas arrive
+                        # instead of posting every split as an immutable message.
+                        chunks = self._truncate_for_stream(
+                            self._accumulated, _safe_limit, _len_fn,
                         )
+                        if len(chunks) <= 1:
+                            # A malformed/legacy adapter result must not leave
+                            # this overflow branch with an unsplittable payload.
+                            chunks = self._split_text_chunks(
+                                self._accumulated, _safe_limit, _len_fn,
+                            )
                         chunks_delivered = False
-                        reply_to = self._message_id or self._initial_reply_to_id
-                        for chunk in chunks:
+                        reply_to = self._initial_reply_to_id
+                        all_heads_delivered = len(chunks) > 1
+                        for chunk in chunks[:-1]:
                             new_id = await self._send_new_chunk(
                                 chunk,
                                 reply_to,
                                 final=got_done,
                             )
-                            if new_id is not None and new_id != reply_to:
-                                chunks_delivered = True
-                        self._accumulated = ""
-                        self._last_sent_text = ""
+                            if new_id is None or new_id == reply_to:
+                                # Failed to deliver a sealed head; keep the
+                                # full accumulated text intact so the gateway's
+                                # fallback path can still deliver it completely.
+                                all_heads_delivered = False
+                                chunks_delivered = False
+                                break
+                            chunks_delivered = True
+                            reply_to = new_id
+
+                        if all_heads_delivered:
+                            self._accumulated = chunks[-1]
+                            # The head chunks are sealed.  Clear the edit target
+                            # so the remaining tail is sent as a fresh active
+                            # chunk, then edited by subsequent deltas.
+                            self._message_id = None
+                            self._message_created_ts = None
+                            self._last_sent_text = ""
+                        else:
+                            # A prior head may have landed before a later head
+                            # failed.  Do not edit that sealed message with the
+                            # unsplit full payload; let the fallback path retry.
+                            self._message_id = None
+                            self._message_created_ts = None
+                            self._last_sent_text = ""
+
                         self._last_edit_time = time.monotonic()
                         if got_done:
-                            # Only claim final delivery if THESE chunks actually
-                            # landed.  ``_already_sent`` may be True from prior
-                            # tool-progress edits or fallback-mode promotion (#10748)
-                            # — that doesn't mean the final answer reached the user.
-                            self._final_response_sent = chunks_delivered
-                            if chunks_delivered:
+                            tail_delivered = True
+                            if self._accumulated:
+                                tail_delivered = await self._send_or_edit(
+                                    self._accumulated, finalize=True,
+                                )
+                            # Only claim final delivery if the sealed chunks and
+                            # final tail actually landed.  ``_already_sent`` may
+                            # be True from prior progress/fallback state (#10748).
+                            self._final_response_sent = chunks_delivered and tail_delivered
+                            if self._final_response_sent:
                                 self._final_content_delivered = True
                             return
                         if got_segment_break:
                             self._message_id = None
                             self._fallback_final_send = False
                             self._fallback_prefix = ""
+                            if not self._accumulated:
+                                continue
 
                         # This iteration consumed a _FLUSH barrier and delivered
                         # the buffered prose via the chunk loop above, then takes
@@ -797,8 +834,8 @@ async def run(self) -> None:
                             self._accumulated, _safe_limit, _len_fn,
                         )
                         split_at = self._accumulated.rfind("\n", 0, _cp_budget)
-                        if split_at < _safe_limit // 2:
-                            split_at = _safe_limit
+                        if split_at < _cp_budget // 2:
+                            split_at = _cp_budget
                         chunk = self._accumulated[:split_at]
                         # finalize=True so the adapter applies platform-specific
                         # rich-text markup (e.g. Telegram MarkdownV2). This
@@ -1076,7 +1113,8 @@ def _continuation_text(self, final_text: str) -> str:
 
     @staticmethod
     def _split_text_chunks(
-        text: str, limit: int,
+        text: str,
+        limit: int,
         len_fn: "Callable[[str], int]" = len,
     ) -> list[str]:
         """Split text into reasonably sized chunks for fallback sends."""
@@ -1087,14 +1125,41 @@ def _split_text_chunks(
         while len_fn(remaining) > limit:
             _cp_budget = _custom_unit_to_cp(remaining, limit, len_fn)
             split_at = remaining.rfind("\n", 0, _cp_budget)
-            if split_at < limit // 2:
-                split_at = limit
+            if split_at < _cp_budget // 2:
+                split_at = _cp_budget
             chunks.append(remaining[:split_at])
             remaining = remaining[split_at:].lstrip("\n")
         if remaining:
             chunks.append(remaining)
         return chunks
 
+    def _truncate_for_stream(
+        self,
+        text: str,
+        limit: int,
+        len_fn: "Callable[[str], int]",
+    ) -> list[str]:
+        """Use the adapter's canonical splitter for streaming overflow.
+
+        Platform adapters may add word-boundary, code-fence, table, or
+        platform-specific formatting rules.  The consumer must not replace
+        those rules with newline-only slicing.  Non-base test doubles and
+        legacy adapters retain the historical two-argument call shape.
+        """
+        truncate = getattr(self.adapter, "truncate_message", None)
+        if not callable(truncate):
+            return self._split_text_chunks(text, limit, len_fn)
+
+        if isinstance(self.adapter, _BasePlatformAdapter):
+            chunks = truncate(text, limit, len_fn=len_fn)
+        else:
+            chunks = truncate(text, limit)
+        if not isinstance(chunks, (list, tuple)) or not all(
+            isinstance(chunk, str) for chunk in chunks
+        ):
+            return self._split_text_chunks(text, limit, len_fn)
+        return list(chunks)
+
     async def _send_fallback_final(self, text: str) -> None:
         """Send the final continuation after streaming edits stop working.
 

tests/gateway/test_stream_consumer.py

@@ -1183,6 +1183,105 @@ async def fake_send(*, chat_id, content, **kwargs):
         )
 
 
+class TestInitialOverflowRollingEdit:
+    @pytest.mark.asyncio
+    async def test_initial_overflow_keeps_last_chunk_as_edit_target(self):
+        """When the first visible flush already overflows, only sealed head
+        chunks should be posted as fixed messages.  The trailing chunk must
+        remain the active edit target so later streamed deltas update that
+        second message instead of overwriting or posting a new one."""
+        adapter = MagicMock()
+        msg_ids = iter(["msg_1", "msg_2"])
+        adapter.send = AsyncMock(
+            side_effect=lambda **kw: SimpleNamespace(
+                success=True,
+                message_id=next(msg_ids),
+            )
+        )
+        adapter.edit_message = AsyncMock(
+            return_value=SimpleNamespace(success=True, message_id="msg_2"),
+        )
+        adapter.MAX_MESSAGE_LENGTH = 700
+
+        config = StreamConsumerConfig(
+            edit_interval=0.01,
+            buffer_threshold=5,
+            cursor=" ▉",
+        )
+        consumer = GatewayStreamConsumer(adapter, "chat_123", config)
+
+        head = "A" * 650
+        tail = "B" * 25
+        consumer.on_delta(head)
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.08)
+        consumer.on_delta(tail)
+        await asyncio.sleep(0.08)
+        consumer.finish()
+        await task
+
+        assert adapter.send.call_count == 2
+        assert adapter.edit_message.call_count >= 1
+        edited_texts = [call.kwargs["content"] for call in adapter.edit_message.call_args_list]
+        assert any("A" * 20 in text and tail in text for text in edited_texts), (
+            "the second overflow chunk should be edited with its existing tail "
+            "plus later deltas, not overwritten by only the later delta"
+        )
+        assert consumer.final_response_sent is True
+
+    @pytest.mark.asyncio
+    async def test_initial_overflow_uses_adapter_fence_aware_split(self):
+        """Initial rolling sends must preserve the adapter's fence contract."""
+        adapter = TestUtf16OverflowDetection()._make_telegram_like_adapter()
+        from gateway.platforms.base import utf16_len
+
+        msg_ids = iter(["msg_1", "msg_2", "msg_3"])
+        adapter.send = AsyncMock(
+            side_effect=lambda **kw: SimpleNamespace(
+                success=True,
+                message_id=next(msg_ids),
+            )
+        )
+        adapter.edit_message = AsyncMock(
+            return_value=SimpleNamespace(success=True, message_id="msg_3"),
+        )
+        raw_limit = 700
+        setattr(adapter, "MAX_MESSAGE_LENGTH", raw_limit)
+        splitter = MagicMock(side_effect=adapter.truncate_message)
+        adapter.truncate_message = splitter
+
+        config = StreamConsumerConfig(
+            edit_interval=0.01,
+            buffer_threshold=5,
+            cursor=" ▉",
+        )
+        consumer = GatewayStreamConsumer(adapter, "chat_fenced", config)
+        fenced = "```python\n" + ("print('x')\n" * 100) + "```"
+        safe_limit = raw_limit - utf16_len(config.cursor) - 100
+        expected_chunks = adapter.truncate_message(
+            fenced, safe_limit, len_fn=adapter.message_len_fn,
+        )
+        splitter.reset_mock()
+
+        consumer.on_delta(fenced)
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.08)
+        consumer.on_delta("\nTail after the fenced stream.")
+        await asyncio.sleep(0.08)
+        consumer.finish()
+        await task
+
+        sent_texts = [call.kwargs["content"] for call in adapter.send.call_args_list]
+        edited_texts = [call.kwargs["content"] for call in adapter.edit_message.call_args_list]
+        assert splitter.call_count >= 1
+        assert all(text.count("```") % 2 == 0 for text in sent_texts + edited_texts)
+        assert len(sent_texts) == len(expected_chunks)
+        assert sent_texts[:-1] == expected_chunks[:-1]
+        assert sent_texts[-1].startswith(expected_chunks[-1])
+        assert any("Tail after the fenced stream." in text for text in edited_texts)
+        assert all(utf16_len(text) <= safe_limit for text in sent_texts)
+
+
 class TestEditOverflowSplitAndDeliver:
     """When edit_message split-and-delivers an oversized payload across the
     original message + N continuations (Telegram >4096 UTF-16), the consumer
@@ -1985,11 +2084,6 @@ async def test_emoji_text_exceeding_utf16_limit_triggers_overflow_split(self):
         adapter.edit_message = AsyncMock(
             return_value=SimpleNamespace(success=True),
         )
-        # truncate_message: emit two halves so we can assert the split fired
-        adapter.truncate_message = MagicMock(
-            side_effect=lambda text, limit, **kw: [text[:len(text)//2], text[len(text)//2:]],
-        )
-
         config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5)
         consumer = GatewayStreamConsumer(adapter, "chat_123", config)
 
@@ -2010,17 +2104,17 @@ async def test_emoji_text_exceeding_utf16_limit_triggers_overflow_split(self):
         consumer.finish()
         await task
 
-        # The fix: stream consumer detects UTF-16 overflow and calls
-        # truncate_message to split. Without the fix, len() would return
-        # 2200 (under 4096) and no split would fire — Telegram would then
-        # reject the send or render \x00 artifacts.
-        adapter.truncate_message.assert_called(), (
+        # The fix: stream consumer detects UTF-16 overflow using the adapter's
+        # length function.  Without that, len() would return 2200 (under the
+        # limit) and Hermes would attempt a single over-limit Telegram send.
+        sent_texts = [call.kwargs["content"] for call in adapter.send.call_args_list]
+        assert len(sent_texts) == 2, (
             "UTF-16 overflow not detected — emoji text bypassed split path"
         )
-        # truncate_message must have been called with len_fn=utf16_len
-        call_kwargs = adapter.truncate_message.call_args[1]
-        assert call_kwargs.get("len_fn") is utf16_len, (
-            f"truncate_message called without utf16_len: {call_kwargs}"
+        max_units = 4096
+        assert all(utf16_len(text) <= max_units for text in sent_texts), (
+            f"split chunks still exceed Telegram UTF-16 limit: "
+            f"{[utf16_len(text) for text in sent_texts]}"
         )
 
     def test_codepoint_only_adapter_falls_back_to_len(self):