diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py
index 3fa87d378c7..f509e00443e 100644
--- a/plugins/platforms/discord/adapter.py
+++ b/plugins/platforms/discord/adapter.py
@@ -2100,7 +2100,19 @@ class DiscordAdapter(BasePlatformAdapter):
         *,
         finalize: bool = False,
     ) -> SendResult:
-        """Edit a previously sent Discord message."""
+        """Edit a previously sent Discord message.
+
+        Discord caps single-message text at 2,000 chars.  Edits that grow
+        past this limit must NOT be silently truncated (the stream consumer
+        would believe the full reply was delivered and stop) and must NOT
+        return failure (the consumer would re-send and create a duplicate).
+
+        Mid-stream (``finalize=False``) we keep editing the original message
+        with a truncated preview — splitting mid-stream would move the edit
+        target to a continuation and the next accumulated-token tick would
+        re-split, looping forever (the Telegram #48648 lesson).  The complete
+        text is delivered when ``finalize=True`` via ``_edit_overflow_split``.
+        """
         if not self._client:
             return SendResult(success=False, error="Not connected")
         try:
@@ -2109,14 +2121,159 @@ class DiscordAdapter(BasePlatformAdapter):
                 channel = await self._client.fetch_channel(int(chat_id))
             msg = await channel.fetch_message(int(message_id))
             formatted = self.format_message(content)
+
+            # Pre-flight: oversized payload.  Final edits split-and-deliver;
+            # streaming edits truncate a one-message preview in place.
             if len(formatted) > self.MAX_MESSAGE_LENGTH:
-                formatted = formatted[:self.MAX_MESSAGE_LENGTH - 3] + "..."
-            await msg.edit(content=formatted)
+                if finalize:
+                    return await self._edit_overflow_split(
+                        channel, msg, message_id, content,
+                    )
+                formatted = self.truncate_message(
+                    formatted, self.MAX_MESSAGE_LENGTH,
+                )[0]
+
+            try:
+                await msg.edit(content=formatted)
+            except Exception as edit_err:
+                # Reactive split-and-deliver: format_message inflation (or a
+                # server-side rule change) can push the payload past 2,000
+                # even when the pre-flight check passed.  Discord reports this
+                # as "error code: 50035 ... Must be 2000 or fewer in length".
+                if self._is_length_overflow_error(edit_err):
+                    if finalize:
+                        return await self._edit_overflow_split(
+                            channel, msg, message_id, content,
+                        )
+                    # Mid-stream: truncate and retry in place (no split).
+                    truncated = self.truncate_message(
+                        formatted, self.MAX_MESSAGE_LENGTH,
+                    )[0]
+                    await msg.edit(content=truncated)
+                else:
+                    raise
             return SendResult(success=True, message_id=message_id)
         except Exception as e:  # pragma: no cover - defensive logging
             logger.error("[%s] Failed to edit Discord message %s: %s", self.name, message_id, e, exc_info=True)
             return SendResult(success=False, error=str(e))
 
+    @staticmethod
+    def _is_length_overflow_error(err: Exception) -> bool:
+        """True when a Discord edit/send failed because text exceeded 2,000.
+
+        Discord returns ``error code: 50035`` with a ``Must be 2000 or fewer
+        in length`` validation detail.  We match on the stable error code plus
+        the length phrasing so unrelated 50035 validation errors (e.g. a bad
+        reply reference) don't get mistaken for an overflow.
+        """
+        text = str(err).lower()
+        return "error code: 50035" in text and (
+            "2000 or fewer" in text or "fewer in length" in text
+        )
+
+    async def _edit_overflow_split(
+        self,
+        channel: Any,
+        msg: Any,
+        message_id: str,
+        content: str,
+    ) -> SendResult:
+        """Deliver an oversized final edit across message + continuations.
+
+        Edit the original ``message_id`` with chunk 1 (fence-aware, with the
+        usual ``(1/N)`` indicator), then send chunks 2..N as new messages each
+        threaded as a reply to the previous chunk so Discord groups them
+        visually.  Returns ``SendResult(success=True, message_id=<last-id>,
+        continuation_message_ids=(...))`` so the stream consumer keeps editing
+        the most recent visible message and can clean up every chunk on a
+        fresh-final.
+
+        On a mid-stream continuation send failure we still report success with
+        however many continuations landed AND a ``partial_overflow``
+        raw_response so the consumer can deliver the missing tail rather than
+        treating a clipped reply as complete — dropping chunks the user already
+        saw would be the worse outcome.  Only a first-chunk edit failure
+        returns ``success=False`` (a real adapter problem, not overflow).
+        """
+        formatted = self.format_message(content)
+        chunks = self.truncate_message(formatted, self.MAX_MESSAGE_LENGTH)
+        if len(chunks) <= 1:
+            # Defensive: caller's pre-flight should guarantee >1 chunk, but if
+            # not, just edit normally.
+            await msg.edit(content=chunks[0] if chunks else formatted)
+            return SendResult(success=True, message_id=message_id)
+
+        # Step 1 — edit the existing message with the first chunk.
+        try:
+            await msg.edit(content=chunks[0])
+        except Exception as e:
+            logger.error(
+                "[%s] Overflow split: first-chunk edit failed: %s",
+                self.name, e, exc_info=True,
+            )
+            return SendResult(success=False, error=str(e))
+
+        # Step 2 — send each remaining chunk threaded as a reply to the prior.
+        continuation_ids: list[str] = []
+        delivered = 1
+        prev_msg = msg
+        for chunk in chunks[1:]:
+            reference = None
+            if hasattr(prev_msg, "to_reference"):
+                try:
+                    reference = prev_msg.to_reference(fail_if_not_exists=False)
+                except Exception:
+                    reference = None
+            try:
+                sent = await channel.send(content=chunk, reference=reference)
+            except Exception as send_err:
+                # Drop the reply anchor and retry once — a deleted/expired
+                # anchor (10008) or system-message reply (50035) shouldn't lose
+                # the chunk.
+                logger.warning(
+                    "[%s] Overflow continuation send failed (%s); retrying without reply reference",
+                    self.name, send_err,
+                )
+                try:
+                    sent = await channel.send(content=chunk, reference=None)
+                except Exception as retry_err:
+                    logger.warning(
+                        "[%s] Overflow split: stopped at %d/%d chunks delivered: %s",
+                        self.name, delivered, len(chunks), retry_err,
+                    )
+                    last_id = continuation_ids[-1] if continuation_ids else message_id
+                    return SendResult(
+                        success=True,
+                        message_id=last_id,
+                        continuation_message_ids=tuple(continuation_ids),
+                        raw_response={
+                            "partial_overflow": True,
+                            "delivered_chunks": delivered,
+                            "total_chunks": len(chunks),
+                            "last_message_id": last_id,
+                            "continuation_message_ids": tuple(continuation_ids),
+                        },
+                    )
+            new_id = str(sent.id)
+            continuation_ids.append(new_id)
+            delivered += 1
+            prev_msg = sent
+
+        last_id = continuation_ids[-1] if continuation_ids else message_id
+        # Keep the history-backfill fast path pointed at the final visible
+        # chunk so a later non-streaming send threads below the full reply.
+        if not _looks_like_nonconversational_history_message(content):
+            self._last_self_message_id[str(channel.id)] = last_id
+        logger.debug(
+            "[%s] Overflow split delivered %d chunks; last_id=%s",
+            self.name, delivered, last_id,
+        )
+        return SendResult(
+            success=True,
+            message_id=last_id,
+            continuation_message_ids=tuple(continuation_ids),
+        )
+
     async def _send_file_attachment(
         self,
         chat_id: str,
diff --git a/tests/gateway/test_discord_edit_message_overflow.py b/tests/gateway/test_discord_edit_message_overflow.py
new file mode 100644
index 00000000000..e49717a2317
--- /dev/null
+++ b/tests/gateway/test_discord_edit_message_overflow.py
@@ -0,0 +1,343 @@
+"""Regression tests for Discord oversized edit_message split-and-deliver.
+
+Issue #27881 surfaced as silent truncation: ``edit_message`` clipped any
+formatted payload over Discord's 2,000-char cap to ``[:1997] + "..."`` and
+returned ``success=True``, so the stream consumer believed the full reply
+landed and the user lost everything past the boundary.
+
+The fix mirrors the proven Telegram contract (and its #48648 lesson):
+
+* **Mid-stream** (``finalize=False``) — never split.  A mid-stream split moves
+  the edit target to a continuation; the next accumulated-token tick re-edits
+  the full text into it and re-splits, looping forever.  We truncate a
+  one-message preview in place instead.
+* **Final** (``finalize=True``) — split-and-deliver: edit chunk 1 in place,
+  send chunks 2..N as reply-threaded continuations, return the LAST visible id
+  in ``message_id`` plus every continuation in ``continuation_message_ids``.
+"""
+
+import sys
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import PlatformConfig
+
+
+def _ensure_discord_mock():
+    if "discord" in sys.modules and hasattr(sys.modules["discord"], "__file__"):
+        return
+    discord_mod = MagicMock()
+    discord_mod.Intents.default.return_value = MagicMock()
+    discord_mod.Client = MagicMock
+    discord_mod.File = MagicMock
+    discord_mod.DMChannel = type("DMChannel", (), {})
+    discord_mod.Thread = type("Thread", (), {})
+    discord_mod.ForumChannel = type("ForumChannel", (), {})
+    ext_mod = MagicMock()
+    commands_mod = MagicMock()
+    commands_mod.Bot = MagicMock
+    ext_mod.commands = commands_mod
+    sys.modules.setdefault("discord", discord_mod)
+    sys.modules.setdefault("discord.ext", ext_mod)
+    sys.modules.setdefault("discord.ext.commands", commands_mod)
+
+
+_ensure_discord_mock()
+
+from plugins.platforms.discord.adapter import DiscordAdapter  # noqa: E402
+
+
+MAX = DiscordAdapter.MAX_MESSAGE_LENGTH  # 2000
+
+
+def _make_adapter():
+    return DiscordAdapter(PlatformConfig(enabled=True, token="***"))
+
+
+def _wire_channel(adapter, *, original_msg, send_side_effect=None):
+    """Wire a fake client whose channel returns ``original_msg`` on fetch and
+    records every ``channel.send`` call."""
+    sends = []
+
+    async def fake_send(*, content, reference=None):
+        sends.append({"content": content, "reference": reference})
+        if send_side_effect is not None:
+            res = send_side_effect(len(sends), content, reference)
+            if res is not None:
+                return res
+        return SimpleNamespace(id=9000 + len(sends))
+
+    channel = SimpleNamespace(
+        id=555,
+        fetch_message=AsyncMock(return_value=original_msg),
+        send=AsyncMock(side_effect=fake_send),
+    )
+    adapter._client = SimpleNamespace(
+        get_channel=lambda _cid: channel,
+        fetch_channel=AsyncMock(return_value=channel),
+    )
+    return channel, sends
+
+
+# --------------------------------------------------------------------------- #
+# Happy path — short edits unchanged
+# --------------------------------------------------------------------------- #
+
+
+class TestEditMessageHappyPath:
+    @pytest.mark.asyncio
+    async def test_short_edit_in_place(self):
+        adapter = _make_adapter()
+        edits = []
+        msg = SimpleNamespace(
+            id=42,
+            edit=AsyncMock(side_effect=lambda *, content: edits.append(content)),
+        )
+        channel, sends = _wire_channel(adapter, original_msg=msg)
+
+        result = await adapter.edit_message("555", "42", "short reply")
+
+        assert result.success is True
+        assert result.message_id == "42"
+        assert result.continuation_message_ids == ()
+        assert edits == ["short reply"]
+        assert sends == []  # no continuations for a short edit
+
+    @pytest.mark.asyncio
+    async def test_no_client_returns_failure(self):
+        adapter = _make_adapter()
+        adapter._client = None
+        result = await adapter.edit_message("555", "42", "x")
+        assert result.success is False
+
+
+# --------------------------------------------------------------------------- #
+# Mid-stream overflow — TRUNCATE, never split (the #48648 lesson)
+# --------------------------------------------------------------------------- #
+
+
+class TestMidStreamOverflowTruncates:
+    @pytest.mark.asyncio
+    async def test_oversized_streaming_edit_truncates_in_place(self):
+        adapter = _make_adapter()
+        edits = []
+        msg = SimpleNamespace(
+            id=42,
+            edit=AsyncMock(side_effect=lambda *, content: edits.append(content)),
+        )
+        channel, sends = _wire_channel(adapter, original_msg=msg)
+
+        big = "p" * 6000
+        result = await adapter.edit_message("555", "42", big, finalize=False)
+
+        # No split: the original message stays the target, no continuations.
+        assert result.success is True
+        assert result.message_id == "42"
+        assert result.continuation_message_ids == ()
+        assert sends == [], "mid-stream overflow must NOT create continuations"
+        # Exactly one in-place edit, clipped to a single chunk under the cap.
+        assert len(edits) == 1
+        assert len(edits[0]) <= MAX
+        # No literal "..." truncation marker leaks in (fence-aware truncation).
+        assert not edits[0].endswith("...")
+
+
+# --------------------------------------------------------------------------- #
+# Final overflow — SPLIT and deliver every chunk
+# --------------------------------------------------------------------------- #
+
+
+class TestFinalOverflowSplits:
+    @pytest.mark.asyncio
+    async def test_oversized_final_edit_splits_and_delivers(self):
+        adapter = _make_adapter()
+        edits = []
+        msg = SimpleNamespace(
+            id=42,
+            to_reference=MagicMock(return_value=SimpleNamespace(kind="ref")),
+            edit=AsyncMock(side_effect=lambda *, content: edits.append(content)),
+        )
+        channel, sends = _wire_channel(adapter, original_msg=msg)
+
+        big = "q" * 6000  # ~3-4 chunks at 2000 cap
+        result = await adapter.edit_message("555", "42", big, finalize=True)
+
+        assert result.success is True
+        # message_id points at the LAST visible continuation, not the original.
+        assert result.continuation_message_ids, "expected continuations"
+        assert result.message_id == result.continuation_message_ids[-1]
+        # chunk 1 edited in place; chunks 2..N sent as new messages.
+        assert len(edits) == 1
+        assert len(sends) == len(result.continuation_message_ids)
+        # Every delivered chunk is under the cap.
+        for c in edits + [s["content"] for s in sends]:
+            assert len(c) <= MAX
+        # No "..." truncation marker anywhere.
+        for c in edits + [s["content"] for s in sends]:
+            assert not c.endswith("...")
+
+    @pytest.mark.asyncio
+    async def test_byte_coverage_preserved(self):
+        adapter = _make_adapter()
+        edits = []
+        msg = SimpleNamespace(
+            id=42,
+            to_reference=MagicMock(return_value=object()),
+            edit=AsyncMock(side_effect=lambda *, content: edits.append(content)),
+        )
+        channel, sends = _wire_channel(adapter, original_msg=msg)
+
+        # Distinctive marker at the very end must survive end-to-end.
+        body = "a" * 5000 + "END_MARKER_XYZ"
+        result = await adapter.edit_message("555", "42", body, finalize=True)
+
+        assert result.success is True
+        delivered = "".join(edits + [s["content"] for s in sends])
+        assert "END_MARKER_XYZ" in delivered
+
+    @pytest.mark.asyncio
+    async def test_continuations_threaded_as_replies(self):
+        adapter = _make_adapter()
+        msg = SimpleNamespace(
+            id=42,
+            to_reference=MagicMock(return_value=SimpleNamespace(tag="orig")),
+            edit=AsyncMock(),
+        )
+        # Each sent continuation must also expose to_reference so the NEXT
+        # chunk can thread under it.
+        channel, sends = _wire_channel(
+            adapter,
+            original_msg=msg,
+            send_side_effect=lambda n, content, ref: SimpleNamespace(
+                id=9000 + n,
+                to_reference=MagicMock(return_value=SimpleNamespace(tag=f"c{n}")),
+            ),
+        )
+
+        result = await adapter.edit_message("555", "42", "z" * 6000, finalize=True)
+
+        assert result.success is True
+        # First continuation replies to the original message's reference.
+        assert sends[0]["reference"] is not None
+        # Later continuations reply to the previous continuation, not None.
+        for s in sends[1:]:
+            assert s["reference"] is not None
+
+    @pytest.mark.asyncio
+    async def test_first_chunk_edit_failure_propagates(self):
+        adapter = _make_adapter()
+        msg = SimpleNamespace(
+            id=42,
+            to_reference=MagicMock(return_value=object()),
+            edit=AsyncMock(side_effect=RuntimeError("hard edit failure")),
+        )
+        channel, sends = _wire_channel(adapter, original_msg=msg)
+
+        result = await adapter.edit_message("555", "42", "w" * 6000, finalize=True)
+
+        assert result.success is False
+        assert "hard edit failure" in (result.error or "")
+        assert sends == []  # never reached the continuation loop
+
+    @pytest.mark.asyncio
+    async def test_mid_continuation_failure_reports_partial(self):
+        adapter = _make_adapter()
+        msg = SimpleNamespace(
+            id=42,
+            to_reference=MagicMock(return_value=object()),
+            edit=AsyncMock(),
+        )
+
+        # First continuation succeeds; second fails both with and without ref.
+        def side(n, content, ref):
+            if n == 1:
+                return SimpleNamespace(id=9001, to_reference=MagicMock(return_value=object()))
+            raise RuntimeError("continuation send failed")
+
+        channel, sends = _wire_channel(adapter, original_msg=msg, send_side_effect=side)
+
+        result = await adapter.edit_message("555", "42", "k" * 6000, finalize=True)
+
+        # Partial delivery still reports success (don't drop chunks the user
+        # already saw) but flags partial_overflow so the consumer retries tail.
+        assert result.success is True
+        assert result.raw_response["partial_overflow"] is True
+        assert result.raw_response["delivered_chunks"] < result.raw_response["total_chunks"]
+        assert result.message_id == "9001"
+
+
+# --------------------------------------------------------------------------- #
+# Reactive overflow — Discord 50035 mid-edit triggers the same branch logic
+# --------------------------------------------------------------------------- #
+
+
+class TestReactiveOverflowDetection:
+    @pytest.mark.asyncio
+    async def test_50035_on_final_edit_triggers_split(self):
+        adapter = _make_adapter()
+        edit_calls = []
+
+        # format_message leaves content under the cap, but the first edit
+        # raises 50035 (server-side rejection); the split path then runs.
+        def edit_effect(*, content):
+            edit_calls.append(content)
+            if len(edit_calls) == 1:
+                raise RuntimeError(
+                    "400 Bad Request (error code: 50035): Invalid Form Body\n"
+                    "In content: Must be 2000 or fewer in length."
+                )
+
+        msg = SimpleNamespace(
+            id=42,
+            to_reference=MagicMock(return_value=object()),
+            edit=AsyncMock(side_effect=edit_effect),
+        )
+        channel, sends = _wire_channel(adapter, original_msg=msg)
+
+        # Content is UNDER the cap so pre-flight passes; the 50035 on edit
+        # forces the reactive split.
+        result = await adapter.edit_message("555", "42", "u" * 1500, finalize=True)
+
+        assert result.success is True
+        # Reactive split re-edited chunk 1 and may add continuations.
+        assert len(edit_calls) >= 1
+
+    @pytest.mark.asyncio
+    async def test_unrelated_50035_is_not_treated_as_overflow(self):
+        adapter = _make_adapter()
+        msg = SimpleNamespace(
+            id=42,
+            edit=AsyncMock(side_effect=RuntimeError(
+                "400 Bad Request (error code: 50035): In message_reference: "
+                "Cannot reply to a system message"
+            )),
+        )
+        channel, sends = _wire_channel(adapter, original_msg=msg)
+
+        result = await adapter.edit_message("555", "42", "small", finalize=True)
+
+        # Not a length error → propagates as a normal failure, no split.
+        assert result.success is False
+        assert sends == []
+
+
+# --------------------------------------------------------------------------- #
+# Overflow detector helper
+# --------------------------------------------------------------------------- #
+
+
+class TestLengthOverflowDetector:
+    def test_matches_length_50035(self):
+        err = RuntimeError(
+            "error code: 50035 ... Must be 2000 or fewer in length."
+        )
+        assert DiscordAdapter._is_length_overflow_error(err) is True
+
+    def test_ignores_non_length_50035(self):
+        err = RuntimeError("error code: 50035: Cannot reply to a system message")
+        assert DiscordAdapter._is_length_overflow_error(err) is False
+
+    def test_ignores_other_errors(self):
+        assert DiscordAdapter._is_length_overflow_error(RuntimeError("timeout")) is False