Telegram MarkdownV2 formatting and chunking fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #43761 fix(telegram): stop cutting long streamed responses (lost tails, deleted heads, raw markdown)

  merged to main

  files: .github/pr-screenshots/telegram-overflow/topic-final-response-clipped.jpg, docs/plans/2026-06-09-003-fix-telegram-stream-overflow-continuations-plan.md, gateway/platforms/base.py, gateway/platforms/telegram.py, gateway/stream_consumer.py, tests/gateway/test_stream_consumer.py, tests/gateway/test_stream_consumer_fresh_final.py, tests/gateway/test_telegram_overflow_partial.py

  Show the diff

  • docs/plans/2026-06-09-003-fix-telegram-stream-overflow-continuations-plan.md

    @@ -0,0 +1,240 @@
    +---
    +title: "fix: Prevent Telegram streamed replies from ending after first overflow chunk"
    +status: active
    +date: 2026-06-09
    +type: fix
    +target_repo: hermes-agent
    +origin: user-reported Telegram topic screenshot
    +---
    +
    +# fix: Prevent Telegram streamed replies from ending after first overflow chunk
    +
    +## Summary
    +
    +Fix a Telegram gateway bug where a long streamed assistant reply can appear to stop mid-answer in a topic after the first overflow chunk. The reported screenshot shows a long Hermes response in the `Nehemiah - Coding` Telegram topic ending at `- The visible tool-call summary`, followed by the user noting that the previous message did not finish streaming to that Telegram topic.
    +
    +The plan targets the streamed edit overflow path, not general model generation. A completed assistant response must either reach Telegram in full across all continuation messages or leave enough state for the gateway fallback path to deliver the remaining content instead of marking the turn complete after a partial delivery.
    +
    +---
    +
    +## Problem Frame
    +
    +Telegram limits message text to 4096 UTF-16 code units. Hermes streams gateway responses by editing a message and, when a streamed message grows past the limit, splitting the overflow into additional Telegram messages. The adapter already has a split-and-deliver path for oversized edits, but the partial-continuation failure contract is weak: if chunk 1 is edited successfully and a later continuation fails, the adapter can still report success for the operation. The stream consumer may then mark the final response delivered even though the visible topic only contains the first part.
    +
    +This is especially visible in Telegram forum topics because a long final response can be split below tool-progress bubbles, and a missing continuation looks exactly like the stream stopped mid-answer.
    +
    +---
    +
    +## Requirements
    +
    +- R1. Long streamed Telegram replies must preserve all final content across overflow chunks.
    +- R2. If any continuation chunk fails after the first overflow edit lands, the gateway must not mark the final response as fully delivered.
    +- R3. Continuation chunks must remain routed to the same Telegram topic/thread as the original response.
    +- R4. The fix must avoid duplicate full-answer sends when all overflow chunks were delivered successfully.
    +- R5. Tests must cover the reported failure shape: a final streamed reply that exceeds Telegram's limit, succeeds on the first edit, fails on a continuation, and must not be treated as complete.
    +
    +---
    +
    +## Key Technical Decisions
    +
    +- Treat overflow delivery as all-or-not-complete. `_edit_overflow_split` should only return a successful final-delivery result when every planned chunk reaches Telegram. Partial delivery is a distinct outcome that downstream code can recover from.
    +- Carry partial-overflow metadata through `SendResult.raw_response` rather than adding a new public dataclass field unless implementation proves the existing result shape is insufficient. The stream consumer already inspects `SendResult` after adapter edits, so a small raw response contract can keep the change contained.
    +- Make the stream consumer responsible for final-delivery truth. The adapter knows which chunks landed, but the consumer owns `_final_response_sent`, `_final_content_delivered`, `_fallback_prefix`, and fallback final-send behaviour.
    +- Keep routing inside Telegram adapter helpers. Continuation sends should continue to use `_thread_kwargs_for_send(...)` with metadata-derived `message_thread_id` and reply anchors so forum topic behaviour stays consistent.
    +
    +---
    +
    +## High-Level Technical Design
    +
    +```mermaid
    +sequenceDiagram
    +    participant C as GatewayStreamConsumer
    +    participant T as TelegramAdapter.edit_message
    +    participant B as Telegram Bot API
    +
    +    C->>T: finalize/edit long accumulated response
    +    T->>B: edit original message with chunk 1
    +    loop remaining chunks
    +        T->>B: send continuation in same topic/thread
    +    end
    +    alt all chunks delivered
    +        T-->>C: success, last message id, continuation ids
    +        C->>C: mark final response delivered
    +    else any continuation failed
    +        T-->>C: partial overflow failure with delivered prefix metadata
    +        C->>C: do not mark final delivered
    +        C->>B: fallback sends missing tail or full final response safely
    +    end
    +```
    +
    +---
    +
    +## Implementation Units
    +
    +### U1. Add a partial-overflow contract for Telegram edit splits
    +
    +**Goal:** Make `TelegramAdapter._edit_overflow_split` distinguish complete overflow delivery from partial delivery.
    +
    +**Requirements:** R1, R2, R4
    +
    +**Dependencies:** None
    +
    +**Files:**
    +- `gateway/platforms/telegram.py`
    +- `tests/gateway/test_telegram_send.py` or the existing Telegram adapter test module that already covers `edit_message` overflow behaviour
    +
    +**Approach:**
    +- Keep the successful path unchanged when every chunk is delivered: return `SendResult(success=True, message_id=<last chunk>, continuation_message_ids=(...))`.
    +- When a continuation fails after the first edit, return a result that clearly indicates partial delivery instead of plain success. Prefer `success=False`, `retryable=True`, and `raw_response` metadata such as delivered chunk count, total chunk count, last delivered message id, and the visible delivered prefix.
    +- Preserve logging, but do not rely on logs as the only signal. The caller must be able to tell partial delivery happened.
    +- Ensure the first edited chunk and all successful continuation chunks still include the existing Markdown/plain-text fallback behaviour.
    +
    +**Patterns to follow:**
    +- Existing overflow handling in `TelegramAdapter.edit_message` and `_edit_overflow_split`.
    +- Existing `SendResult` semantics in `gateway/platforms/base.py`, especially `retryable`, `raw_response`, and `continuation_message_ids`.
    +
    +**Test scenarios:**
    +- Oversized finalized edit where all continuations succeed returns success, the last continuation id, and all continuation ids.
    +- Oversized finalized edit where the first continuation send fails returns a partial-overflow failure and does not report success.
    +- Oversized finalized edit where one continuation succeeds and a later continuation fails reports the last delivered continuation id and delivered count in raw metadata.
    +- A continuation MarkdownV2 formatting failure still retries plain text before being treated as a delivery failure.
    +
    +**Verification:** Adapter tests prove complete overflow remains successful and partial overflow is observable by the caller.
    +
    +### U2. Teach the stream consumer to recover from partial overflow
    +
    +**Goal:** Ensure a partial Telegram overflow does not set `_final_response_sent` or `_final_content_delivered` unless the full response reached the user.
    +
    +**Requirements:** R1, R2, R4, R5
    +
    +**Dependencies:** U1
    +
    +**Files:**
    +- `gateway/stream_consumer.py`
    +- `tests/gateway/test_stream_consumer.py` or a focused new `tests/gateway/test_stream_consumer_telegram_overflow.py`
    +
    +**Approach:**
    +- In `_send_or_edit`, when `adapter.edit_message(...)` returns a partial-overflow failure, update consumer state to reflect the last visible prefix/message and enter fallback delivery for the missing content.
    +- Avoid treating `_already_sent` as final delivery. A partial visible message can be true while final delivery is false.
    +- Use the delivered-prefix metadata if available so `_send_fallback_final(...)` sends only the missing tail. If implementation finds the prefix is unreliable after Markdown formatting, prefer sending the complete final response as a fresh fallback message rather than silently dropping the tail.
    +- Keep the existing success handling for `continuation_message_ids` when the adapter delivered all chunks.
    +
    +**Patterns to follow:**
    +- Existing fallback mode in `GatewayStreamConsumer._send_or_edit` and `_send_fallback_final`.
    +- Existing comments around `_final_response_sent`, `_final_content_delivered`, and `_fallback_prefix` for prior partial-delivery regressions.
    +
    +**Test scenarios:**
    +- A final streamed response that overflows and receives a complete-success edit split sets final-delivery flags and does not invoke fallback.
    +- A final streamed response whose adapter reports partial overflow does not set final-delivery flags immediately.
    +- After partial overflow, fallback delivery sends the remaining tail and then marks final content delivered only if the fallback send succeeds.
    +- If fallback delivery also fails, the consumer leaves final-delivery false so the gateway's non-streaming final-send safety path can still run.
    +
    +**Verification:** Stream consumer tests reproduce the screenshot shape by simulating first chunk visible and continuation failure, then assert the final answer is not suppressed.
    +
    +### U3. Preserve Telegram topic/thread routing for overflow and fallback continuations
    +
    +**Goal:** Ensure overflow recovery messages land in the same Telegram forum topic or DM topic fallback context.
    +
    +**Requirements:** R3
    +
    +**Dependencies:** U1, U2
    +
    +**Files:**
    +- `gateway/platforms/telegram.py`
    +- `gateway/stream_consumer.py`
    +- `tests/gateway/test_stream_consumer_thread_routing.py`
    +- Relevant Telegram adapter routing tests, if existing coverage is closer there
    +
    +**Approach:**
    +- Keep passing `metadata` through every overflow continuation and fallback send.
    +- Keep reply anchors where valid, but do not let a missing reply anchor drop the `message_thread_id` for normal forum topics.
    +- For private DM topic fallback metadata, preserve the existing stricter anchor behaviour documented in the adapter comments.
    +
    +**Patterns to follow:**
    +- `TelegramAdapter._thread_kwargs_for_send(...)`.
    +- Existing tests around Telegram topic recovery and stream consumer thread routing.
    +
    +**Test scenarios:**
    +- Overflow continuations include `message_thread_id` for a forum topic.
    +- A continuation retry after `reply message not found` keeps forum topic routing when allowed.
    +- Partial-overflow fallback sends receive the same metadata passed to the original stream consumer.
    +
    +**Verification:** Thread-routing assertions inspect fake bot calls and confirm all continuation/fallback messages carry the expected topic metadata.
    +
    +### U4. Add issue evidence and PR body traceability
    +
    +**Goal:** Make the upstream issue and PR clearly trace the user-visible bug and verification evidence.
    +
    +**Requirements:** R5
    +
    +**Dependencies:** U1, U2, U3
    +
    +**Files:**
    +- GitHub issue body created via `gh issue create`
    +- PR body using `.github/PULL_REQUEST_TEMPLATE.md`
    +
    +**Approach:**
    +- Create a GitHub issue with the screenshot evidence: the long message in the `Nehemiah - Coding` Telegram topic stops at `- The visible tool-call summary`, and the user's reply says the previous message did not finish streaming to that Telegram topic.
    +- Reference affected component as Gateway and platform as Telegram.
    +- In the PR body, link the issue with `Fixes #...`, describe the split-delivery contract change, and include the screenshot or attach it if GitHub upload is available.
    +- Follow `CONTRIBUTING.md` and the repository PR template exactly.
    +
    +**Patterns to follow:**
    +- `.github/ISSUE_TEMPLATE/bug_report.yml`
    +- `.github/PULL_REQUEST_TEMPLATE.md`
    +
    +**Test scenarios:**
    +- Test expectation: none, this is tracker and PR documentation work.
    +
    +**Verification:** The GitHub issue exists with screenshot evidence or an explicit screenshot reference, and the PR body links the issue and lists the tests run.
    +
    +---
    +
    +## Scope Boundaries
    +
    +### In Scope
    +
    +- Telegram streamed response overflow splitting and recovery.
    +- Stream consumer final-delivery truth for partial overflow delivery.
    +- Topic/thread metadata preservation for overflow and fallback continuation sends.
    +- Focused unit tests around adapter and stream consumer behaviour.
    +
    +### Out of Scope
    +
    +- Changing model streaming semantics in `run_agent.py`.
    +- Reworking Telegram draft streaming, which is DM-only and not the forum-topic path in the screenshot.
    +- Changing general platform message splitting for Discord, Slack, WhatsApp, or Matrix unless a shared helper must be corrected for the Telegram fix.
    +- Altering tool-progress display settings or terminal progress rendering.
    +
    +### Deferred to Follow-Up Work
    +
    +- Broader observability for gateway delivery completeness across all messaging platforms.
    +- A user-facing resend/recover command for a previous truncated response.
    +
    +---
    +
    +## Risks & Mitigations
    +
    +- Risk: fallback recovery duplicates already-visible first chunks. Mitigation: use delivered-prefix metadata where reliable and add tests for no-duplicate complete-success behaviour.
    +- Risk: preserving forum topic routing while dropping invalid reply anchors is easy to regress. Mitigation: include fake bot call assertions for `message_thread_id` and reply behaviour.
    +- Risk: MarkdownV2 formatting can alter visible/raw prefix comparisons. Mitigation: keep fallback conservative; duplicate content is preferable to silently missing content, but tests should keep the common path tail-only.
    +
    +---
    +
    +## Sources & Research
    +
    +- User-provided screenshot at `/root/.hermes/image_cache/img_f664e68f6ddf.jpg`.
    +- `gateway/stream_consumer.py` streamed edit, overflow, fallback, and final-delivery state handling.
    +- `gateway/platforms/telegram.py` Telegram send/edit overflow splitting and topic routing helpers.
    +- `gateway/platforms/base.py` `SendResult` contract and shared message chunking helper.
    +- `tests/gateway/test_stream_consumer.py`, `tests/gateway/test_stream_consumer_thread_routing.py`, and Telegram adapter tests for focused regression coverage.
    +
    +---
    +
    +## Verification Strategy
    +
    +- Run focused Telegram adapter overflow tests.
    +- Run focused stream consumer overflow/fallback tests.
    +- Run topic-routing tests affected by metadata changes.
    +- Run the gateway test subset around Telegram send/edit, stream consumer, and run progress if touched.
    +- Before PR creation, ensure `git diff` contains only the plan, implementation, tests, and PR/issue-relevant documentation for this bug.

  • gateway/platforms/base.py

    @@ -1545,6 +1545,13 @@ class SendResult:
         message_id: Optional[str] = None
         error: Optional[str] = None
         raw_response: Any = None
    +    # Adapter-specific metadata.  Cross-layer contracts that affect delivery
    +    # semantics must be documented at the producer and consumer sites.  Current
    +    # known contract: Telegram edit overflow partials set
    +    # raw_response["partial_overflow"] with delivered_chunks, total_chunks,
    +    # last_message_id, delivered_prefix, and continuation_message_ids so the
    +    # stream consumer can send the missing tail instead of marking a clipped
    +    # response complete.
         retryable: bool = False  # True for transient connection errors — base will retry automatically
         # When the adapter had to split an oversized payload across multiple
         # platform messages (e.g. Telegram edit_message overflow split-and-deliver),

  • gateway/platforms/telegram.py

    @@ -2348,10 +2348,15 @@ async def _edit_overflow_split(
                         )
                     except Exception as fmt_err:
                         if "not modified" not in str(fmt_err).lower():
    +                        logger.warning(
    +                            "[%s] Overflow split: MarkdownV2 first-chunk edit "
    +                            "failed, falling back to plain text: %s",
    +                            self.name, fmt_err,
    +                        )
                             await self._bot.edit_message_text(
                                 chat_id=int(chat_id),
                                 message_id=int(message_id),
    -                            text=first_chunk,
    +                            text=_strip_mdv2(first_chunk),
                             )
                 else:
                     await self._bot.edit_message_text(
    @@ -2379,6 +2384,7 @@ async def _edit_overflow_split(
             # are already correctly sized).  Best-effort MarkdownV2 with plain
             # fallback, mirroring send().
             continuation_ids: list[str] = []
    +        delivered_chunks = [first_chunk]
             prev_id = message_id
             thread_id = self._metadata_thread_id(metadata)
             for chunk in chunks[1:]:
    @@ -2392,7 +2398,14 @@ async def _edit_overflow_split(
                 )
                 for use_markdown in (True, False) if finalize else (False,):
                     try:
    -                    text = self.format_message(chunk) if use_markdown else chunk
    +                    if use_markdown:
    +                        text = self.format_message(chunk)
    +                    else:
    +                        # Plain attempt: on finalize the MarkdownV2 attempt
    +                        # failed, so degrade to clean stripped text, never
    +                        # the raw chunk (raw ** / ``` markers would render
    +                        # literally); streaming previews stay raw.
    +                        text = _strip_mdv2(chunk) if finalize else chunk
                         sent_msg = await self._bot.send_message(
                             chat_id=int(chat_id),
                             text=text,
    @@ -2418,7 +2431,7 @@ async def _edit_overflow_split(
                             try:
                                 sent_msg = await self._bot.send_message(
                                     chat_id=int(chat_id),
    -                                text=chunk,
    +                                text=_strip_mdv2(chunk) if finalize else chunk,
                                     **retry_thread_kwargs,
                                     **self._link_preview_kwargs(),
                                     **self._notification_kwargs(metadata),
    @@ -2442,17 +2455,37 @@ async def _edit_overflow_split(
                         break
                 if sent_msg is None:
                     # Continuation failed — the user has chunk 1 + however many
    -                # continuations succeeded.  Report success with what we got
    -                # so the stream consumer knows the edit landed; the
    -                # remaining tail is lost on this attempt and the next
    -                # streaming tick may retry.
    +                # continuations succeeded, but NOT the full response.  Do not
    +                # report success: the stream consumer treats a successful edit
    +                # as final delivery on got_done, which would suppress fallback
    +                # delivery and leave the Telegram topic clipped after the last
    +                # delivered chunk.
                     logger.warning(
                         "[%s] Overflow split: stopped at %d/%d chunks delivered",
                         self.name, 1 + len(continuation_ids), len(chunks),
                     )
    -                break
    +                delivered_prefix = "".join(
    +                    re.sub(r" \(\d+/\d+\)$", "", delivered)
    +                    for delivered in delivered_chunks
    +                )
    +                return SendResult(
    +                    success=False,
    +                    message_id=prev_id,
    +                    error="overflow_continuation_failed",
    +                    retryable=True,
    +                    raw_response={
    +                        "partial_overflow": True,
    +                        "delivered_chunks": 1 + len(continuation_ids),
    +                        "total_chunks": len(chunks),
    +                        "last_message_id": prev_id,
    +                        "delivered_prefix": delivered_prefix,
    +                        "continuation_message_ids": tuple(continuation_ids),
    +                    },
    +                    continuation_message_ids=tuple(continuation_ids),
    +                )
                 new_id = str(getattr(sent_msg, "message_id", "")) or prev_id
                 continuation_ids.append(new_id)
    +            delivered_chunks.append(chunk)
                 prev_id = new_id
     
             last_id = continuation_ids[-1] if continuation_ids else message_id

  • gateway/stream_consumer.py

    @@ -147,8 +147,15 @@ def __init__(
             self._edit_supported = True  # Disabled when progressive edits are no longer usable
             self._last_edit_time = 0.0
             self._last_sent_text = ""   # Track last-sent text to skip redundant edits
    +        # True when the most recent _send_or_edit split-and-delivered across
    +        # continuation messages (the adapter adopted a new message id).
    +        self._last_edit_overflowed = False
             self._fallback_final_send = False
             self._fallback_prefix = ""
    +        # True when fallback is sending only the missing tail after a partial
    +        # Telegram overflow delivery.  In that case the already-visible prefix
    +        # is intentional content, not a stale preview to delete.
    +        self._fallback_preserve_partial_messages = False
             self._flood_strikes = 0         # Consecutive flood-control edit failures
             self._current_edit_interval = self.cfg.edit_interval  # Adaptive backoff
             self._final_response_sent = False
    @@ -261,6 +268,7 @@ def _reset_segment_state(self, *, preserve_no_edit: bool = False) -> None:
             self._last_sent_text = ""
             self._fallback_final_send = False
             self._fallback_prefix = ""
    +        self._fallback_preserve_partial_messages = False
             # #29346: a tool/segment boundary means what we delivered was an interim
             # preamble, not the final answer — clear the flags so a premature setter
             # can't fool the gateway. Safe: got_done returns before any reset, and
    @@ -581,14 +589,20 @@ async def run(self) -> None:
                         if self._accumulated:
                             if self._fallback_final_send:
                                 await self._send_fallback_final(self._accumulated)
    -                        elif (
    -                            current_update_visible
    -                            and not self._adapter_requires_finalize
    +                        elif current_update_visible and (
    +                            not self._adapter_requires_finalize
    +                            or self._last_edit_overflowed
                             ):
                                 # Mid-stream edit above already delivered the
                                 # final accumulated content.  Skip the redundant
    -                            # final edit — but only for adapters that don't
    -                            # need an explicit finalize signal.
    +                            # final edit for adapters that don't need an
    +                            # explicit finalize signal, and for any adapter
    +                            # when that edit split-and-delivered across
    +                            # continuations: the split edit carried
    +                            # finalize=True itself, and re-finalizing with
    +                            # the full text would overflow-split again into
    +                            # the adopted continuation, duplicating chunks
    +                            # on screen.
                                 self._final_response_sent = True
                                 self._final_content_delivered = True
                             elif self._message_id:
    @@ -647,11 +661,21 @@ async def run(self) -> None:
                     await asyncio.sleep(0.05)  # Small yield to not busy-loop
     
             except asyncio.CancelledError:
    -            # Best-effort final edit on cancellation
    +            # Best-effort final edit on cancellation.  finalize=True so
    +            # REQUIRES_EDIT_FINALIZE platforms (Telegram) apply final
    +            # formatting — a plain edit here would leave the entire reply
    +            # rendered as a raw streaming preview while the success flags
    +            # below suppress the gateway's formatted re-send.
    +            # is_turn_final=False keeps _try_fresh_final from setting
    +            # _final_response_sent itself; this handler owns the flags.
                 _best_effort_ok = False
                 if self._accumulated and self._message_id:
                     try:
    -                    _best_effort_ok = bool(await self._send_or_edit(self._accumulated))
    +                    _best_effort_ok = bool(
    +                        await self._send_or_edit(
    +                            self._accumulated, finalize=True, is_turn_final=False,
    +                        )
    +                    )
                     except Exception:
                         pass
                 # Only confirm final delivery if the best-effort send above
    @@ -867,11 +891,21 @@ async def _send_fallback_final(self, text: str) -> None:
                 self._notify_new_message()
     
             # Remove the frozen partial message so the user only sees the
    -        # complete fallback response.  Best-effort — if the platform doesn't
    +        # complete fallback response.  ONLY safe when the fallback re-sent
    +        # the FULL final text (continuation == final_text).  When the
    +        # prefix-based dedup above sent only the missing TAIL, the partial
    +        # message IS the head of the answer — deleting it leaves the user
    +        # with only the last part of the response (the "Gemini sent only
    +        # the second half" symptom).  Best-effort — if the platform doesn't
             # implement ``delete_message``, the delete fails (flood control still
             # active, bot lacks permission, message too old to delete), the
             # partial remains but at least the full answer was delivered.
    -        if stale_message_id and stale_message_id != last_message_id:
    +        if (
    +            stale_message_id
    +            and stale_message_id != last_message_id
    +            and not self._fallback_preserve_partial_messages
    +            and continuation == final_text
    +        ):
                 delete_fn = getattr(self.adapter, "delete_message", None)
                 if delete_fn is not None:
                     try:
    @@ -888,6 +922,7 @@ async def _send_fallback_final(self, text: str) -> None:
             self._final_content_delivered = True
             self._last_sent_text = chunks[-1]
             self._fallback_prefix = ""
    +        self._fallback_preserve_partial_messages = False
     
         def _is_flood_error(self, result) -> bool:
             """Check if a SendResult failure is due to flood control / rate limiting."""
    @@ -1208,6 +1243,7 @@ async def _send_or_edit(
                     return True
                 # Failure already disabled drafts for this run; fall through to
                 # the regular edit/send path below.
    +        self._last_edit_overflowed = False
             try:
                 if self._message_id is not None:
                     if self._edit_supported:
    @@ -1264,6 +1300,7 @@ async def _send_or_edit(
                                 and result.message_id
                                 and result.message_id != self._message_id
                             ):
    +                            self._last_edit_overflowed = True
                                 self._message_id = str(result.message_id)
                                 self._message_created_ts = time.monotonic()
                                 self._last_sent_text = ""
    @@ -1274,6 +1311,35 @@ async def _send_or_edit(
                             self._flood_strikes = 0
                             return True
                         else:
    +                        raw_response = getattr(result, "raw_response", None)
    +                        if isinstance(raw_response, dict) and raw_response.get("partial_overflow"):
    +                            # Telegram edited/sent one or more overflow chunks,
    +                            # but not the complete response.  Preserve the
    +                            # visible prefix so the got_done fallback sends the
    +                            # missing tail instead of marking a clipped topic
    +                            # reply as final delivery.
    +                            self._message_id = str(
    +                                raw_response.get("last_message_id")
    +                                or result.message_id
    +                                or self._message_id
    +                            )
    +                            delivered_prefix = raw_response.get("delivered_prefix")
    +                            if isinstance(delivered_prefix, str) and delivered_prefix:
    +                                self._last_sent_text = delivered_prefix
    +                                self._fallback_prefix = delivered_prefix
    +                                self._fallback_preserve_partial_messages = text.startswith(
    +                                    delivered_prefix
    +                                )
    +                            else:
    +                                self._fallback_prefix = self._visible_prefix()
    +                                self._fallback_preserve_partial_messages = False
    +                            self._fallback_final_send = True
    +                            self._edit_supported = False
    +                            self._already_sent = True
    +                            if getattr(result, "continuation_message_ids", ()):
    +                                self._notify_new_message()
    +                            return False
    +
                             # Edit failed.  If this looks like flood control / rate
                             # limiting, use adaptive backoff: double the edit interval
                             # and retry on the next cycle.  Only permanently disable

  • tests/gateway/test_stream_consumer.py

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

  • tests/gateway/test_stream_consumer_fresh_final.py

    @@ -347,6 +347,200 @@ async def test_multi_tool_call_turn_delivers_final_once(self):
             assert any("answer is 42" in t for t in self._delivered_texts(adapter))
     
     
    +class TestCancelledBestEffortDeliveryFinalizes:
    +    """Cancel-path best-effort delivery must go through the finalize path.
    +
    +    The gateway cancels the consumer shortly after finish(). The
    +    CancelledError handler re-delivers the accumulated text; previously it
    +    did so with finalize=False, so REQUIRES_EDIT_FINALIZE platforms
    +    (Telegram) kept the plain streaming preview — the whole final reply
    +    rendered with raw markdown markers — while the success flags still
    +    suppressed the gateway's formatted re-send.
    +    """
    +
    +    @pytest.mark.asyncio
    +    async def test_cancel_best_effort_edit_is_finalized(self):
    +        adapter = _make_adapter()
    +        adapter.REQUIRES_EDIT_FINALIZE = True
    +        consumer = GatewayStreamConsumer(
    +            adapter=adapter,
    +            chat_id="chat",
    +            config=StreamConsumerConfig(
    +                edit_interval=0.01, buffer_threshold=5, cursor=" ▉",
    +            ),
    +        )
    +        consumer.on_delta("Reply with **bold** and `code` markers.")
    +        task = asyncio.create_task(consumer.run())
    +        await asyncio.sleep(0.05)  # preview lands; message_id set
    +        task.cancel()
    +        await asyncio.gather(task, return_exceptions=True)
    +
    +        finalize_edits = [
    +            c for c in adapter.edit_message.call_args_list
    +            if c.kwargs.get("finalize")
    +        ]
    +        assert finalize_edits, (
    +            "cancel best-effort delivery must use finalize=True so "
    +            "REQUIRES_EDIT_FINALIZE platforms apply final formatting"
    +        )
    +        assert consumer.final_response_sent is True
    +        assert consumer.final_content_delivered is True
    +
    +    @pytest.mark.asyncio
    +    async def test_cancel_best_effort_failure_keeps_gateway_resend_possible(self):
    +        adapter = _make_adapter()
    +        adapter.REQUIRES_EDIT_FINALIZE = True
    +        consumer = GatewayStreamConsumer(
    +            adapter=adapter,
    +            chat_id="chat",
    +            config=StreamConsumerConfig(
    +                edit_interval=0.01, buffer_threshold=5, cursor=" ▉",
    +            ),
    +        )
    +        consumer.on_delta("Reply with **bold** and `code` markers.")
    +        task = asyncio.create_task(consumer.run())
    +        await asyncio.sleep(0.05)
    +        # Best-effort delivery at cancel time fails.
    +        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(
    +            success=False, error="boom",
    +        ))
    +        task.cancel()
    +        await asyncio.gather(task, return_exceptions=True)
    +
    +        assert consumer.final_response_sent is False
    +        assert consumer.final_content_delivered is False
    +
    +    @pytest.mark.asyncio
    +    async def test_cancel_without_preview_makes_no_delivery_attempt(self):
    +        adapter = _make_adapter()
    +        adapter.REQUIRES_EDIT_FINALIZE = True
    +        consumer = GatewayStreamConsumer(
    +            adapter=adapter,
    +            chat_id="chat",
    +            config=StreamConsumerConfig(
    +                edit_interval=0.01, buffer_threshold=5, cursor=" ▉",
    +            ),
    +        )
    +        task = asyncio.create_task(consumer.run())
    +        await asyncio.sleep(0.02)
    +        task.cancel()
    +        await asyncio.gather(task, return_exceptions=True)
    +
    +        adapter.edit_message.assert_not_called()
    +        assert consumer.final_response_sent is False
    +        assert consumer.final_content_delivered is False
    +
    +    @pytest.mark.asyncio
    +    async def test_cancel_with_fresh_final_enabled_delivers_and_flags_via_handler(self):
    +        """With fresh_final_after_seconds enabled and an aged preview, the
    +        finalized cancel-path delivery is eligible for fresh-final
    +        (delete + fresh send). is_turn_final=False keeps _try_fresh_final
    +        from setting the flags itself; the cancel handler sets them after
    +        the successful delivery."""
    +        adapter = _make_adapter()
    +        adapter.REQUIRES_EDIT_FINALIZE = True
    +        adapter.send.side_effect = [
    +            SimpleNamespace(success=True, message_id="initial_preview"),
    +            SimpleNamespace(success=True, message_id="fresh_final"),
    +        ]
    +        consumer = GatewayStreamConsumer(
    +            adapter=adapter,
    +            chat_id="chat",
    +            config=StreamConsumerConfig(
    +                edit_interval=0.01, buffer_threshold=5, cursor=" ▉",
    +                fresh_final_after_seconds=0.001,
    +            ),
    +        )
    +        consumer.on_delta("Reply with **bold** and `code` markers.")
    +        task = asyncio.create_task(consumer.run())
    +        await asyncio.sleep(0.05)
    +        consumer._message_created_ts = 0.0  # force the preview stale
    +        task.cancel()
    +        await asyncio.gather(task, return_exceptions=True)
    +
    +        # Fresh-final engaged: a second send replaced the stale preview.
    +        assert adapter.send.call_count == 2
    +        adapter.delete_message.assert_awaited_once_with("chat", "initial_preview")
    +        # Flags were set by the cancel handler after successful delivery.
    +        assert consumer.final_response_sent is True
    +        assert consumer.final_content_delivered is True
    +
    +
    +class TestGotDoneOverflowSplitNotRefinalized:
    +    """A got_done finalize edit that split-and-delivered across continuation
    +    messages must not be followed by the redundant requires-finalize edit.
    +
    +    After a split, the consumer adopts the last continuation as the live
    +    message and the redundant finalize edit re-submits the FULL accumulated
    +    text against it; the adapter pre-flights that into another overflow
    +    split, editing chunk 1 over the continuation and re-sending the rest,
    +    so the user sees duplicated chunks. The finalize signal was already
    +    carried by the split edit itself.
    +    """
    +
    +    def _consumer(self, adapter):
    +        # High interval/threshold so the only edit is the got_done finalize.
    +        return GatewayStreamConsumer(
    +            adapter=adapter,
    +            chat_id="chat",
    +            config=StreamConsumerConfig(
    +                edit_interval=10.0, buffer_threshold=10_000, cursor=" ▉",
    +            ),
    +        )
    +
    +    @pytest.mark.asyncio
    +    async def test_split_finalize_edit_is_not_refinalized(self):
    +        adapter = _make_adapter()
    +        adapter.REQUIRES_EDIT_FINALIZE = True
    +        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(
    +            success=True,
    +            message_id="cont_2",
    +            continuation_message_ids=("cont_2",),
    +        ))
    +        consumer = self._consumer(adapter)
    +        consumer.on_delta("oversize **markdown** final reply")
    +        task = asyncio.create_task(consumer.run())
    +        await asyncio.sleep(0.05)  # preview send lands; no interval edits
    +        consumer.finish()
    +        await task
    +
    +        finalize_edits = [
    +            c for c in adapter.edit_message.call_args_list
    +            if c.kwargs.get("finalize")
    +        ]
    +        assert len(finalize_edits) == 1, (
    +            "split finalize edit must not be re-finalized; the redundant "
    +            "edit re-splits the full text into the adopted continuation "
    +            "and duplicates chunks on screen"
    +        )
    +        assert consumer.final_response_sent is True
    +        assert consumer.final_content_delivered is True
    +
    +    @pytest.mark.asyncio
    +    async def test_non_split_finalize_edit_still_gets_explicit_refinalize(self):
    +        """The narrow fix must not regress the requires-finalize contract:
    +        a normal (non-split) got_done edit is still followed by the
    +        explicit finalize edit (#25010 semantics unchanged)."""
    +        adapter = _make_adapter()
    +        adapter.REQUIRES_EDIT_FINALIZE = True
    +        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(
    +            success=True, message_id="initial_preview",
    +        ))
    +        consumer = self._consumer(adapter)
    +        consumer.on_delta("short final reply")
    +        task = asyncio.create_task(consumer.run())
    +        await asyncio.sleep(0.05)
    +        consumer.finish()
    +        await task
    +
    +        finalize_edits = [
    +            c for c in adapter.edit_message.call_args_list
    +            if c.kwargs.get("finalize")
    +        ]
    +        assert len(finalize_edits) == 2
    +        assert consumer.final_response_sent is True
    +
    +
     class TestStreamConsumerConfigFreshFinalField:
         """The dataclass field must exist and default to 0 (disabled)."""
     

  • tests/gateway/test_telegram_overflow_partial.py

    @@ -0,0 +1,140 @@
    +"""Regression coverage for partial Telegram overflow delivery."""
    +
    +from types import SimpleNamespace
    +from unittest.mock import AsyncMock, MagicMock
    +
    +import pytest
    +
    +from gateway.config import PlatformConfig
    +from gateway.platforms.base import SendResult
    +from gateway.platforms.telegram import TelegramAdapter
    +from gateway.stream_consumer import GatewayStreamConsumer
    +
    +
    +def _message(message_id: int | str) -> SimpleNamespace:
    +    return SimpleNamespace(message_id=message_id)
    +
    +
    [redacted:email]
    +def telegram_adapter() -> TelegramAdapter:
    +    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
    +    adapter._bot = MagicMock()
    +    object.__setattr__(adapter, "MAX_MESSAGE_LENGTH", 160)
    +    return adapter
    +
    +
    [redacted:email]
    +async def test_edit_overflow_split_reports_success_when_all_continuations_land(telegram_adapter):
    +    """Complete overflow delivery keeps the existing successful contract."""
    +    content = "word " * 120
    +    telegram_adapter._bot.edit_message_text = AsyncMock(return_value=True)
    +    telegram_adapter._bot.send_message = AsyncMock(
    +        side_effect=[_message(202), _message(203), _message(204), _message(205)]
    +    )
    +
    +    result = await telegram_adapter._edit_overflow_split(
    +        "12345", "201", content, finalize=False, metadata={"thread_id": "77"}
    +    )
    +
    +    assert result.success is True
    +    assert result.message_id == result.continuation_message_ids[-1]
    +    assert result.raw_response is None
    +    assert telegram_adapter._bot.edit_message_text.await_count == 1
    +    assert telegram_adapter._bot.send_message.await_count == len(result.continuation_message_ids)
    +    for call in telegram_adapter._bot.send_message.await_args_list:
    +        assert call.kwargs["message_thread_id"] == 77
    +
    +
    [redacted:email]
    +async def test_edit_overflow_split_reports_later_partial_failure_after_some_continuations_land(telegram_adapter):
    +    """Partial metadata tracks the last delivered continuation before failure."""
    +    content = "word " * 120
    +    telegram_adapter._bot.edit_message_text = AsyncMock(return_value=True)
    +    telegram_adapter._bot.send_message = AsyncMock(
    +        side_effect=[
    +            _message(202),
    +            RuntimeError("telegram send failed"),
    +            RuntimeError("telegram send failed"),
    +        ]
    +    )
    +
    +    result = await telegram_adapter._edit_overflow_split(
    +        "12345", "201", content, finalize=False, metadata={"thread_id": "77"}
    +    )
    +
    +    assert result.success is False
    +    assert result.message_id == "202"
    +    assert result.raw_response["partial_overflow"] is True
    +    assert result.raw_response["delivered_chunks"] == 2
    +    assert result.raw_response["last_message_id"] == "202"
    +    assert result.continuation_message_ids == ("202",)
    +
    +
    [redacted:email]
    +async def test_edit_overflow_split_reports_partial_failure_when_continuation_fails(telegram_adapter):
    +    """A failed continuation must not be reported as final delivery."""
    +    content = "word " * 120
    +    telegram_adapter._bot.edit_message_text = AsyncMock(return_value=True)
    +    telegram_adapter._bot.send_message = AsyncMock(
    +        side_effect=[RuntimeError("telegram send failed"), RuntimeError("telegram send failed")]
    +    )
    +
    +    result = await telegram_adapter._edit_overflow_split(
    +        "12345", "201", content, finalize=False, metadata={"thread_id": "77"}
    +    )
    +
    +    assert result.success is False
    +    assert result.retryable is True
    +    assert result.error == "overflow_continuation_failed"
    +    assert result.message_id == "201"
    +    assert result.raw_response["partial_overflow"] is True
    +    assert result.raw_response["delivered_chunks"] == 1
    +    assert result.raw_response["total_chunks"] > 1
    +    assert result.raw_response["last_message_id"] == "201"
    +    assert result.raw_response["delivered_prefix"]
    +    assert result.continuation_message_ids == ()
    +
    +
    [redacted:email]
    +async def test_stream_consumer_fallback_sends_tail_after_partial_overflow():
    +    """A partial overflow edit enters fallback instead of marking final delivered."""
    +    adapter = MagicMock()
    +    adapter.MAX_MESSAGE_LENGTH = 4096
    +    adapter.edit_message = AsyncMock(
    +        return_value=SendResult(
    +            success=False,
    +            message_id="preview-1",
    +            error="overflow_continuation_failed",
    +            retryable=True,
    +            raw_response={
    +                "partial_overflow": True,
    +                "delivered_chunks": 1,
    +                "total_chunks": 2,
    +                "last_message_id": "preview-1",
    +                "delivered_prefix": "hello ",
    +            },
    +        )
    +    )
    +    adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="tail-1"))
    +    adapter.delete_message = AsyncMock(return_value=True)
    +
    +    consumer = GatewayStreamConsumer(adapter, "chat-1", metadata={"thread_id": "77"})
    +    consumer._message_id = "preview-1"
    +    consumer._last_sent_text = "hello "
    +
    +    ok = await consumer._send_or_edit("hello world", finalize=True)
    +
    +    assert ok is False
    +    assert consumer.final_response_sent is False
    +    assert consumer.final_content_delivered is False
    +    assert consumer._fallback_final_send is True
    +    assert consumer._fallback_prefix == "hello "
    +
    +    await consumer._send_fallback_final("hello world")
    +
    +    adapter.send.assert_awaited_once()
    +    assert adapter.send.await_args.kwargs["content"] == "world"
    +    assert adapter.send.await_args.kwargs["metadata"] == {"thread_id": "77"}
    +    adapter.delete_message.assert_not_awaited()
    +    assert consumer.final_response_sent is True
    +    assert consumer.final_content_delivered is True