fix(telegram): stripped-text fallbacks, re-finalize skip, and tail-only delete guard

Follow-ups on top of the two salvaged GodsBoy commits, all live-validated
against the real Telegram Bot API:

- _edit_overflow_split finalize fallbacks degrade to _strip_mdv2() clean
  text instead of putting raw **markdown** markers on screen (salvaged
  from PR #43463 minus its format-first sizing — live probes show
  Telegram's 4096 limit counts PARSED text, so MarkdownV2 escape
  inflation cannot cause MESSAGE_TOO_LONG and sizing against formatted
  wire length only causes premature splits and fragment messages).
- Skip the redundant requires-finalize edit after a got_done edit that
  split-and-delivered (salvaged from PR #43463): re-finalizing re-splits
  the full text into the adopted continuation and duplicates chunks.
- _send_fallback_final only deletes the stale partial message when the
  fallback re-sent the COMPLETE final text. When the prefix dedup sent
  only the missing tail, the partial IS the head of the answer; deleting
  it left users with only the second half of long responses (live-
  reproduced: flood-control during a long stream -> head deleted,
  ratio 0.54 of content visible). This is the third bug behind the
  'Telegram cut messages' reports and was present on main and both PRs.

tests/gateway/test_stream_consumer.py

@@ -794,9 +794,11 @@ class TestSegmentBreakOnToolBoundary:
         )
 
     @pytest.mark.asyncio
-    async def test_fallback_final_deletes_partial_after_chunks_succeed(self):
-        """After fallback chunks land, the frozen partial must be deleted so
-        the user sees only the complete response (#16668)."""
+    async def test_fallback_final_deletes_partial_after_full_resend(self):
+        """After fallback re-sends the COMPLETE response, the frozen partial
+        must be deleted so the user sees only the complete response (#16668).
+        Full resend happens when the visible prefix doesn't match the final
+        text (e.g. post-segment-break content, #10807)."""
         adapter = MagicMock()
         adapter.send = AsyncMock(
             return_value=SimpleNamespace(success=True, message_id="msg_new"),
@@ -810,14 +812,49 @@ class TestSegmentBreakOnToolBoundary:
         config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5)
         consumer = GatewayStreamConsumer(adapter, "chat_123", config)
 
-        # Seed the consumer as if it already edited a partial message that
-        # later got stuck (flood control etc.) — _message_id is the stale id.
+        # The stale partial shows pre-tool text that is NOT a prefix of the
+        # final response — fallback re-sends the complete final text.
+        consumer._message_id = "msg_partial"
+        consumer._last_sent_text = "Let me check that for you…"
+
+        await consumer._send_fallback_final("Working on it. Done!")
+
+        adapter.delete_message.assert_awaited_once_with("chat_123", "msg_partial")
+        assert consumer._final_response_sent is True
+
+    @pytest.mark.asyncio
+    async def test_fallback_final_keeps_partial_after_tail_only_send(self):
+        """When the fallback sends only the missing TAIL (visible prefix
+        matches the final text), the partial message IS the head of the
+        answer — deleting it would leave the user with only the last part
+        of the response (the 'model sent only the second half' bug)."""
+        adapter = MagicMock()
+        adapter.send = AsyncMock(
+            return_value=SimpleNamespace(success=True, message_id="msg_new"),
+        )
+        adapter.edit_message = AsyncMock(
+            return_value=SimpleNamespace(success=True),
+        )
+        adapter.delete_message = AsyncMock(return_value=None)
+        adapter.MAX_MESSAGE_LENGTH = 4096
+
+        config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5)
+        consumer = GatewayStreamConsumer(adapter, "chat_123", config)
+
+        # Visible partial is a true prefix of the final response — the
+        # fallback dedup sends only the tail.
         consumer._message_id = "msg_partial"
         consumer._last_sent_text = "Working on i"
 
         await consumer._send_fallback_final("Working on it. Done!")
 
-        adapter.delete_message.assert_awaited_once_with("chat_123", "msg_partial")
+        # Tail was sent...
+        sent_contents = [
+            c.kwargs.get("content", "") for c in adapter.send.call_args_list
+        ]
+        assert any("Done!" in s and "Working on i" not in s for s in sent_contents)
+        # ...and the head-bearing partial was NOT deleted.
+        adapter.delete_message.assert_not_awaited()
         assert consumer._final_response_sent is True
 
     @pytest.mark.asyncio

tests/gateway/test_stream_consumer_fresh_final.py

@@ -466,6 +466,81 @@ class TestCancelledBestEffortDeliveryFinalizes:
         assert consumer.final_content_delivered is True
 
 
+class TestGotDoneOverflowSplitNotRefinalized:
+    """A got_done finalize edit that split-and-delivered across continuation
+    messages must not be followed by the redundant requires-finalize edit.
+
+    After a split, the consumer adopts the last continuation as the live
+    message and the redundant finalize edit re-submits the FULL accumulated
+    text against it; the adapter pre-flights that into another overflow
+    split, editing chunk 1 over the continuation and re-sending the rest,
+    so the user sees duplicated chunks. The finalize signal was already
+    carried by the split edit itself.
+    """
+
+    def _consumer(self, adapter):
+        # High interval/threshold so the only edit is the got_done finalize.
+        return GatewayStreamConsumer(
+            adapter=adapter,
+            chat_id="chat",
+            config=StreamConsumerConfig(
+                edit_interval=10.0, buffer_threshold=10_000, cursor=" ▉",
+            ),
+        )
+
+    @pytest.mark.asyncio
+    async def test_split_finalize_edit_is_not_refinalized(self):
+        adapter = _make_adapter()
+        adapter.REQUIRES_EDIT_FINALIZE = True
+        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(
+            success=True,
+            message_id="cont_2",
+            continuation_message_ids=("cont_2",),
+        ))
+        consumer = self._consumer(adapter)
+        consumer.on_delta("oversize **markdown** final reply")
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.05)  # preview send lands; no interval edits
+        consumer.finish()
+        await task
+
+        finalize_edits = [
+            c for c in adapter.edit_message.call_args_list
+            if c.kwargs.get("finalize")
+        ]
+        assert len(finalize_edits) == 1, (
+            "split finalize edit must not be re-finalized; the redundant "
+            "edit re-splits the full text into the adopted continuation "
+            "and duplicates chunks on screen"
+        )
+        assert consumer.final_response_sent is True
+        assert consumer.final_content_delivered is True
+
+    @pytest.mark.asyncio
+    async def test_non_split_finalize_edit_still_gets_explicit_refinalize(self):
+        """The narrow fix must not regress the requires-finalize contract:
+        a normal (non-split) got_done edit is still followed by the
+        explicit finalize edit (#25010 semantics unchanged)."""
+        adapter = _make_adapter()
+        adapter.REQUIRES_EDIT_FINALIZE = True
+        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(
+            success=True, message_id="initial_preview",
+        ))
+        consumer = self._consumer(adapter)
+        consumer.on_delta("short final reply")
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.05)
+        consumer.finish()
+        await task
+
+        finalize_edits = [
+            c for c in adapter.edit_message.call_args_list
+            if c.kwargs.get("finalize")
+        ]
+        assert len(finalize_edits) == 2
+        assert consumer.final_response_sent is True
+
+
 class TestStreamConsumerConfigFreshFinalField:
     """The dataclass field must exist and default to 0 (disabled)."""