feat(gateway): structured stream-event protocol + Telegram draft formatting parity

gateway/config.py

@@ -361,10 +361,17 @@ class StreamingConfig:
     #             fall back to edit-based when not.
     #   "draft" — explicitly request native drafts; falls back to edit when
     #             the platform/chat doesn't support them.
-    #   "edit"  — progressive editMessageText only (legacy/default
-    #             behaviour).
+    #   "edit"  — progressive editMessageText only (legacy behaviour).
     #   "off"   — disable streaming entirely.
-    transport: str = "edit"
+    #
+    # Default is "auto": prefer native draft streaming on platforms that
+    # support it (Telegram DMs via sendMessageDraft, Bot API 9.5+) and fall
+    # back to edit-based streaming everywhere else.  This is safe as a global
+    # default because adapters without draft support (Discord, Slack, Matrix,
+    # …) report supports_draft_streaming() == False and transparently use the
+    # edit path — so "auto" never regresses non-Telegram platforms, it only
+    # upgrades the chats that can render the smoother native preview.
+    transport: str = "auto"
     edit_interval: float = DEFAULT_STREAMING_EDIT_INTERVAL
     buffer_threshold: int = DEFAULT_STREAMING_BUFFER_THRESHOLD
     cursor: str = DEFAULT_STREAMING_CURSOR
@@ -393,7 +400,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "StreamingConfig":
             return cls()
         return cls(
             enabled=_coerce_bool(data.get("enabled"), False),
-            transport=data.get("transport", "edit"),
+            transport=data.get("transport", "auto"),
             edit_interval=_coerce_float(
                 data.get("edit_interval"), DEFAULT_STREAMING_EDIT_INTERVAL,
             ),

gateway/platforms/base.py

@@ -1916,6 +1916,84 @@ async def send_draft(
             f"{type(self).__name__} does not implement send_draft"
         )
 
+    # ── Structured stream-event rendering ────────────────────────────────
+    #
+    # These methods let an adapter decide *how* to present each structured
+    # streaming event (see gateway/stream_events.py).  The default
+    # implementations reproduce the historical behavior exactly: assistant
+    # text/commentary/segment events delegate to the stream consumer, and
+    # tool events render the same "emoji tool_name: preview" chrome the
+    # gateway has always produced.  Adapters override these to be more native
+    # to their platform (e.g. Telegram streaming a MarkdownV2 ```bash``` block
+    # as a draft; iMessage eating tool chrome it cannot format).
+    #
+    # The contract is presentation-only: nothing rendered here is persisted to
+    # conversation history.  History is owned by the agent; what an adapter
+    # chooses to "eat" must never change the bytes the agent stored.
+
+    def render_message_event(self, event: Any, sink: Any) -> None:
+        """Render a MessageChunk / MessageStop / Commentary onto the sink.
+
+        Default: map onto the stream consumer's existing primitives, preserving
+        today's behavior 1:1.  ``sink`` is a GatewayStreamConsumer.
+        """
+        from gateway.stream_events import MessageChunk, MessageStop, Commentary
+
+        if isinstance(event, MessageChunk):
+            if event.text:
+                sink.on_delta(event.text)
+        elif isinstance(event, MessageStop):
+            # An intermediate stop (text → tool → text) is a segment break;
+            # the terminal stop is signalled by the gateway via finish(),
+            # not here, so we only break segments on non-final stops.
+            if not event.final:
+                sink.on_segment_break()
+        elif isinstance(event, Commentary):
+            if event.text:
+                sink.on_commentary(event.text)
+
+    def format_tool_event(self, event: Any, *, mode: str = "all",
+                          preview_max_len: int = 40) -> Optional[str]:
+        """Return the rendered chrome for a ToolCallChunk, or None to eat it.
+
+        Reproduces the gateway's historical tool-progress formatting: an emoji
+        for the tool, the tool name, and a short argument preview (or the full
+        args dict in ``verbose`` mode).  Adapters that cannot render tool chrome
+        (no message editing, plain-text only) should override to return None so
+        the event is dropped rather than spamming separate bubbles.
+
+        ``mode`` is the resolved tool-progress mode ("all" / "new" / "verbose");
+        ``preview_max_len`` mirrors the ``tool_preview_length`` config (0 means
+        "no cap" in verbose mode).
+        """
+        from gateway.stream_events import ToolCallChunk
+        if not isinstance(event, ToolCallChunk):
+            return None
+
+        from agent.display import get_tool_emoji
+        emoji = get_tool_emoji(event.tool_name, default="⚙️")
+
+        if mode == "verbose":
+            if event.args:
+                import json
+                args_str = json.dumps(event.args, ensure_ascii=False, default=str)
+                if preview_max_len > 0 and len(args_str) > preview_max_len:
+                    args_str = args_str[:preview_max_len - 3] + "..."
+                return f"{emoji} {event.tool_name}({list(event.args.keys())})\n{args_str}"
+            if event.preview:
+                return f"{emoji} {event.tool_name}: \"{event.preview}\""
+            return f"{emoji} {event.tool_name}..."
+
+        # "all" / "new": short preview, capped (default 40 to keep gateway
+        # progress bubbles compact — they persist as permanent messages).
+        preview = event.preview
+        if preview:
+            cap = preview_max_len if preview_max_len > 0 else 40
+            if len(preview) > cap:
+                preview = preview[:cap - 3] + "..."
+            return f"{emoji} {event.tool_name}: \"{preview}\""
+        return f"{emoji} {event.tool_name}..."
+
     @property
     def has_fatal_error(self) -> bool:
         return self._fatal_error_message is not None

gateway/platforms/telegram.py

@@ -2521,31 +2521,55 @@ async def send_draft(
         text = content if len(content) <= self.MAX_MESSAGE_LENGTH else \
             self.truncate_message(content, self.MAX_MESSAGE_LENGTH, len_fn=utf16_len)[0]
 
-        kwargs: Dict[str, Any] = {
-            "chat_id": int(chat_id),
-            "draft_id": int(draft_id),
-            "text": text,
-        }
         thread_id = self._metadata_thread_id(metadata)
-        if thread_id is not None:
-            kwargs["message_thread_id"] = thread_id
 
-        try:
-            ok = await self._bot.send_message_draft(**kwargs)
-            if ok:
-                # Drafts have no message_id; we report success without one
-                # so the caller knows the animation frame landed.
-                return SendResult(success=True, message_id=None)
-            return SendResult(success=False, error="draft_rejected")
-        except Exception as e:
-            # Most likely: BadRequest because this bot/chat doesn't allow
-            # drafts, or a transient server hiccup.  The caller treats any
-            # failure as "fall back to edit-based for this response".
-            logger.debug(
-                "[%s] sendMessageDraft failed (chat=%s draft_id=%s): %s",
-                self.name, chat_id, draft_id, e,
-            )
-            return SendResult(success=False, error=str(e))
+        # Apply the same MarkdownV2 conversion the regular ``send`` path uses
+        # so the animated draft preview renders with identical formatting to
+        # the final message.  Without this, the draft streams as raw text and
+        # the final ``sendMessage`` (which DOES use MarkdownV2) snaps into
+        # formatted output, producing a jarring visual shift at the end of the
+        # response.  We try MarkdownV2 first and fall back to plain text if a
+        # malformed escape would be rejected — mirroring the (True, False)
+        # retry the streaming send loop uses — so a single bad token never
+        # kills draft streaming for the whole response.
+        for use_markdown in (True, False):
+            kwargs: Dict[str, Any] = {
+                "chat_id": int(chat_id),
+                "draft_id": int(draft_id),
+                "text": self.format_message(text) if use_markdown else text,
+            }
+            if use_markdown:
+                kwargs["parse_mode"] = ParseMode.MARKDOWN_V2
+            if thread_id is not None:
+                kwargs["message_thread_id"] = thread_id
+
+            try:
+                ok = await self._bot.send_message_draft(**kwargs)
+                if ok:
+                    # Drafts have no message_id; we report success without one
+                    # so the caller knows the animation frame landed.
+                    return SendResult(success=True, message_id=None)
+                return SendResult(success=False, error="draft_rejected")
+            except Exception as e:
+                # A MarkdownV2 parse failure (BadRequest "can't parse entities")
+                # is recoverable: retry once as plain text.  Any other failure
+                # (chat doesn't allow drafts, transient hiccup) — or a failure
+                # on the plain-text attempt — propagates to the caller, which
+                # treats it as "fall back to edit-based for this response".
+                if use_markdown and self._is_bad_request_error(e):
+                    logger.debug(
+                        "[%s] sendMessageDraft MarkdownV2 rejected, retrying "
+                        "as plain text (chat=%s draft_id=%s): %s",
+                        self.name, chat_id, draft_id, e,
+                    )
+                    continue
+                logger.debug(
+                    "[%s] sendMessageDraft failed (chat=%s draft_id=%s): %s",
+                    self.name, chat_id, draft_id, e,
+                )
+                return SendResult(success=False, error=str(e))
+
+        return SendResult(success=False, error="draft_rejected")
 
     async def _send_message_with_thread_fallback(self, **kwargs):
         """Send a Telegram message, retrying once without message_thread_id

gateway/stream_dispatch.py

@@ -0,0 +1,132 @@
+"""Adapter-driven dispatch of structured stream events to a delivery sink.
+
+``GatewayEventDispatcher`` is the seam Tobi asked for: the agent emits typed
+events (gateway/stream_events.py), and the *adapter* decides how each one is
+delivered.  The dispatcher holds an adapter + the stream consumer (sink) + the
+resolved per-channel presentation settings (tool-progress mode, preview length)
+and routes each event through the adapter's render hooks.
+
+Message/commentary/segment events flow into the consumer (native draft on
+Telegram DMs, edit-in-place elsewhere).  Tool events are formatted by the
+adapter — which may return None to *eat* the event on platforms that can't
+render tool chrome — and the rendered line is enqueued onto the same tool
+progress queue the gateway already drains, so the two no longer race through
+independent code paths.
+
+This module deliberately has no platform knowledge and no asyncio: it is a thin
+synchronous router callable from the agent's worker thread, exactly like the
+callbacks it replaces.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable, Optional
+
+from gateway.stream_events import (
+    Commentary,
+    GatewayNotice,
+    LongToolHint,
+    MessageChunk,
+    MessageStop,
+    StreamEvent,
+    ToolCallChunk,
+    ToolCallFinished,
+)
+
+logger = logging.getLogger("gateway.stream_events")
+
+
+class GatewayEventDispatcher:
+    """Route typed stream events through an adapter onto a delivery sink.
+
+    Parameters
+    ----------
+    adapter:
+        The platform adapter.  Provides ``render_message_event`` and
+        ``format_tool_event`` (BasePlatformAdapter defaults reproduce today's
+        behavior; adapters may override for native rendering).
+    sink:
+        The GatewayStreamConsumer for assistant-text delivery.  May be None
+        when streaming is disabled, in which case message events are dropped
+        (the final response still goes out via the normal send path).
+    enqueue_tool_line:
+        Callback that places a rendered tool-progress line onto the gateway's
+        progress queue (the same queue ``send_progress_messages`` drains).  May
+        be None when tool progress is disabled for this channel.
+    tool_mode:
+        Resolved tool-progress mode for this channel ("all" / "new" / "verbose"
+        / "off").
+    preview_max_len:
+        Resolved ``tool_preview_length`` (0 = no cap in verbose mode).
+    on_long_tool / on_notice:
+        Optional hooks for LongToolHint / GatewayNotice events, letting the
+        gateway own the "should I surface this here?" decision.
+    """
+
+    def __init__(
+        self,
+        adapter: Any,
+        sink: Any = None,
+        *,
+        enqueue_tool_line: Optional[Callable[[Any], None]] = None,
+        tool_mode: str = "all",
+        preview_max_len: int = 40,
+        on_long_tool: Optional[Callable[[LongToolHint], None]] = None,
+        on_notice: Optional[Callable[[GatewayNotice], None]] = None,
+    ) -> None:
+        self.adapter = adapter
+        self.sink = sink
+        self._enqueue_tool_line = enqueue_tool_line
+        self.tool_mode = tool_mode or "all"
+        self.preview_max_len = preview_max_len
+        self._on_long_tool = on_long_tool
+        self._on_notice = on_notice
+        # "new" mode dedup — only report when the tool changes.
+        self._last_tool: Optional[str] = None
+
+    def dispatch(self, event: StreamEvent) -> None:
+        """Route a single event.  Never raises into the agent's worker thread."""
+        try:
+            self._dispatch(event)
+        except Exception:  # presentation must never break the agent loop
+            logger.debug("stream-event dispatch error", exc_info=True)
+
+    def _dispatch(self, event: StreamEvent) -> None:
+        if isinstance(event, (MessageChunk, MessageStop, Commentary)):
+            if self.sink is not None:
+                self.adapter.render_message_event(event, self.sink)
+            return
+
+        if isinstance(event, ToolCallChunk):
+            if self.tool_mode == "off" or self._enqueue_tool_line is None:
+                return
+            # "new" mode: only emit when the tool changes.
+            if self.tool_mode == "new" and event.tool_name == self._last_tool:
+                return
+            self._last_tool = event.tool_name
+            line = self.adapter.format_tool_event(
+                event, mode=self.tool_mode, preview_max_len=self.preview_max_len,
+            )
+            # None == adapter chose to eat this event (can't render tool chrome).
+            if line:
+                self._enqueue_tool_line(line)
+            return
+
+        if isinstance(event, ToolCallFinished):
+            # Default: no chrome on completion (matches today — the gateway only
+            # rendered "started" events).  Completion drives onboarding hints.
+            return
+
+        if isinstance(event, LongToolHint):
+            if self._on_long_tool is not None:
+                self._on_long_tool(event)
+            return
+
+        if isinstance(event, GatewayNotice):
+            if self._on_notice is not None:
+                self._on_notice(event)
+            return
+
+
+__all__ = ["GatewayEventDispatcher"]