diff --git a/contributors/emails/akitani@akitaninoMac-mini.local b/contributors/emails/akitani@akitaninoMac-mini.local
new file mode 100644
index 000000000000..37e10cd297b5
--- /dev/null
+++ b/contributors/emails/akitani@akitaninoMac-mini.local
@@ -0,0 +1 @@
+2001Y
diff --git a/contributors/emails/gonzalofrancoceballos@Gonzalos-Mac-mini.local b/contributors/emails/gonzalofrancoceballos@Gonzalos-Mac-mini.local
new file mode 100644
index 000000000000..059333aea848
--- /dev/null
+++ b/contributors/emails/gonzalofrancoceballos@Gonzalos-Mac-mini.local
@@ -0,0 +1 @@
+gonzalofrancoceballos
diff --git a/contributors/emails/mehrzad.karami@gmail.com b/contributors/emails/mehrzad.karami@gmail.com
new file mode 100644
index 000000000000..243d842cf783
--- /dev/null
+++ b/contributors/emails/mehrzad.karami@gmail.com
@@ -0,0 +1,2 @@
+mzkarami
+# PR #66204 contributor identity
diff --git a/contributors/emails/skywind5487@gmail.com b/contributors/emails/skywind5487@gmail.com
new file mode 100644
index 000000000000..75126e8d0202
--- /dev/null
+++ b/contributors/emails/skywind5487@gmail.com
@@ -0,0 +1 @@
+Skywind5487
diff --git a/contributors/emails/z23@users.noreply.github.com b/contributors/emails/z23@users.noreply.github.com
new file mode 100644
index 000000000000..3fded3ad4998
--- /dev/null
+++ b/contributors/emails/z23@users.noreply.github.com
@@ -0,0 +1 @@
+z23
diff --git a/contributors/emails/zhangk1985@gmail.com b/contributors/emails/zhangk1985@gmail.com
new file mode 100644
index 000000000000..2ca7af13813e
--- /dev/null
+++ b/contributors/emails/zhangk1985@gmail.com
@@ -0,0 +1 @@
+kylezh
diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py
index 96a728abd7c1..c44f39921b1d 100644
--- a/gateway/platforms/base.py
+++ b/gateway/platforms/base.py
@@ -5895,7 +5895,26 @@ def truncate_message(
 
             # Everything remaining fits in one final chunk
             if _len(prefix) + _len(remaining) <= max_length - INDICATOR_RESERVE:
-                chunks.append(prefix + remaining)
+                final_chunk = prefix + remaining
+                # Check fence balance: if carry_lang was set, the chunk
+                # starts with an opening fence.  Walk the remaining text
+                # to see if the code block was closed; if not, close it.
+                _final_in_code = carry_lang is not None
+                _final_lang = carry_lang or ""
+                if _final_in_code:
+                    for _line in remaining.split("\n"):
+                        _stripped = _line.strip()
+                        if _stripped.startswith("```"):
+                            if _final_in_code:
+                                _final_in_code = False
+                                _final_lang = ""
+                            else:
+                                _final_in_code = True
+                                _tag = _stripped[3:].strip()
+                                _final_lang = _tag.split()[0] if _tag else ""
+                    if _final_in_code:
+                        final_chunk += FENCE_CLOSE
+                chunks.append(final_chunk)
                 break
 
             # Find a natural split point (prefer newlines, then spaces).
diff --git a/gateway/run.py b/gateway/run.py
index e4c4248cc848..cf5436dbb795 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -13634,6 +13634,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
             if _show_reasoning_effective and response and not _intentional_silence:
                 last_reasoning = agent_result.get("last_reasoning")
                 if last_reasoning:
+                    from gateway.stream_consumer import escape_code_fences_for_display
                     # Collapse long reasoning to keep messages readable
                     lines = last_reasoning.strip().splitlines()
                     if len(lines) > 15:
@@ -13665,6 +13666,9 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                         )
                         response = f"> 💭 **Reasoning:**\n{_quoted}\n\n{response}"
                     else:
+                        # Escape ``` inside reasoning so inner fences don't
+                        # break the outer code block used to render it.
+                        display_reasoning = escape_code_fences_for_display(display_reasoning)
                         response = f"💭 **Reasoning:**\n```\n{display_reasoning}\n```\n\n{response}"
 
             # Runtime-metadata footer — only on the FINAL message of the turn.
diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py
index a269aa8198d3..e60e9117031f 100644
--- a/gateway/stream_consumer.py
+++ b/gateway/stream_consumer.py
@@ -42,13 +42,7 @@
 
 # Sentinel to signal the stream is complete
 _DONE = object()
-
-# Sentinel to signal a tool boundary — finalize current message and start a
-# new one so that subsequent text appears below tool progress messages.
 _NEW_SEGMENT = object()
-
-# Queue marker for a completed assistant commentary message emitted between
-# API/tool iterations (for example: "I'll inspect the repo first.").
 _COMMENTARY = object()
 
 # Queue marker for a synchronous flush barrier.  Enqueued as
@@ -61,6 +55,76 @@
 _FLUSH = object()
 
 
+def escape_code_fences_for_display(text: str) -> str:
+    """Escape triple-backtick markers so text can be safely wrapped
+    inside an outer ``` code block without breaking the fence.
+
+    When reasoning content contains ``` (e.g. the model quotes code
+    in its thinking), wrapping it in an outer ``` for display causes
+    the inner fence to break the outer block.  Solution: replace each
+    `` ``` `` with `` \\`\\`\\` `` before wrapping.
+
+    Returns:
+        The input text with each `` ``` `` replaced by `` \\`\\`\\` ``,
+        or the input unchanged if no triple-backticks are present.
+    """
+    if not isinstance(text, str) or "```" not in text:
+        return text
+    return text.replace("```", "\\`\\`\\`")
+
+
+def ensure_closed_code_fences(text: str) -> str:
+    """Append a closing `` ``` `` fence and/or `` ` `` if the text has
+    orphaned code-block or inline-code markers.
+
+    When model output is truncated mid-code-block (e.g. by token limits
+    or a finish_reason="length"), the resulting message has an unclosed
+    code fence.  On Discord, Slack, and other platforms this causes
+    everything after the orphaned fence to render as a single code block.
+    The same problem applies to inline-code spans closed by a single
+    backtick: an orphaned `` ` `` makes the remainder of the message
+    render as inline code.
+
+    Triple-backtick: count `` ``` `` occurrences.  If odd, append a
+    closing fence on its own line.  This is safe because nested
+    triple-backtick fences (e.g. a literal `` ``` `` inside a code block)
+    are exceedingly rare in model output and, when they do appear, the
+    extra closing fence just creates a brief empty code block at the end
+    of the message — far less harmful than the entire message being one
+    giant code block.
+
+    Single backtick: after balancing triple-backtick fences, strip all
+    complete `` ```…``` `` regions and count remaining standalone `` ` ``.
+    If odd, append a closing inline-code backtick.  Same trade-off: a
+    stray closing backtick may produce a brief empty inline-code span,
+    which is far less harmful than the rest of the message being rendered
+    as inline code.
+
+    Returns:
+        The input text with closing markers appended if needed, or the
+        input text unchanged.
+    """
+    if not isinstance(text, str) or not text:
+        return text
+
+    # Step 1: fix triple-backtick code-block fences (existing logic)
+    if text.count("```") % 2 == 1:
+        text = text.rstrip("\n") + "\n```"
+
+    # Step 2: fix single-backtick inline-code spans
+    # Remove complete ```…``` regions so their internal backticks don't
+    # pollute the standalone count.  Also remove any trailing unclosed
+    # ``` that leaks through (defence in depth).
+    import re
+    without_fences = re.sub(r"```.*?```", "", text, flags=re.DOTALL)
+    without_fences = re.sub(r"```[^`]*$", "", without_fences)
+
+    if without_fences.count("`") % 2 == 1:
+        text = text + "`"
+
+    return text
+
+
 @dataclass
 class StreamConsumerConfig:
     """Runtime config for a single stream consumer instance."""
@@ -744,39 +808,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 +898,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
@@ -1074,26 +1175,103 @@ def _continuation_text(self, final_text: str) -> str:
             return final_text[len(prefix):].lstrip()
         return final_text
 
+    @staticmethod
+    def _balance_fences_across_chunks(chunks: "list[str]") -> "list[str]":
+        """Close orphaned ``` fences at each chunk boundary and reopen on the next.
+
+        When a split lands inside a triple-backtick code block, the head chunk
+        would render everything after the orphaned fence as code, and the tail
+        chunk's content would lose its code formatting.  Mirror
+        ``BasePlatformAdapter.truncate_message``'s contract: close the fence at
+        the end of the chunk and reopen it (with the original language tag) at
+        the start of the next one, so EVERY delivered chunk is fence-balanced
+        on its own.
+        """
+        if len(chunks) <= 1:
+            return chunks
+        out: "list[str]" = []
+        carry_lang: "Optional[str]" = None
+        for chunk in chunks:
+            prefix = f"```{carry_lang}\n" if carry_lang is not None else ""
+            in_code = carry_lang is not None
+            lang = carry_lang or ""
+            for line in chunk.split("\n"):
+                stripped = line.strip()
+                if stripped.startswith("```"):
+                    if in_code:
+                        in_code = False
+                        lang = ""
+                    else:
+                        in_code = True
+                        tag = stripped[3:].strip()
+                        lang = tag.split()[0] if tag else ""
+            body = prefix + chunk
+            if in_code:
+                body += "\n```"
+                carry_lang = lang
+            else:
+                carry_lang = None
+            out.append(body)
+        return out
+
     @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."""
+        """Split text into reasonably sized chunks for fallback sends.
+
+        Chunks are fence-balanced: a split inside a ``` code block closes the
+        fence on the head chunk and reopens it on the tail, so no chunk leaves
+        the rest of a message rendering as one giant code block.
+        """
         if len_fn(text) <= limit:
             return [text]
+        # Reserve headroom for the close/reopen fence markers the balancing
+        # pass may add, so balanced chunks stay within the platform limit.
+        split_limit = limit
+        if "```" in text:
+            split_limit = max(limit - 16, limit // 2, 1)
         chunks: list[str] = []
         remaining = text
-        while len_fn(remaining) > limit:
-            _cp_budget = _custom_unit_to_cp(remaining, limit, len_fn)
+        while len_fn(remaining) > split_limit:
+            _cp_budget = _custom_unit_to_cp(remaining, split_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
+        return GatewayStreamConsumer._balance_fences_across_chunks(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.
@@ -1101,6 +1279,10 @@ async def _send_fallback_final(self, text: str) -> None:
         Retries each chunk once on flood-control failures with a short delay.
         """
         final_text = self._clean_for_display(text)
+        # Ensure balanced code fences before computing continuation,
+        # so the closing fence reaches the user even when the fallback
+        # only delivers the tail after mid-stream edits failed.
+        final_text = ensure_closed_code_fences(final_text)
         continuation = self._continuation_text(final_text)
         self._fallback_final_send = False
         if not continuation.strip():
@@ -1758,6 +1940,12 @@ async def _send_or_edit(
         # Media files are delivered as native attachments after the stream
         # finishes (via _deliver_media_from_response in gateway/run.py).
         text = self._clean_for_display(text)
+        # Ensure code fences are balanced before send/edit.  Model output
+        # truncated mid-code-block (e.g. finish_reason="length") leaves an
+        # orphaned ``` which, on Discord/Slack/Matrix, causes the entire
+        # remaining output to render as a single code block.  This covers
+        # the streaming edit path (G2) and first-send path alike.
+        text = ensure_closed_code_fences(text)
         # A bare streaming cursor is not meaningful user-visible content and
         # can render as a stray tofu/white-box message on some clients.
         visible_without_cursor = text
diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py
index 7a416449ef21..61bba6b14e4e 100644
--- a/plugins/platforms/slack/adapter.py
+++ b/plugins/platforms/slack/adapter.py
@@ -16,6 +16,7 @@
 import os
 import re
 import time
+import unicodedata
 from dataclasses import dataclass, field
 from typing import Callable, Dict, Optional, Any, Tuple, List
 
@@ -128,6 +129,119 @@ def _slack_file_marker(file_obj: Dict[str, Any]) -> str:
         return f"[audio: {name}]"
     return f"[file: {name} ({mimetype})]" if mimetype else f"[file: {name}]"
 
+
+# ── GFM markdown table preprocessing ──────────────────────────────────────
+# Slack mrkdwn does not render GFM-style pipe tables — they appear as literal
+# pipes. Wrapping in ``` fences makes them render as monospace preformatted
+# text, and padding cells to per-column max display width (with East-Asian
+# Wide / CJK awareness) keeps the columns aligned for the reader.
+
+_TABLE_SEPARATOR_RE = re.compile(
+    r"^\s*\|?\s*:?-+:?\s*(?:\|\s*:?-+:?\s*){1,}\|?\s*$"
+)
+
+
+def _is_table_row(line: str) -> bool:
+    """Return True if *line* could plausibly be a table data row."""
+    stripped = line.strip()
+    return bool(stripped) and "|" in stripped
+
+
+def _disp_width(s: str) -> int:
+    """Monospace display width: East-Asian Wide / Full-width chars count as 2."""
+    return sum(2 if unicodedata.east_asian_width(c) in "WF" else 1 for c in s)
+
+
+def _pad(cell: str, width: int) -> str:
+    """Right-pad *cell* with spaces until its display width equals *width*."""
+    delta = width - _disp_width(cell)
+    return cell + (" " * delta if delta > 0 else "")
+
+
+def _split_table_row(line: str) -> List[str]:
+    """Split a ``| a | b | c |`` row into trimmed cells (outer pipes optional)."""
+    s = line.strip()
+    if s.startswith("|"):
+        s = s[1:]
+    if s.endswith("|"):
+        s = s[:-1]
+    return [c.strip() for c in s.split("|")]
+
+
+def _align_table(rows: List[str]) -> List[str]:
+    """Re-emit a markdown table with cells padded to per-column max display width.
+
+    *rows[0]* is the header, *rows[1]* is the GFM separator (regenerated to
+    match new column widths), and *rows[2:]* are data rows. Cells are
+    normalized to a uniform column count (missing cells filled with empty
+    strings) before width calculation.
+    """
+    if len(rows) < 2:
+        return rows
+    parsed = [_split_table_row(r) for r in rows]
+    n_cols = max(len(r) for r in parsed)
+    for r in parsed:
+        while len(r) < n_cols:
+            r.append("")
+    sep_idx = 1
+    parsed[sep_idx] = ["---"] * n_cols  # placeholder; regenerated below
+    widths = [max(_disp_width(r[c]) for r in parsed) for c in range(n_cols)]
+    out: List[str] = []
+    for idx, row in enumerate(parsed):
+        if idx == sep_idx:
+            cells = ["-" * widths[c] for c in range(n_cols)]
+        else:
+            cells = [_pad(row[c], widths[c]) for c in range(n_cols)]
+        out.append("| " + " | ".join(cells) + " |")
+    return out
+
+
+def _wrap_markdown_tables(text: str) -> str:
+    """Wrap GFM pipe tables in ``` fences and align column widths.
+
+    Detected by a row containing ``|`` immediately followed by a delimiter row
+    matching :data:`_TABLE_SEPARATOR_RE`. Subsequent pipe-containing non-blank
+    lines are consumed as the table body. Tables already inside fenced code
+    blocks are left alone.
+    """
+    if not text or "|" not in text or "-" not in text:
+        return text
+
+    lines = text.split("\n")
+    out: List[str] = []
+    in_fence = False
+    i = 0
+    while i < len(lines):
+        line = lines[i]
+        stripped = line.lstrip()
+        if stripped.startswith("```"):
+            in_fence = not in_fence
+            out.append(line)
+            i += 1
+            continue
+        if in_fence:
+            out.append(line)
+            i += 1
+            continue
+        if (
+            "|" in line
+            and i + 1 < len(lines)
+            and _TABLE_SEPARATOR_RE.match(lines[i + 1])
+        ):
+            block = [line, lines[i + 1]]
+            j = i + 2
+            while j < len(lines) and _is_table_row(lines[j]):
+                block.append(lines[j])
+                j += 1
+            out.append("```")
+            out.extend(_align_table(block))
+            out.append("```")
+            i = j
+            continue
+        out.append(line)
+        i += 1
+    return "\n".join(out)
+
 # ContextVar carrying the user_id of the slash-command invoker.
 # Set in _handle_slash_command, read in send() to match the correct
 # stashed response_url when multiple users issue commands on the same
@@ -288,8 +402,8 @@ def _render_inline_elements(elements: list) -> str:
                 pieces.append(el.get("text", ""))
             elif el_type == "link":
                 url = el.get("url", "")
-                text = el.get("text", "") or url
-                pieces.append(f"{text} ({url})")
+                text = el.get("text", "")
+                pieces.append(f"{text} ({url})" if text and text != url else url)
             elif el_type == "channel":
                 pieces.append(f"<#{el.get('channel_id', '')}>")
             elif el_type == "user":
@@ -398,6 +512,28 @@ def _extract_text_from_slack_attachments(attachments: list) -> str:
     return "\n".join(line for line in lines if line).strip()
 
 
+_SLACK_MRKDWN_LINK_RE = re.compile(
+    r"<((?:https?|mailto):[^>|]+)(?:\|([^>]+))?>"
+)
+
+
+def _normalize_slack_text_for_dedupe(text: str) -> str:
+    """Canonicalize equivalent Slack plain-text and rich-block link forms.
+
+    Slack serializes the same authored link as ``<url|label>`` in the event's
+    plain ``text`` field and as a structured ``link`` element in ``blocks``.
+    Comparing those raw strings makes a normal rich-text message look like
+    additional quoted content and appends the whole message a second time.
+    """
+
+    def _link(match: re.Match) -> str:
+        url, label = match.group(1), match.group(2)
+        return f"{label} ({url})" if label and label != url else url
+
+    canonical = _SLACK_MRKDWN_LINK_RE.sub(_link, text or "")
+    return re.sub(r"\s+", " ", canonical).strip()
+
+
 def _serialize_slack_blocks_for_agent(blocks: list, max_chars: int = 6000) -> str:
     """Return a compact, redacted JSON view of the current message's Block Kit payload."""
     if not blocks:
@@ -3088,6 +3224,47 @@ def _rich_blocks_enabled(self) -> bool:
             return False
         return str(raw).strip().lower() in {"1", "true", "yes", "on"}
 
+    def _markdown_blocks_enabled(self) -> bool:
+        """Whether to render outbound messages via Slack's ``markdown`` block.
+
+        Opt-in via ``platforms.slack.extra.markdown_blocks`` (config.yaml).
+        Slack's Block Kit ``markdown`` block accepts *standard* markdown
+        (tables, headers, task lists, fenced code with syntax highlighting,
+        links) and lets Slack do the translation natively — eliminating the
+        lossy markdown→mrkdwn conversion for the rendered layout.  The
+        mrkdwn-converted ``text`` field is always kept as the
+        notification/search/accessibility fallback, and the block-rejection
+        retry path drops blocks and re-sends plain mrkdwn on surfaces or
+        workspaces where the block type is not accepted — so enabling this
+        can never lose a message.
+
+        Kept opt-in rather than default because Slack documents the block for
+        "apps that use platform AI features" and caps cumulative ``markdown``
+        block text at 12,000 characters per payload; availability on every
+        plan tier / app type is not guaranteed.
+        """
+        raw = self.config.extra.get("markdown_blocks")
+        if raw is None:
+            return False
+        return str(raw).strip().lower() in {"1", "true", "yes", "on"}
+
+    # Slack caps the cumulative text of all ``markdown`` blocks in a single
+    # payload at 12,000 characters.  Leave margin for the feedback block.
+    _MARKDOWN_BLOCK_MAX = 11_500
+
+    def _markdown_block_payload(self, content: str) -> Optional[list]:
+        """Return a ``markdown`` block payload for ``content``, or ``None``.
+
+        Declines (returns ``None``) for empty content and for content over
+        Slack's 12k cumulative markdown-block cap — the caller then falls
+        through to the rich_blocks renderer or the plain mrkdwn text path.
+        """
+        if not content or not content.strip():
+            return None
+        if len(content) > self._MARKDOWN_BLOCK_MAX:
+            return None
+        return [{"type": "markdown", "text": content}]
+
     def _feedback_buttons_enabled(self) -> bool:
         """Whether to include Slack AI feedback buttons on final responses."""
         raw = self.config.extra.get("feedback_buttons")
@@ -3130,13 +3307,28 @@ def _append_feedback_block(self, blocks: Optional[list]) -> Optional[list]:
         return [*blocks, self._feedback_block()]
 
     def _maybe_blocks(self, content: str) -> Optional[list]:
-        """Render ``content`` to Block Kit blocks when the feature is enabled.
-
-        Returns ``None`` when rich blocks are disabled, or when the renderer
-        declines (empty / too complex / unexpected shape) — the caller then
-        falls back to the plain ``text`` payload. A ``text`` fallback is ALWAYS
-        sent alongside blocks, so this can safely return ``None`` at any time.
+        """Render ``content`` to Block Kit blocks when a block mode is enabled.
+
+        Preference order:
+
+        1. ``markdown_blocks`` — Slack's native ``markdown`` block renders the
+           *raw* standard markdown (tables, headers, code fences with syntax
+           highlighting) with Slack doing the translation (#8552).
+        2. ``rich_blocks`` — the local Block Kit renderer (headers, dividers,
+           ``rich_text`` lists, native ``table`` blocks).
+
+        Returns ``None`` when both are disabled, or when the renderer
+        declines (empty / too long / too complex / unexpected shape) — the
+        caller then falls back to the plain ``text`` payload. A ``text``
+        fallback is ALWAYS sent alongside blocks, so this can safely return
+        ``None`` at any time, and the block-rejection retry path recovers
+        when Slack rejects the payload (e.g. a surface without ``markdown``
+        block support).
         """
+        if self._markdown_blocks_enabled():
+            md_blocks = self._markdown_block_payload(content)
+            if md_blocks:
+                return sanitize_blocks(self._append_feedback_block(md_blocks))
         if not self._rich_blocks_enabled():
             return None
         try:
@@ -3149,15 +3341,21 @@ def _maybe_blocks(self, content: str) -> Optional[list]:
     def format_message(self, content: str) -> str:
         """Convert standard markdown to Slack mrkdwn format.
 
-        Protected regions (code blocks, inline code) are extracted first so
-        their contents are never modified.  Standard markdown constructs
-        (headers, bold, italic, links) are translated to mrkdwn syntax.
+        GFM-style pipe tables are first wrapped in ``` fences and column-
+        aligned (with CJK display-width awareness) so they render as monospace
+        preformatted text instead of literal-pipe noise. Then protected
+        regions (code blocks — including the table fences just emitted —
+        and inline code) are extracted so their contents are never modified.
+        Standard markdown constructs (headers, bold, italic, links) are
+        translated to mrkdwn syntax.
         Broadcast mentions are escaped before entity protection so model output
         cannot trigger workspace- or channel-wide notifications by default.
         """
         if not content:
             return content
 
+        content = _wrap_markdown_tables(content)
+
         placeholders: dict = {}
         counter = [0]
 
@@ -3178,10 +3376,25 @@ def _ph(value: str) -> str:
             lambda m: m.group(0).replace("<", "&lt;", 1), text
         )
 
-        # 1) Protect fenced code blocks (``` ... ```)
+        # 1) Protect fenced code blocks (``` ... ```).  Slack's mrkdwn does not
+        # strip the optional language tag like GitHub-flavored markdown — it
+        # renders ```text\nfoo\n``` as a code block whose literal first line
+        # is "text".  Drop the tag from the opening fence before stashing.
+        # Stripping only fires for a genuine opening fence — a ``` at the
+        # start of a line, tagged with a single token (no spaces/backticks).
+        # The outer regex below deliberately matches loosely, so it can also
+        # group from a mid-line ``` (e.g. an inline ```span```); that first
+        # line is real content and must survive byte-for-byte.  This pass
+        # runs first, so match positions refer to the original message.
+        def _protect_fence(m):
+            block = m.group(0)
+            if m.start() == 0 or m.string[m.start() - 1] == "\n":
+                block = re.sub(r"\A```[^\s`]+[ \t]*(\r?\n)", r"```\1", block)
+            return _ph(block)
+
         text = re.sub(
             r"(```(?:[^\n]*\n)?[\s\S]*?```)",
-            lambda m: _ph(m.group(0)),
+            _protect_fence,
             text,
         )
 
@@ -3215,7 +3428,14 @@ def _convert_markdown_link(m):
 
         # 6) Escape Slack control characters in remaining plain text.
         # Unescape first so already-escaped input doesn't get double-escaped.
-        text = text.replace("&amp;", "&").replace("&lt;", "<").replace("&gt;", ">")
+        # Single pass: sequential str.replace would re-scan its own output, so
+        # the & from "&amp;" could pair with a following "lt;" and decode twice
+        # ("&amp;lt;" → "&lt;" → "<"), destroying literal entity text.
+        text = re.sub(
+            r"&(amp|lt|gt);",
+            lambda m: {"amp": "&", "lt": "<", "gt": ">"}[m.group(1)],
+            text,
+        )
         text = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
 
         # 7) Convert headers (## Title) → *Title* (bold)
@@ -3235,9 +3455,20 @@ def _convert_header(m):
         )
 
         # 9) Convert bold: **text** → *text* (Slack bold)
+        # Slack's mrkdwn parser fails to recognize the closing * when it is
+        # immediately preceded by non-word characters (e.g. ), ], }, ., :, —).
+        # This causes the parser to silently truncate the rest of the message.
+        # Insert a zero-width space (U+200B) between the last character and
+        # the closing * whenever the last character is not alphanumeric or _.
+        def _convert_bold(m):
+            inner = m.group(1)
+            if inner and not (inner[-1].isalnum() or inner[-1] == "_"):
+                return _ph(f"*{inner}\u200b*")
+            return _ph(f"*{inner}*")
+
         text = re.sub(
             r"\*\*(.+?)\*\*",
-            lambda m: _ph(f"*{m.group(1)}*"),
+            _convert_bold,
             text,
         )
 
@@ -4602,7 +4833,12 @@ async def _handle_slack_message(
                 # Only append if the blocks contain text not already present
                 # in the plain text field (avoids duplication).
                 stripped_blocks = blocks_text.strip()
-                if stripped_blocks and stripped_blocks not in text.strip():
+                block_text_is_duplicate = (
+                    stripped_blocks in text.strip()
+                    or _normalize_slack_text_for_dedupe(stripped_blocks)
+                    == _normalize_slack_text_for_dedupe(text)
+                )
+                if stripped_blocks and not block_text_is_duplicate:
                     logger.debug(
                         "Slack: extracted additional text from blocks "
                         "(likely quoted/forwarded content; chars=%d)",
diff --git a/plugins/platforms/slack/block_kit.py b/plugins/platforms/slack/block_kit.py
index 5f8dc4d0c1b1..c2767208ccc6 100644
--- a/plugins/platforms/slack/block_kit.py
+++ b/plugins/platforms/slack/block_kit.py
@@ -536,19 +536,40 @@ def flush_para() -> None:
 
 
 def _split_text(text: str, limit: int) -> List[str]:
-    """Split ``text`` into <= ``limit``-char chunks on line, then hard, boundaries."""
+    """Split ``text`` into <= ``limit``-char chunks on line, then hard, boundaries.
+
+    Chunks are fence-balanced: when a split lands inside a ``` code span that
+    survived into section text (the renderer normally routes fenced blocks to
+    ``rich_text_preformatted``, but mrkdwn text can still carry fences), the
+    fence is closed at the end of the chunk and reopened on the next so each
+    section renders correctly on its own.
+    """
     if len(text) <= limit:
         return [text]
+    # Reserve headroom for the close/reopen markers the balancing pass adds.
+    split_limit = max(limit - 8, limit // 2, 1) if "```" in text else limit
     out: List[str] = []
     remaining = text
-    while len(remaining) > limit:
-        cut = remaining.rfind("\n", 0, limit)
+    while len(remaining) > split_limit:
+        cut = remaining.rfind("\n", 0, split_limit)
         if cut <= 0:
-            cut = limit
+            cut = split_limit
         out.append(remaining[:cut])
         remaining = remaining[cut:].lstrip("\n")
     if remaining:
         out.append(remaining)
+    if len(out) > 1 and "```" in text:
+        balanced: List[str] = []
+        reopen = False
+        for chunk in out:
+            if reopen:
+                chunk = "```\n" + chunk
+            odd = chunk.count("```") % 2 == 1
+            if odd:
+                chunk += "\n```"
+            reopen = odd
+            balanced.append(chunk)
+        out = balanced
     return out
 
 
diff --git a/tests/gateway/test_code_fence_tracking.py b/tests/gateway/test_code_fence_tracking.py
new file mode 100644
index 000000000000..b43224d49838
--- /dev/null
+++ b/tests/gateway/test_code_fence_tracking.py
@@ -0,0 +1,659 @@
+"""
+Tests for code fence tracking across message split / truncation / streaming paths.
+
+The central problem: when a message contains triple-backtick code blocks (```)
+and gets split (1/2)(2/2) or truncated mid-stream, Discord renders the entire
+remaining output as a single code block unless the fences are properly closed
+and reopened.
+
+Three code paths matter:
+  1. BasePlatformAdapter.truncate_message()    — non-streaming split (HAS fence tracking)
+  2. GatewayStreamConsumer._send_or_edit()     — streaming send (NO fence tracking)
+  3. GatewayStreamConsumer._split_text_chunks()— fallback final send (NO fence tracking)
+
+Known gap: truncate_message closes orphaned fences on INTERMEDIATE chunks but
+NOT on the FINAL chunk (line 4853-4854: ``if _len(prefix) + _len(remaining)
+<= max_length - INDICATOR_RESERVE: chunks.append(prefix + remaining); break``
+skips the fence-closing check that intermediate chunks get at line 4904-4922).
+
+Test categories:
+  A. truncate_message — reasoning-fence format (basic)
+  B. truncate_message — unclosed fence (content ≤ max_length → passes through)
+  C. truncate_message — multiple alternating ``` blocks
+  D. truncate_message — last chunk gap (intermediate closes, final may not)
+  E. _filter_and_accumulate — preserves ``` outside think blocks
+  F. _split_text_chunks — NO fence tracking (GAP)
+  G. Reasoning truncation — short content (passes through unfixed)
+  H. Reasoning truncation — long content (intermediate closed, last may not)
+  I. Integration: what a fix would look like
+"""
+
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch, ANY
+
+from gateway.platforms.base import BasePlatformAdapter
+from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig, ensure_closed_code_fences
+
+
+# ── helpers ───────────────────────────────────────────────────────────────
+
+def _len_with_indicator(text: str) -> int:
+    """Simulate the length after INDICATOR_RESERVE (10) is subtracted."""
+    return len(text)
+
+
+def _count_fences(text: str) -> int:
+    """Count triple-backtick code fence markers in text."""
+    return text.count("```")
+
+
+def _odd_fences(text: str) -> bool:
+    """Return True if text has an odd number of ``` markers."""
+    return _count_fences(text) % 2 == 1
+
+
+def _assert_balanced(chunks, label="chunk"):
+    """Assert every chunk in a list has an even number of ``` markers."""
+    for i, chunk in enumerate(chunks):
+        assert not _odd_fences(chunk), (
+            f"{label} {i+1}/{len(chunks)} has odd ``` count "
+            f"(unbalanced fence)\n  preview: {chunk[:120]}..."
+        )
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  A. truncate_message — reasoning-fence format (short content)
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestTruncateMessageShort:
+    """Content that fits in one message (≤ max_length)."""
+
+    def test_short_no_split(self):
+        """Content under max_length passes through unchanged."""
+        content = "💭 **Reasoning:**\n```\nthinking\n```\nHere is the answer."
+        assert BasePlatformAdapter.truncate_message(content, 500) == [content]
+
+    def test_short_unclosed_fence_passes_through(self):
+        """Short content with unclosed ``` is returned as-is (no fix)."""
+        content = "💭 **Reasoning:**\n```\ncut off"
+        result = BasePlatformAdapter.truncate_message(content, 500)
+        assert result == [content]
+        assert _odd_fences(result[0]), "Short unclosed content stays unclosed"
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  B. truncate_message — split forces fence close on INTERMEDIATE chunks
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestTruncateMessageIntermediateCloses:
+    """When splitting, intermediate chunks that end inside a code block get
+    an auto-closing fence appended."""
+
+    def test_split_inside_fence_closes_first_chunk(self):
+        """First split lands inside ``` → first chunk gets closing fence."""
+        body = "\n".join(f"line{i}" for i in range(50))
+        content = f"💭 **Reasoning:**\n```\n{body}\n```\nDone."
+        max_len = 150
+        chunks = BasePlatformAdapter.truncate_message(content, max_len)
+        assert len(chunks) >= 2
+
+        # Intermediate chunks (all except possibly the last) should be
+        # balanced.  The last chunk may or may not be balanced depending
+        # on whether its content includes the closing ```.
+        for i, chunk in enumerate(chunks[:-1]):
+            assert not _odd_fences(chunk), (
+                f"Intermediate chunk {i+1}/{len(chunks)} has odd ```"
+            )
+
+    def test_multiple_fences_across_chunks(self):
+        """Reasoning block + code block across multiple chunks — each
+        intermediate chunk closes orphaned fences."""
+        content = (
+            "💭 **Reasoning:**\n```\n" + "x" * 50 + "\n```\n"
+            "Main answer:\n```python\n"
+            + "\n".join(f"line{i}" for i in range(30))
+            + "\n```\nend"
+        )
+        chunks = BasePlatformAdapter.truncate_message(content, 150)
+        assert len(chunks) >= 2
+        for i, chunk in enumerate(chunks[:-1]):
+            assert not _odd_fences(chunk), (
+                f"Intermediate chunk {i+1} has odd ```"
+            )
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  C. truncate_message — carry_lang reopens on next chunk
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestTruncateMessageCarryLang:
+    """When a chunk ends mid-code-block, the language tag is carried to
+    the next chunk for reopening."""
+
+    def test_carry_lang_reopens_with_tag(self):
+        """Second chunk reopens with same language tag as first."""
+        body = "\n".join(f"// line{i}" for i in range(50))
+        content = f"```python\n{body}\n```\nend"
+        chunks = BasePlatformAdapter.truncate_message(content, 120)
+        assert len(chunks) >= 2
+
+        first = chunks[0]
+        # First chunk: content ends in code block → gets closing fence
+        # Strip the (1/N) indicator before checking
+        first_clean = first.rsplit(" (", 1)[0]
+        assert first_clean.endswith("```"), f"First chunk end: {first_clean[-30:]}"
+
+        second = chunks[1]
+        # Second chunk reopens with ```python (from carry_lang)
+        # The prefix is "```python\n" prepended by truncate_message
+        second_stripped = second.lstrip()
+        assert second_stripped.startswith("```python"), (
+            f"Second chunk should reopen with ```python, "
+            f"got start: {second[:60]}..."
+        )
+
+    def test_carry_lang_empty_tag(self):
+        """``` without language tag reopens as bare ```."""
+        body = "\n".join(f"x{i}" for i in range(50))
+        content = f"```\n{body}\n```"
+        chunks = BasePlatformAdapter.truncate_message(content, 100)
+        assert len(chunks) >= 2
+        second = chunks[1]
+        second_stripped = second.lstrip()
+        assert second_stripped.startswith("```"), (
+            f"Should reopen with bare ```"
+        )
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  D. truncate_message — THE GAP: last chunk does not auto-close
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestTruncateMessageLastChunkGap:
+    """The final chunk (when ``remaining`` fits) is appended via the
+    early-break path at line 4853-4854 of base.py, which does NOT run
+    the fence-balance check.  If the remaining content has an odd count
+    of ```, so does the final chunk."""
+
+    def test_last_chunk_can_have_odd_fence_when_content_unclosed(self):
+        """Content with unclosed ``` where the last chunk fits → no fix."""
+        long_body = "\n".join(f"line{i}" for i in range(100))
+        content = f"```\n{long_body}"
+        # The first split happens at ~186 chars, last chunk is small
+        chunks = BasePlatformAdapter.truncate_message(content, 150)
+        assert len(chunks) >= 2
+        # The last chunk may have odd ``` because the remaining content
+        # (after carry_lang prefix) doesn't contain a closing ```
+        last = chunks[-1]
+        # Strip the (N/N) indicator
+        last_clean = last.rsplit(" (", 1)[0]
+        if _odd_fences(last_clean):
+            # This demonstrates the GAP — last chunk has unbalanced fence
+            pass  # Not asserting — the gap is real
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  E. _filter_and_accumulate — think-tag state machine: fence impact
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestFilterAndAccumulate:
+    """GatewayStreamConsumer._filter_and_accumulate strips <think> tags
+    but must not corrupt ``` outside them."""
+
+    @staticmethod
+    def _consumer():
+        cfg = StreamConsumerConfig(buffer_only=True)
+        return GatewayStreamConsumer(
+            adapter=MagicMock(), chat_id="12345", config=cfg,
+        )
+
+    def test_plain_text_preserved(self):
+        c = self._consumer()
+        c._filter_and_accumulate("Hello world")
+        assert c._accumulated == "Hello world"
+
+    def test_fence_outside_think_preserved(self):
+        c = self._consumer()
+        c._filter_and_accumulate("```\ncode\n```\nmain")
+        assert _count_fences(c._accumulated) == 2
+        assert "main" in c._accumulated
+
+    def test_fence_inside_think_is_stripped(self):
+        c = self._consumer()
+        c._filter_and_accumulate(
+            "before\n<think>\n```python\nx = 1\n```\n</think>\nafter"
+        )
+        assert "```" not in c._accumulated
+        assert "before" in c._accumulated
+        assert "after" in c._accumulated
+
+    def test_truncated_think_discards_content(self):
+        """<think> without closing tag discards everything after."""
+        c = self._consumer()
+        c._filter_and_accumulate("before\n<think>\n```\ncode")
+        assert "```" not in c._accumulated
+        assert "before" in c._accumulated
+        assert c._in_think_block
+
+    def test_consecutive_think_blocks(self):
+        c = self._consumer()
+        c._filter_and_accumulate(
+            "<think>\n```\nfirst\n```\n</think>"
+        )
+        c._filter_and_accumulate(
+            "<think>\n```\nsecond\n```\n</think>"
+        )
+        assert "```" not in c._accumulated
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  F. _split_text_chunks — NO fence tracking (fallback final path)
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestSplitTextChunks:
+    """GatewayStreamConsumer._split_text_chunks is a simple text splitter
+    with NO code-fence awareness.  Used by _send_fallback_final."""
+
+    def test_split_inside_fence_does_not_close(self):
+        """Split lands inside ``` → no auto-close."""
+        long = "\n".join(f"code{i}" for i in range(30))
+        text = f"```python\n{long}\n```"
+        chunks = GatewayStreamConsumer._split_text_chunks(text, 60)
+        assert len(chunks) >= 2
+        # First chunk may have odd ``` — no fence tracking
+        # Just verify chunks are of type str and non-empty
+        assert all(isinstance(c, str) and c for c in chunks)
+
+    def test_no_metadata(self):
+        """Chunks are plain strings — no carry_lang."""
+        long = "\n".join(f"line{i}" for i in range(30))
+        chunks = GatewayStreamConsumer._split_text_chunks(long, 60)
+        assert len(chunks) >= 2
+        assert all(isinstance(c, str) for c in chunks)
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  G. Reasoning truncation — model cut off mid-reasoning-block
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestReasoningTruncation:
+    """When the model runs out of tokens mid-reasoning-block."""
+
+    DISCORD_LIMIT = 2000
+
+    def test_short_truncation_unfixed(self):
+        """Fits in one message → truncate_message passes through unfixed."""
+        truncated = "💭 **Reasoning:**\n```\nI was thinking about"
+        chunks = BasePlatformAdapter.truncate_message(truncated, self.DISCORD_LIMIT)
+        assert chunks == [truncated]
+        assert _odd_fences(chunks[0])
+
+    def test_long_truncation_last_chunk_gap(self):
+        """Spans multiple chunks → intermediate chunks close, last may not."""
+        long_body = "\n".join(f"line{i}" for i in range(100))
+        truncated = f"💭 **Reasoning:**\n```\n{long_body}"
+        chunks = BasePlatformAdapter.truncate_message(truncated, 150)
+
+        assert len(chunks) >= 2
+        # All intermediate chunks must be balanced
+        for i, chunk in enumerate(chunks[:-1]):
+            assert not _odd_fences(chunk), (
+                f"Intermediate chunk {i+1}/{len(chunks)} should be balanced"
+            )
+
+        # The LAST chunk may or may not be balanced — this is a KNOWN GAP.
+        # When the last chunk fits via the early-break path (line 4853-4854),
+        # the carry_lang prefix is prepended but no closing fence is added.
+        last = chunks[-1]
+        last_clean = last.rsplit(" (", 1)[0]
+        count = _count_fences(last_clean)
+        assert count % 2 == 0 or count % 2 == 1, "Real gap — either outcome possible"
+        if _odd_fences(last_clean):
+            # This IS the gap: last chunk has ``` prefix but no closing ```
+            pass
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  H. Stream consumer — unclosed fence in final send (GAP)
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestStreamConsumerFinalSendGap:
+    """The stream consumer's normal final-send path (_send_or_edit via
+    run()) does not check for or fix unclosed ```.  The _accumulated
+    text goes to the adapter verbatim.
+
+    Note: This class uses synchronous tests because pytest-asyncio is not
+    installed in this project (existing stream consumer tests use it but
+    the conftest may register the marker differently).  We test the
+    accumulator behaviour directly.
+    """
+
+    def test_accumulator_has_no_fence_closing(self):
+        """Unit-level: _filter_and_accumulate does not track fence state."""
+        cfg = StreamConsumerConfig(buffer_only=True)
+        c = GatewayStreamConsumer(
+            adapter=MagicMock(), chat_id="12345", config=cfg,
+        )
+        c._filter_and_accumulate("A\n```\nunclosed")
+        assert "```" in c._accumulated
+        assert _odd_fences(c._accumulated), "GAP: no fence closing"
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  I. ensure_closed_code_fences — triple-backtick fence balancing
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestEnsureClosedCodeFences:
+    """Unit tests for the standalone ensure_closed_code_fences helper."""
+
+    # ── triple backtick ──────────────────────────────────────────────
+
+    def test_closes_unclosed_triple(self):
+        """Unclosed ``` gets a closing fence appended."""
+        assert not _odd_fences(ensure_closed_code_fences(
+            "💭 **Reasoning:**\n```\ncut off"
+        ))
+
+    def test_noop_balanced_triple(self):
+        """Already-balanced ``` blocks are unchanged."""
+        t = "```\nblock\n```\ncontent"
+        assert ensure_closed_code_fences(t) == t
+
+    def test_noop_no_fence(self):
+        """Plain text without fences passes through."""
+        t = "plain text"
+        assert ensure_closed_code_fences(t) == t
+
+    def test_noop_already_ends_with_close(self):
+        """Text ending with a balanced ``` is unchanged."""
+        t = "```\nblock\n```"
+        assert ensure_closed_code_fences(t) == t
+
+    def test_closes_unclosed_mid_message(self):
+        """``` in the middle (not at end) still gets closed when odd."""
+        t = "before\n```\nunclosed block\nmore text here"
+        result = ensure_closed_code_fences(t)
+        assert not _odd_fences(result)
+        assert result.endswith("\n```")
+
+    # ── single backtick ──────────────────────────────────────────────
+
+    def test_closes_unclosed_single(self):
+        """Orphaned single backtick gets a closing backtick appended."""
+        result = ensure_closed_code_fences("Here is `inline code")
+        assert result == "Here is `inline code`"
+
+    def test_noop_balanced_single(self):
+        """Paired single backticks are unchanged."""
+        t = "Here is `inline code` and more text."
+        assert ensure_closed_code_fences(t) == t
+
+    def test_single_inside_triple_ignored(self):
+        """Backticks inside ``` regions are NOT counted for single-bt parity."""
+        t = "```\n`code` inside\n```\noutside `text`"
+        # outside has paired `text` → balanced, triple-blocks are balanced → no change
+        assert ensure_closed_code_fences(t) == t
+
+    def test_single_outside_unclosed_after_triple(self):
+        """Unclosed single backtick outside ``` blocks gets fixed."""
+        t = "```\nblock\n```\noutside `text"
+        result = ensure_closed_code_fences(t)
+        assert result == "outside `text`" or result.endswith("`")
+
+    def test_both_triple_and_single_unclosed(self):
+        """Both ``` and ` unclosed → both get closed."""
+        result = ensure_closed_code_fences("```\ncode\nstill open `inline")
+        assert result.endswith("`")
+        assert "```\ncode\nstill open `inline`\n```" in result or result.count("```") % 2 == 0
+
+    def test_noop_empty_or_none(self):
+        """Empty/None returns unchanged."""
+        assert ensure_closed_code_fences("") == ""
+        assert ensure_closed_code_fences(None) is None
+
+    def test_single_inline_code_in_prose(self):
+        """Realistic prose with `handle: \"...\"` inline code."""
+        # `handle: "abc"` is open – unbalanced single backtick
+        t = (
+            'LLM 看到 `_headroom.retrieval.handle` 的值，就是它要傳給'
+            ' `headroom_retrieve(hash="125f4ae286e24ad8c0816907"` 的那個字串。'
+        )
+        result = ensure_closed_code_fences(t)
+        # After fix: the last unclosed ` gets closed at the end
+        assert result.count("`") % 2 == 0
+
+    def test_multiple_single_backtick_pairs(self):
+        """Multiple correctly-paired single backtick spans are unchanged."""
+        t = "Use `cmd1` for X and `cmd2` for Y."
+        assert ensure_closed_code_fences(t) == t
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  J. Missing: edit path bypasses truncate_message (GAP G2/G3)
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestEditPathBypass:
+    """When _send_or_edit has an existing _message_id, it calls
+    _edit_message() directly — bypassing truncate_message entirely.
+    This means code-fence tracking is NEVER applied to streaming edits."""
+
+    def test_edit_path_does_not_call_truncate_message(self):
+        """With _message_id set, _send_or_edit calls _edit_message
+        without passing through truncate_message."""
+        adapter = MagicMock()
+        adapter.edit_message = AsyncMock(return_value=MagicMock(
+            success=True, message_id="msg_1",
+        ))
+        adapter.MAX_MESSAGE_LENGTH = 2000
+        adapter.message_len_fn = len
+
+        config = StreamConsumerConfig(
+            buffer_only=False, transport="edit",
+            edit_interval=9999, buffer_threshold=9999,
+        )
+        consumer = GatewayStreamConsumer(
+            adapter=adapter, chat_id="12345", config=config,
+        )
+
+        # Simulate: already have a message to edit
+        consumer._message_id = "msg_1"
+        consumer._already_sent = True
+
+        # Spy on truncate_message
+        original = BasePlatformAdapter.truncate_message
+        called = []
+
+        def _spy(content, max_len, len_fn=None, **kw):
+            called.append(True)
+            return original(content, max_len, len_fn=len_fn, **kw)
+
+        with patch.object(BasePlatformAdapter, 'truncate_message', _spy):
+            import asyncio
+            result = asyncio.run(
+                consumer._send_or_edit("Hello world\n```\nunclosed",
+                                       finalize=True)
+            )
+
+        # truncate_message should NOT have been called — edit path
+        assert len(called) == 0, (
+            f"Edit path should NOT call truncate_message, called {len(called)} times"
+        )
+        # edit_message should have been called instead
+        adapter.edit_message.assert_called_once()
+
+    def test_first_send_path_calls_adapter_send(self):
+        """Without _message_id, _send_or_edit calls adapter.send
+        (not edit_message)."""
+        adapter = MagicMock()
+        adapter.send = AsyncMock(return_value=MagicMock(
+            success=True, message_id="msg_new",
+        ))
+        adapter.MAX_MESSAGE_LENGTH = 2000
+        adapter.message_len_fn = len
+
+        config = StreamConsumerConfig(
+            buffer_only=False, transport="edit",
+            edit_interval=9999, buffer_threshold=9999,
+        )
+        consumer = GatewayStreamConsumer(
+            adapter=adapter, chat_id="12345", config=config,
+        )
+        import asyncio
+        asyncio.run(
+            consumer._send_or_edit("Hello world\n```\nunclosed",
+                                   finalize=True)
+        )
+        # First-send path calls adapter.send, not edit_message
+        adapter.send.assert_called_once()
+        adapter.edit_message.assert_not_called()
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  K. Missing: overflow split first chunk (GAP G3)
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestOverflowSplitFenceGap:
+    """run() overflow split loop (lines 567-601) splits at newlines
+    without fence awareness.  The first chunk goes through edit path
+    (_send_or_edit with _message_id set) which has NO fence tracking."""
+
+    def test_overflow_split_first_chunk_no_fence_tracking(self):
+        """Simulate the overflow split loop's behaviour:
+        first chunk split inside ```, sent through edit path — no close."""
+        adapter = MagicMock()
+        adapter.edit_message = AsyncMock(return_value=MagicMock(
+            success=True, message_id="msg_1",
+        ))
+        adapter.MAX_MESSAGE_LENGTH = 2000
+        adapter.message_len_fn = len
+
+        config = StreamConsumerConfig(
+            buffer_only=False, transport="edit",
+            edit_interval=9999, buffer_threshold=9999,
+        )
+        consumer = GatewayStreamConsumer(
+            adapter=adapter, chat_id="12345", config=config,
+        )
+        consumer._message_id = "msg_1"
+        consumer._already_sent = True
+        consumer._edit_supported = True
+
+        # accumulated starts with ``` that gets split
+        consumer._accumulated = "```\n" + "\n".join(f"line{i}" for i in range(30)) + "\n```end"
+
+        # Safe limit small enough to force overflow
+        _safe_limit = 60
+        _raw_limit = 2000
+        _len_fn = len
+        _cp_budget = _len_fn(consumer._accumulated[:60])  # simulate
+
+        split_at = consumer._accumulated.rfind("\n", 0, _cp_budget)
+        chunk = consumer._accumulated[:split_at]
+        remaining = consumer._accumulated[split_at:].lstrip("\n")
+
+        # First chunk should have odd ``` (no close from edit path)
+        first_odd = _odd_fences(chunk)
+        # Second part (remaining) may or may not — depends on split point
+        # Just document the behaviour
+        if first_odd:
+            pass  # This demonstrates the gap: edit path doesn't close fence
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  L. Missing: fallback final with unclosed fence (GAP G4)
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestFallbackFinalFenceGap:
+    """_send_fallback_final uses _split_text_chunks (no fence tracking)
+    then each chunk goes through adapter.send() → truncate_message().
+    But since _split_text_chunks already split to ≤ limit, truncate_message
+    returns the chunk verbatim — unclosed fence passes through."""
+
+    def test_split_text_chunks_preserves_unclosed_fence(self):
+        """_split_text_chunks split inside ``` — chunks still have odd ```"""
+        long = "\n".join(f"code{i}" for i in range(30))
+        text = f"```\n{long}\n```"
+        chunks = GatewayStreamConsumer._split_text_chunks(text, 80)
+        assert len(chunks) >= 2
+        # At least some chunks may have odd ``` (no fence tracking)
+        odd_ones = [c for c in chunks if _odd_fences(c)]
+        # Just document: _split_text_chunks doesn't guarantee balanced fences
+
+    def test_fallback_final_truncate_message_noop(self):
+        """When fallback chunks are ≤ limit, truncate_message returns
+        them verbatim — no fence fixing."""
+        chunk = "```python\ndef foo():\n    pass\n"
+        # This chunk is under 2000 chars → truncate_message returns [chunk]
+        result = BasePlatformAdapter.truncate_message(chunk, 2000)
+        assert result == [chunk], (
+            "truncate_message no-op when content ≤ max_length"
+        )
+        assert _odd_fences(result[0]), "Unclosed fence passes through"
+
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+#  M. Widened: every chunk boundary is fence-balanced (C11 salvage)
+# ═══════════════════════════════════════════════════════════════════════════
+
+class TestSplitTextChunksFenceBalanced:
+    """_split_text_chunks now closes orphaned fences at each boundary and
+    reopens them on the next chunk (mirrors truncate_message's contract),
+    so the fallback-final path can never leave a chunk rendering the rest
+    of the message as one giant code block."""
+
+    def test_every_chunk_balanced_bare_fence(self):
+        long = "\n".join(f"code{i}" for i in range(30))
+        text = f"```\n{long}\n```"
+        chunks = GatewayStreamConsumer._split_text_chunks(text, 80)
+        assert len(chunks) >= 2
+        _assert_balanced(chunks, "fallback chunk")
+
+    def test_every_chunk_balanced_lang_fence_reopens_with_tag(self):
+        long = "\n".join(f"print({i})" for i in range(40))
+        text = f"```python\n{long}\n```"
+        chunks = GatewayStreamConsumer._split_text_chunks(text, 90)
+        assert len(chunks) >= 2
+        _assert_balanced(chunks, "fallback chunk")
+        # Continuation chunks reopen with the original language tag
+        for chunk in chunks[1:]:
+            assert chunk.startswith("```python"), (
+                f"continuation should reopen with ```python: {chunk[:40]!r}"
+            )
+
+    def test_prose_only_split_unchanged(self):
+        """No fences → behaviour identical to the plain splitter."""
+        text = "\n".join(f"line {i}" for i in range(50))
+        chunks = GatewayStreamConsumer._split_text_chunks(text, 60)
+        assert len(chunks) >= 2
+        assert "```" not in "".join(chunks)
+        # Round-trips the content (modulo the newline trimming at cuts)
+        assert "".join(c.replace("\n", "") for c in chunks) == text.replace("\n", "")
+
+    def test_unclosed_input_final_chunk_closed(self):
+        """Input truncated mid-block (finish_reason=length) → last chunk
+        still balanced."""
+        long = "\n".join(f"row{i}" for i in range(40))
+        text = f"```\n{long}"  # never closed
+        chunks = GatewayStreamConsumer._split_text_chunks(text, 80)
+        assert len(chunks) >= 2
+        _assert_balanced(chunks, "fallback chunk")
+
+    def test_balanced_chunks_respect_limit(self):
+        long = "\n".join(f"code{i}" for i in range(30))
+        text = f"```\n{long}\n```"
+        limit = 80
+        chunks = GatewayStreamConsumer._split_text_chunks(text, limit)
+        for chunk in chunks:
+            assert len(chunk) <= limit, (
+                f"balanced chunk exceeds limit: {len(chunk)} > {limit}"
+            )
+
+    def test_multiple_blocks_alternating(self):
+        text = (
+            "intro\n```\n" + "\n".join("a" * 10 for _ in range(10)) + "\n```\n"
+            "middle prose\n```js\n" + "\n".join("b" * 10 for _ in range(10)) + "\n```\nend"
+        )
+        chunks = GatewayStreamConsumer._split_text_chunks(text, 70)
+        assert len(chunks) >= 2
+        _assert_balanced(chunks, "fallback chunk")
diff --git a/tests/gateway/test_escape_reasoning_fences.py b/tests/gateway/test_escape_reasoning_fences.py
new file mode 100644
index 000000000000..3a54ea09dcc3
--- /dev/null
+++ b/tests/gateway/test_escape_reasoning_fences.py
@@ -0,0 +1,45 @@
+"""
+Tests for escape_code_fences_for_display.
+
+B1: Escape triple-backtick markers inside reasoning text before wrapping
+    in an outer ``` fence, so inner ``` doesn't break the outer block.
+"""
+
+import pytest
+from gateway.stream_consumer import escape_code_fences_for_display
+
+
+class TestEscapeCodeFencesForDisplay:
+    """escape_code_fences_for_display prevents inner ``` from breaking
+    the outer code block used to render reasoning."""
+
+    def test_no_fence_passthrough(self):
+        text = "plain reasoning text"
+        assert escape_code_fences_for_display(text) == text
+
+    def test_single_fence_escaped(self):
+        text = "model used ```python\nx = 1\n``` in its thinking"
+        result = escape_code_fences_for_display(text)
+        assert "```" not in result
+        assert "\\`\\`\\`" in result
+
+    def test_multiple_fences_all_escaped(self):
+        text = "```\nblock1\n``` and ```python\nblock2\n```"
+        result = escape_code_fences_for_display(text)
+        assert result.count("```") == 0
+        assert result.count("\\`\\`\\`") == 4
+
+    def test_empty_string(self):
+        assert escape_code_fences_for_display("") == ""
+
+    def test_none_returns_none(self):
+        assert escape_code_fences_for_display(None) is None
+
+    def test_integration_with_outer_fence(self):
+        """Simulates the gateway's reasoning wrapping logic."""
+        raw = "thinking about:\n```python\nprint('hi')\n```\nok"
+        escaped = escape_code_fences_for_display(raw)
+        wrapped = f"💭 **Reasoning:**\n```\n{escaped}\n```\n\nHere's the answer."
+        # The outer ``` should not be broken by inner ```
+        assert wrapped.count("```") == 2  # only outer open + close
+        assert "\\`\\`\\`" in wrapped
diff --git a/tests/gateway/test_slack.py b/tests/gateway/test_slack.py
index 80c28b53376f..be8896f9d88d 100644
--- a/tests/gateway/test_slack.py
+++ b/tests/gateway/test_slack.py
@@ -2681,6 +2681,49 @@ async def test_rich_text_blocks_do_not_duplicate_plain_text(self, adapter):
         msg_event = adapter.handle_message.call_args[0][0]
         assert msg_event.text == "hello world"
 
+    @pytest.mark.asyncio
+    async def test_rich_text_blocks_do_not_duplicate_semantically_equal_slack_links(
+        self, adapter
+    ):
+        """Slack's plain ``text`` uses mrkdwn links while rich_text blocks use
+        structured links. They are the same authored message and must not be
+        appended as a second copy merely because their serializations differ."""
+        event = self._make_event(
+            text=(
+                "Review <https://github.com/acme/design/pull/7|PR #7> and "
+                "<http://preview.example.com>."
+            ),
+            blocks=[
+                {
+                    "type": "rich_text",
+                    "elements": [
+                        {
+                            "type": "rich_text_section",
+                            "elements": [
+                                {"type": "text", "text": "Review "},
+                                {
+                                    "type": "link",
+                                    "url": "https://github.com/acme/design/pull/7",
+                                    "text": "PR #7",
+                                },
+                                {"type": "text", "text": " and "},
+                                {
+                                    "type": "link",
+                                    "url": "http://preview.example.com",
+                                },
+                                {"type": "text", "text": "."},
+                            ],
+                        }
+                    ],
+                }
+            ],
+        )
+
+        await adapter._handle_slack_message(event)
+
+        msg_event = adapter.handle_message.call_args[0][0]
+        assert msg_event.text == event["text"]
+
     @pytest.mark.asyncio
     async def test_rich_text_quotes_and_lists_are_extracted(self, adapter):
         """Nested quote and list content should be surfaced from rich_text blocks."""
@@ -3915,7 +3958,60 @@ def test_strikethrough(self, adapter):
         assert adapter.format_message("~~deleted~~") == "~deleted~"
 
     def test_code_block_preserved(self, adapter):
+        # Slack mrkdwn doesn't recognize language tags — it would render the
+        # tag as a literal first line of the code block — so the converter
+        # strips it.  Body content is still passed through verbatim.
         code = "```python\nx = **not bold**\n```"
+        assert adapter.format_message(code) == "```\nx = **not bold**\n```"
+
+    def test_code_block_strips_language_tag(self, adapter):
+        # Regression: Slack rendered a literal "text" line at the top of code
+        # blocks containing raw command output because the LLM emitted
+        # ```text fences and the converter passed them through unchanged.
+        code = "```text\nhello world\nline 2\n```"
+        assert adapter.format_message(code) == "```\nhello world\nline 2\n```"
+
+    def test_code_block_no_language_tag_unchanged(self, adapter):
+        code = "```\nplain output\n```"
+        assert adapter.format_message(code) == code
+
+    def test_inline_triple_backtick_unchanged(self, adapter):
+        # Single-line ```hello``` has no newline after the opening fence, so
+        # nothing should be stripped.
+        code = "```hello```"
+        assert adapter.format_message(code) == code
+
+    def test_mid_line_triple_backticks_content_preserved(self, adapter):
+        # The fence-protection regex matches loosely, so the inline
+        # ```pip install foo``` span is grouped as an "opening fence" whose
+        # first line is real content.  Stripping only fires for a ``` at the
+        # start of a line, so the span survives byte-for-byte.
+        text = "Use ```pip install foo``` then:\n```bash\ncode\n```"
+        assert adapter.format_message(text) == text
+
+    def test_mid_line_single_token_span_preserved(self, adapter):
+        # A single-token inline span that wraps across a newline looks
+        # exactly like a language tag — the line-start guard is what keeps
+        # the word "quotes" from being stripped as one.
+        text = "Wrap it in ```quotes\nlike this\n```"
+        assert adapter.format_message(text) == text
+
+    def test_back_to_back_fences_second_token_preserved(self, adapter):
+        # The second ``` group starts mid-line (right after the previous
+        # closing fence), so its first token is content, not a tag.
+        text = "```\nx\n``````b\ny\n```"
+        assert adapter.format_message(text) == text
+
+    def test_code_block_lang_tag_trailing_spaces_stripped(self, adapter):
+        code = "```python  \nx = 1\n```"
+        assert adapter.format_message(code) == "```\nx = 1\n```"
+
+    def test_code_block_crlf_lang_tag_stripped_preserves_crlf(self, adapter):
+        code = "```python\r\nx = 1\r\n```"
+        assert adapter.format_message(code) == "```\r\nx = 1\r\n```"
+
+    def test_code_block_crlf_no_tag_unchanged(self, adapter):
+        code = "```\r\nplain output\r\n```"
         assert adapter.format_message(code) == code
 
     def test_inline_code_preserved(self, adapter):
@@ -4004,6 +4100,16 @@ def test_pre_escaped_gt_not_double_escaped(self, adapter):
         """Already-escaped &gt; in plain text must not become &amp;gt;."""
         assert adapter.format_message("5 &gt; 3") == "5 &gt; 3"
 
+    def test_escaped_entity_text_not_double_decoded(self, adapter):
+        """&amp;lt; is the wire form of the literal text &lt; — it must survive.
+
+        The unescape pass must not re-scan its own output: decoding &amp; to &
+        first must not let the resulting & combine with a following lt; into a
+        second decode, or the literal text is silently destroyed.
+        """
+        assert adapter.format_message("&amp;lt;") == "&amp;lt;"
+        assert adapter.format_message("&amp;gt;") == "&amp;gt;"
+
     def test_mixed_raw_and_escaped_entities(self, adapter):
         """Raw & and pre-escaped &amp; coexist correctly."""
         result = adapter.format_message("AT&T and &amp; entity")
@@ -4069,9 +4175,9 @@ def test_channel_link_preserved(self, adapter):
     # --- Additional edge cases ---
 
     def test_message_only_code_block(self, adapter):
-        """Entire message is a fenced code block — no conversion."""
+        """Entire message is a fenced code block — body preserved, lang tag dropped."""
         code = "```python\nx = 1\n```"
-        assert adapter.format_message(code) == code
+        assert adapter.format_message(code) == "```\nx = 1\n```"
 
     def test_multiline_mixed_formatting(self, adapter):
         """Multi-line message with headers, bold, links, code, and blockquotes."""
@@ -4265,7 +4371,8 @@ async def test_edit_message_formats_streaming_updates(self, adapter):
         )
         assert result2.success is True
         kwargs2 = adapter._app.client.chat_update.call_args.kwargs
-        assert kwargs2["text"] == "*Done!* See <https://example.com|results>"
+        # ZWSP guard (#35144): bold ending in non-word char gets U+200B before closing *
+        assert kwargs2["text"] == "*Done!\u200b* See <https://example.com|results>"
 
     @pytest.mark.asyncio
     async def test_edit_message_formats_code_and_bold(self, adapter):
@@ -4276,8 +4383,10 @@ async def test_edit_message_formats_code_and_bold(self, adapter):
         result = await adapter.edit_message("C123", "ts1", content)
         assert result.success is True
         kwargs = adapter._app.client.chat_update.call_args.kwargs
-        assert kwargs["text"].startswith("*Result:*")
-        assert "```python\nprint('hello')\n```" in kwargs["text"]
+        # ZWSP guard (#35144): trailing ":" inside bold gets U+200B before closing *
+        assert kwargs["text"].startswith("*Result:\u200b*")
+        # Language tag is stripped — Slack mrkdwn would render it as a literal line
+        assert "```\nprint('hello')\n```" in kwargs["text"]
 
     @pytest.mark.asyncio
     async def test_edit_message_formats_blockquote_in_stream(self, adapter):
@@ -4288,7 +4397,8 @@ async def test_edit_message_formats_blockquote_in_stream(self, adapter):
         result = await adapter.edit_message("C123", "ts1", content)
         assert result.success is True
         kwargs = adapter._app.client.chat_update.call_args.kwargs
-        assert kwargs["text"].startswith("> *Important:*")
+        # ZWSP guard (#35144): trailing ":" inside bold gets U+200B before closing *
+        assert kwargs["text"].startswith("> *Important:\u200b*")
         assert "normal line" in kwargs["text"]
 
     @pytest.mark.asyncio
@@ -7968,3 +8078,213 @@ async def test_delta_refresh_marks_new_images(
         assert "[image: fresh.png]" in msg_event.channel_context
         # No cold-start hydrate → no root image download.
         a._download_slack_file.assert_not_called()
+
+# =========================================================================
+# Markdown table preprocessing (Slack mrkdwn does not render GFM tables)
+# =========================================================================
+
+from plugins.platforms.slack.adapter import (  # noqa: E402
+    _wrap_markdown_tables,
+    _align_table,
+    _disp_width,
+    _is_table_row,
+)
+
+
+class TestWrapMarkdownTables:
+    """``_wrap_markdown_tables`` wraps GFM pipe tables in ``` fences AND
+    aligns columns by per-column max display width, so Slack monospace
+    code-block rendering shows readable, aligned columns even with CJK
+    content (mirrors the TUI rendering)."""
+
+    def test_basic_table_wrapped(self):
+        text = (
+            "Scores:\n\n"
+            "| Player | Score |\n"
+            "|--------|-------|\n"
+            "| Alice  | 150   |\n"
+            "| Bob    | 120   |\n"
+            "\nEnd."
+        )
+        out = _wrap_markdown_tables(text)
+        # Wrapped in fence
+        assert "```\n| Player" in out
+        assert out.count("```") == 2
+        # Surrounding prose preserved
+        assert out.startswith("Scores:")
+        assert out.endswith("End.")
+
+    def test_columns_aligned_after_wrap(self):
+        """All rows in the wrapped block should have identical character length."""
+        text = (
+            "| short | long_header_name |\n"
+            "|---|---|\n"
+            "| a | bbb |"
+        )
+        out = _wrap_markdown_tables(text)
+        body = [ln for ln in out.split("\n") if ln.startswith("|")]
+        widths = {len(ln) for ln in body}
+        assert len(widths) == 1, f"row widths drift: {widths}"
+
+    def test_cjk_columns_aligned(self):
+        """CJK characters count as 2 display columns; alignment must respect that."""
+        text = (
+            "| Workflow | 状态 |\n"
+            "|---|---|\n"
+            "| ci | active |\n"
+            "| dep | 7 成功 |"
+        )
+        out = _wrap_markdown_tables(text)
+        body = [ln for ln in out.split("\n") if ln.startswith("|")]
+        # Display widths (not raw char counts) should be uniform
+        display_widths = {_disp_width(ln) for ln in body}
+        assert len(display_widths) == 1, f"display widths drift: {display_widths}"
+
+    def test_no_table_returns_unchanged(self):
+        text = "Just a paragraph with | one pipe but no table."
+        assert _wrap_markdown_tables(text) == text
+
+    def test_table_inside_existing_fence_untouched(self):
+        text = (
+            "```\n"
+            "| inside | a fence |\n"
+            "|---|---|\n"
+            "| x | y |\n"
+            "```"
+        )
+        # Content already inside ``` should be passed through verbatim.
+        assert _wrap_markdown_tables(text) == text
+
+    def test_alignment_separators_supported(self):
+        """Separator rows with :--- / ---: / :---: alignment markers match."""
+        text = (
+            "| Name | Age | City |\n"
+            "|:-----|----:|:----:|\n"
+            "| Ada  |  30 | NYC  |"
+        )
+        out = _wrap_markdown_tables(text)
+        assert out.count("```") == 2
+
+    def test_two_consecutive_tables_wrapped_separately(self):
+        text = (
+            "| A | B |\n|---|---|\n| 1 | 2 |\n"
+            "\n"
+            "| C | D |\n|---|---|\n| 3 | 4 |"
+        )
+        out = _wrap_markdown_tables(text)
+        # Two separate fence pairs (4 ``` total)
+        assert out.count("```") == 4
+
+    def test_bare_pipe_table_wrapped(self):
+        """Tables without outer pipes (GFM allows this) are still detected."""
+        text = "head1 | head2\n--- | ---\na | b\nc | d"
+        out = _wrap_markdown_tables(text)
+        assert out.count("```") == 2
+        assert "head1" in out
+
+    def test_empty_input(self):
+        assert _wrap_markdown_tables("") == ""
+
+    def test_single_pipe_no_table(self):
+        text = "this | that"  # no separator row → not a table
+        assert _wrap_markdown_tables(text) == text
+
+
+class TestAlignTable:
+    def test_normalizes_column_count(self):
+        """Rows with mismatched column counts get padded to the max."""
+        rows = [
+            "| a | b |",
+            "|---|---|",
+            "| 1 |",            # short
+            "| 2 | 3 | extra |",  # long
+        ]
+        out = _align_table(rows)
+        # All rows should have same number of `|` chars after padding
+        pipe_counts = {ln.count("|") for ln in out}
+        assert len(pipe_counts) == 1
+
+    def test_pads_to_max_display_width(self):
+        rows = [
+            "| short | longer_header |",
+            "|---|---|",
+            "| a | b |",
+        ]
+        out = _align_table(rows)
+        # All output rows have same character length
+        assert len({len(ln) for ln in out}) == 1
+
+    def test_regenerates_separator_row(self):
+        """Separator row is regenerated to match the (wider) column widths."""
+        rows = [
+            "| short | longer_header |",
+            "|---|---|",
+            "| a | b |",
+        ]
+        out = _align_table(rows)
+        sep = out[1]
+        # Original separator was 6 dashes total; the new one must be longer
+        assert sep.count("-") > 6
+
+    def test_too_few_rows_returned_unchanged(self):
+        rows = ["| only header |"]
+        assert _align_table(rows) == rows
+
+
+class TestDispWidth:
+    def test_ascii_one_per_char(self):
+        assert _disp_width("hello") == 5
+
+    def test_empty_string(self):
+        assert _disp_width("") == 0
+
+    def test_cjk_two_per_char(self):
+        assert _disp_width("成功") == 4
+        assert _disp_width("过去") == 4
+
+    def test_mixed_ascii_and_cjk(self):
+        # "5 成功" = 1 + 1 + 2 + 2 = 6
+        assert _disp_width("5 成功") == 6
+
+    def test_full_width_punctuation(self):
+        # ， is U+FF0C (full-width comma), east_asian_width = F
+        assert _disp_width("a，b") == 4  # 1 + 2 + 1
+
+
+class TestIsTableRow:
+    def test_recognizes_pipe_row(self):
+        assert _is_table_row("| a | b |") is True
+
+    def test_rejects_blank(self):
+        assert _is_table_row("") is False
+        assert _is_table_row("   ") is False
+
+    def test_rejects_no_pipe(self):
+        assert _is_table_row("just text") is False
+
+
+class TestFormatMessageTableIntegration:
+    """format_message() routes GFM tables through the fence-wrap path."""
+
+    @pytest.fixture
+    def adapter(self):
+        config = PlatformConfig(enabled=True, extra={})
+        a = SlackAdapter.__new__(SlackAdapter)
+        a.config = config
+        return a
+
+    def test_table_wrapped_and_protected(self, adapter):
+        text = "| a | b |\n|---|---|\n| **1** | 2 |"
+        out = adapter.format_message(text)
+        # Wrapped in a fence and protected from mrkdwn conversion:
+        assert out.count("```") == 2
+        assert "**1**" in out  # bold markers inside the fence stay literal
+
+    def test_table_fence_carries_no_language_tag(self, adapter):
+        """The emitted table fence must survive the lang-tag strip pass."""
+        text = "| a | b |\n|---|---|\n| 1 | 2 |"
+        out = adapter.format_message(text)
+        first_fence_line = next(
+            ln for ln in out.split("\n") if ln.startswith("```")
+        )
+        assert first_fence_line == "```"
diff --git a/tests/gateway/test_slack_block_kit.py b/tests/gateway/test_slack_block_kit.py
index d745ccdefde7..6ecd976eaf31 100644
--- a/tests/gateway/test_slack_block_kit.py
+++ b/tests/gateway/test_slack_block_kit.py
@@ -429,3 +429,34 @@ def test_payload_capped_at_50_blocks(self):
 
     def test_never_raises_on_garbage(self):
         assert sanitize_blocks([{"no_type": True}, "not-a-dict", 42]) is None
+
+
+class TestSplitTextFenceBalanced:
+    """_split_text closes/reopens ``` fences at section chunk boundaries."""
+
+    def test_fenced_split_every_chunk_balanced(self):
+        from plugins.platforms.slack.block_kit import _split_text
+
+        text = "```\n" + "\n".join("y" * 20 for _ in range(30)) + "\n```"
+        chunks = _split_text(text, 100)
+        assert len(chunks) >= 2
+        for i, chunk in enumerate(chunks):
+            assert chunk.count("```") % 2 == 0, (
+                f"chunk {i} has unbalanced fences: {chunk[:60]!r}"
+            )
+
+    def test_fenced_split_respects_limit(self):
+        from plugins.platforms.slack.block_kit import _split_text
+
+        text = "```\n" + "\n".join("y" * 20 for _ in range(30)) + "\n```"
+        limit = 100
+        for chunk in _split_text(text, limit):
+            assert len(chunk) <= limit
+
+    def test_prose_split_unchanged(self):
+        from plugins.platforms.slack.block_kit import _split_text
+
+        text = "\n".join(f"line {i}" for i in range(60))
+        chunks = _split_text(text, 80)
+        assert len(chunks) >= 2
+        assert all("```" not in c for c in chunks)
diff --git a/tests/gateway/test_slack_block_kit_adapter.py b/tests/gateway/test_slack_block_kit_adapter.py
index 59351d72e916..0923e34c787b 100644
--- a/tests/gateway/test_slack_block_kit_adapter.py
+++ b/tests/gateway/test_slack_block_kit_adapter.py
@@ -341,3 +341,91 @@ def ensure_and_bind(_group, import_fn, target_globals, *, prompt):
         assert result.success is False
         assert result.retryable is True
         assert result.error_kind == "transient"
+
+
+# ---------------------------------------------------------------------------
+# markdown_blocks mode — Slack's native ``markdown`` Block Kit block (#8552)
+# ---------------------------------------------------------------------------
+
+
+class TestMarkdownBlockMode:
+    """Opt-in ``markdown_blocks`` renders raw standard markdown via Slack's
+    native ``markdown`` block, keeping the mrkdwn ``text`` fallback."""
+
+    @pytest.mark.asyncio
+    async def test_disabled_by_default(self):
+        adapter, client = _make_adapter()
+        await adapter.send("C1", RICH_TABLE_MD)
+        kwargs = client.chat_postMessage.await_args.kwargs
+        assert "blocks" not in kwargs
+
+    @pytest.mark.asyncio
+    async def test_enabled_sends_markdown_block_with_raw_content(self):
+        adapter, client = _make_adapter({"markdown_blocks": True})
+        await adapter.send("C1", RICH_TABLE_MD)
+        kwargs = client.chat_postMessage.await_args.kwargs
+        blocks = kwargs["blocks"]
+        assert blocks[0]["type"] == "markdown"
+        # RAW standard markdown, not mrkdwn-converted — Slack translates it
+        assert blocks[0]["text"] == RICH_TABLE_MD
+        # mrkdwn fallback text is still present for notifications/search
+        assert kwargs["text"]
+
+    @pytest.mark.asyncio
+    async def test_text_fallback_is_mrkdwn_converted(self):
+        adapter, client = _make_adapter({"markdown_blocks": True})
+        await adapter.send("C1", "**bold**")
+        kwargs = client.chat_postMessage.await_args.kwargs
+        assert kwargs["blocks"][0]["text"] == "**bold**"
+        assert kwargs["text"] == "*bold*"  # mrkdwn conversion for fallback
+
+    @pytest.mark.asyncio
+    async def test_markdown_block_preferred_over_rich_blocks(self):
+        adapter, client = _make_adapter(
+            {"markdown_blocks": True, "rich_blocks": True}
+        )
+        await adapter.send("C1", RICH_TABLE_MD)
+        blocks = client.chat_postMessage.await_args.kwargs["blocks"]
+        assert blocks[0]["type"] == "markdown"
+
+    @pytest.mark.asyncio
+    async def test_over_cap_falls_back_to_rich_or_text(self):
+        adapter, client = _make_adapter({"markdown_blocks": True})
+        big = "x" * (SlackAdapter._MARKDOWN_BLOCK_MAX + 1)
+        payload = adapter._markdown_block_payload(big)
+        assert payload is None  # declines >12k cumulative markdown cap
+
+    @pytest.mark.asyncio
+    async def test_rejection_retries_without_blocks(self):
+        """Workspaces/surfaces without markdown-block support degrade to
+        the plain mrkdwn text payload instead of dropping the message."""
+        adapter, client = _make_adapter({"markdown_blocks": True})
+        client.chat_postMessage = AsyncMock(
+            side_effect=[SlackRejectedBlocks(), {"ts": "111.222"}]
+        )
+        result = await adapter.send("C1", RICH_TABLE_MD)
+        assert result.success is True
+        assert client.chat_postMessage.await_count == 2
+        retry_kwargs = client.chat_postMessage.await_args_list[1].kwargs
+        assert "blocks" not in retry_kwargs
+        assert retry_kwargs["text"]
+
+    @pytest.mark.asyncio
+    async def test_edit_finalize_uses_markdown_block(self):
+        adapter, client = _make_adapter({"markdown_blocks": True})
+        await adapter.edit_message("C1", "111.222", RICH_TABLE_MD, finalize=True)
+        kwargs = client.chat_update.await_args.kwargs
+        assert kwargs["blocks"][0]["type"] == "markdown"
+        assert kwargs["blocks"][0]["text"] == RICH_TABLE_MD
+
+    @pytest.mark.asyncio
+    async def test_edit_streaming_stays_plain(self):
+        adapter, client = _make_adapter({"markdown_blocks": True})
+        await adapter.edit_message("C1", "111.222", RICH_TABLE_MD, finalize=False)
+        kwargs = client.chat_update.await_args.kwargs
+        assert "blocks" not in kwargs
+
+    def test_empty_content_declines(self):
+        adapter, _ = _make_adapter({"markdown_blocks": True})
+        assert adapter._markdown_block_payload("") is None
+        assert adapter._markdown_block_payload("   ") is None
diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py
index f1b8ffa399cd..db28bdcf04b5 100644
--- a/tests/gateway/test_stream_consumer.py
+++ b/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):