Skip to content

fix(gateway): finish the split-delivery bug class behind the swallowed-finals fix (salvages #78558) #79669

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kshitijk4poor merged 3 commits into from
Aug 5, 2026
+370 −43
Merged

fix(gateway): finish the split-delivery bug class behind the swallowed-finals fix (salvages #78558) #79669

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3d3b395
fix(gateway): stop payload-less split delivery from swallowing finals
HexLab98 Aug 4, 2026
dafd478
test(gateway): cover payload-less split-delivery final-send swallow
HexLab98 Aug 4, 2026
5d65529
fix(gateway): finish the split-delivery bug class so the fix cannot d…
kshitijk4poor Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 19 additions & 6 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24828,9 +24828,10 @@ def _stream_confirmed_final_delivery(
# stream completion never reached any API call (#71643).
# Reconcile the recorded turn-final payload against the
# completed response; only a demonstrable mismatch (False)
# overrides the flag — None (no record / multi-message split
# delivery) keeps the legacy trust so overflow splits are not
# re-sent.
# overrides the flag — including payload-less multi-message
# split delivery (#78541). None (no record on a non-split
# legacy path) keeps the legacy trust so ambiguous-timeout
# dedup is not regressed.
matcher = getattr(consumer, "delivered_final_matches", None)
if callable(matcher):
try:
Expand Down Expand Up @@ -25621,8 +25622,9 @@ def _run_sync_with_timeout_lifecycle():
# Reconcile the consumer's recorded turn-final payload against the
# completed response: on a demonstrable mismatch (False) neither
# final_response_sent nor final_content_delivered may suppress the
# normal final send. None (no record / multi-message split
# delivery) keeps legacy trust; the failed-finalize family
# normal final send. False also covers payload-less multi-message
# split delivery (#78541). None (no record on a non-split legacy
# path) keeps legacy trust; the failed-finalize family
# (#51828 / #33793) is unaffected because those paths leave the
# flags False or record the complete fallback payload.
_stale_finalized = False
Expand Down Expand Up @@ -25666,9 +25668,20 @@ def _run_sync_with_timeout_lifecycle():
# user gets one corrected message; on edit failure fall through
# with already_sent unset so the normal final send delivers the
# complete text.
#
# Not valid for a multi-message split delivery: there
# ``message_id`` is only the LAST chunk, so editing it with the
# complete response would repeat every sealed head chunk's text
# inside the tail message. Fall through to the normal final send
# instead (#78541).
_sc_msg_id = _sc.message_id
_sc_adapter = getattr(_sc, "adapter", None)
if _sc_msg_id and _sc_msg_id != "__no_edit__" and _sc_adapter is not None:
if getattr(_sc, "_turn_split_delivery", False):
logger.info(
"Stale streamed finalize detected for session %s on a multi-message split; skipping the in-place reconciliation edit and delivering the complete response via normal final send (#78541).",
session_key or "?",
)
elif _sc_msg_id and _sc_msg_id != "__no_edit__" and _sc_adapter is not None:
try:
_reconcile_res = await _sc_adapter.edit_message(
chat_id=source.chat_id,
Expand Down
125 changes: 90 additions & 35 deletions gateway/stream_consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,10 @@ def __init__(
self._initial_reply_to_id = initial_reply_to_id
self._queue: queue.Queue = queue.Queue()
self._accumulated = ""
# Full segment text mirror of ``_accumulated`` that is NOT truncated
# when overflow splits seal head chunks. Used to record a reconciliable
# turn-final payload for multi-message deliveries (#78541).
self._stream_ledger = ""
self._message_id: Optional[str] = None
# Wall-clock timestamp (time.monotonic) when ``_message_id`` was
# first assigned from a successful first-send. Used by the
Expand Down Expand Up @@ -270,9 +274,11 @@ def __init__(
self._delivered_final_text: Optional[str] = None
# True when the current turn's answer was delivered across multiple
# sealed messages (overflow split / adapter continuation adoption).
# Payload-equality against a single recorded string is meaningless in
# that shape, so delivered_final_matches() falls back to legacy trust
# rather than risking a duplicate re-send of a multi-message reply.
# When a payload was recorded (via ``_stream_ledger`` /
# ``_record_turn_final_payload``), ``delivered_final_matches`` can still
# reconcile. Payload-less split delivery must NOT inherit legacy trust
# (#78541) — that combination was swallowing complete Telegram group
# replies after an early/partial multi-message delivery.
self._turn_split_delivery = False
self._delivered_commentary_texts: list[str] = []
# Retains the finalized visible text of each streaming segment so
Expand Down Expand Up @@ -405,20 +411,33 @@ async def _edit_message(
pass
return await self.adapter.edit_message(**kwargs)

def _append_accumulated(self, text: str) -> None:
"""Append to the live buffer and the split-stable stream ledger."""
if not text:
return
self._accumulated += text
self._stream_ledger += text

def _record_turn_final_payload(self, text: str) -> None:
"""Record the exact cleaned payload of a turn-final delivery.
"""Record what the user has actually seen as this turn's final answer.

Normalized the same way ``_send_or_edit`` normalizes outgoing text
(media-directive strip + fence closing) so the gateway can compare it
against the completed ``final_response`` (#71643). No-op when the turn
was delivered across multiple sealed messages — payload equality is
undefined there and ``delivered_final_matches`` returns ``None``.
against the completed ``final_response`` (#71643).

``text`` is what the *calling* path just delivered. On a multi-message
split that is only the trailing chunk — the overflow paths truncate
``_accumulated`` once head chunks are sealed — so ``_stream_ledger``
(the un-truncated segment text) is preferred there and ``text`` is
ignored. Without that substitution a split turn records a tail-only
payload, which the gateway reads as a mismatch and re-sends on top of
an answer the user already received (#78541).
"""
if self._turn_split_delivery:
self._delivered_final_text = None
return
source = text or ""
if self._turn_split_delivery and self._stream_ledger:
source = self._stream_ledger
self._delivered_final_text = ensure_closed_code_fences(
self._clean_for_display(text or "")
self._clean_for_display(source)
).strip()

def delivered_final_matches(self, final_text: str) -> Optional[bool]:
Expand All @@ -432,22 +451,24 @@ def delivered_final_matches(self, final_text: str) -> Optional[bool]:
delivered segment/commentary) matches ``final_text``; suppressing
the normal final send is safe.
- ``False`` — a turn-final delivery was recorded but its payload
demonstrably differs from ``final_text``; the user has NOT seen the
complete response and the normal final send must run.
- ``None`` — no payload comparison is possible (multi-message split
delivery, or a legacy/uncertain path that recorded nothing). The
caller keeps the pre-existing flag-trusting behavior so overflow
splits and ambiguous-timeout dedup are not regressed.
demonstrably differs from ``final_text``, OR this was a
payload-less multi-message split delivery (#78541) whose flag
alone must not suppress the normal final send.
- ``None`` — no payload comparison is possible on a non-split
legacy/uncertain path that recorded nothing. The caller keeps
the pre-existing flag-trusting behavior so ambiguous-timeout
dedup is not regressed.
"""
if self._turn_split_delivery:
return None
if self._delivered_final_text is None:
return None
target = ensure_closed_code_fences(
self._clean_for_display(final_text or "")
).strip()
if not target:
return None
if self._delivered_final_text is None:
if self._turn_split_delivery:
# #78541: refuse legacy trust for payload-less split delivery.
return False
return None
if self._delivered_final_text.strip() == target:
return True
# A segment break / commentary may have delivered the final text
Expand Down Expand Up @@ -541,6 +562,7 @@ def _reset_segment_state(self, *, preserve_no_edit: bool = False) -> None:
self._message_id = None
self._message_created_ts = None
self._accumulated = ""
self._stream_ledger = ""
self._last_sent_text = ""
self._fallback_final_send = False
self._fallback_prefix = ""
Expand Down Expand Up @@ -666,7 +688,7 @@ def _filter_and_accumulate(self, text: str) -> None:

if best_len:
# Emit text before the tag, enter think block
self._accumulated += buf[:best_idx]
self._append_accumulated(buf[:best_idx])
self._in_think_block = True
buf = buf[best_idx + best_len:]
else:
Expand All @@ -678,7 +700,7 @@ def _filter_and_accumulate(self, text: str) -> None:
if lower_buf.endswith(tag_lower[:i]) and i > held_back:
held_back = i
if held_back:
self._accumulated += buf[:-held_back]
self._append_accumulated(buf[:-held_back])
self._think_buffer = buf[-held_back:]
else:
# No (partial) open tag — but the model may have
Expand All @@ -687,7 +709,7 @@ def _filter_and_accumulate(self, text: str) -> None:
# matched open, or when upstream stripping is
# incomplete). Strip those before accumulating so
# they never reach the user.
self._accumulated += self._strip_orphan_close_tags(buf)
self._append_accumulated(self._strip_orphan_close_tags(buf))
return

@classmethod
Expand Down Expand Up @@ -732,7 +754,7 @@ def _flush_think_buffer(self) -> None:
if self._think_buffer and not self._in_think_block:
# Strip any orphan close tags that may have been held back —
# see _filter_and_accumulate for context.
self._accumulated += self._strip_orphan_close_tags(self._think_buffer)
self._append_accumulated(self._strip_orphan_close_tags(self._think_buffer))
self._think_buffer = ""

async def run(self) -> None:
Expand Down Expand Up @@ -926,6 +948,16 @@ async def run(self) -> None:
self._message_created_ts = None
self._last_sent_text = ""

if chunks_delivered:
# A sealed head is on screen, so this turn is now a
# multi-message delivery. Flag it BEFORE the tail
# send below: the fresh-final route replaces every
# tracked preview with one message, which is only
# valid while the active message holds the whole
# answer. Once heads are sealed it does not, and
# deleting them would drop delivered text (#78541).
self._turn_split_delivery = True

self._last_edit_time = time.monotonic()
if got_done:
tail_delivered = True
Expand All @@ -939,11 +971,12 @@ async def run(self) -> None:
self._final_response_sent = chunks_delivered and tail_delivered
if self._final_response_sent:
self._final_content_delivered = True
# Multi-message split delivery — payload
# equality against a single record is
# undefined (#71643).
# Multi-message split delivery — record the
# unsplit ledger payload so the gateway can
# still reconcile against final_response
# (#71643, #78541).
self._turn_split_delivery = True
self._delivered_final_text = None
self._record_turn_final_payload(self._accumulated)
return
if got_segment_break:
self._message_id = None
Expand Down Expand Up @@ -1400,7 +1433,11 @@ async def _send_fallback_final(self, text: str) -> None:
self._final_response_sent = True
self._final_content_delivered = True
# The visible partial equals the complete final text (#71643).
self._delivered_final_text = final_text.strip()
# Route through the recorder so a split turn records the full
# ledger rather than this tail-only payload — an unrecorded or
# tail-only split now reads as a mismatch and would re-send
# text the user already has (#78541).
self._record_turn_final_payload(final_text)
return

raw_limit = getattr(self.adapter, "MAX_MESSAGE_LENGTH", 4096)
Expand Down Expand Up @@ -1506,7 +1543,10 @@ async def _send_fallback_final(self, text: str) -> None:
# The fallback delivered the complete ``final_text`` (as one message
# or prefix + continuation chunks that union to it), so record it as
# the turn-final payload for the gateway's reconciliation (#71643).
self._delivered_final_text = final_text.strip()
# On a split turn ``final_text`` is only the tail — the recorder
# substitutes the unsplit ledger so the sealed heads count as
# delivered too (#78541).
self._record_turn_final_payload(final_text)
self._last_sent_text = chunks[-1]
self._fallback_prefix = ""
self._fallback_preserve_partial_messages = False
Expand Down Expand Up @@ -1578,7 +1618,9 @@ async def _send_empty_fallback_final(self, final_text: str) -> str:
self._final_response_sent = True
self._final_content_delivered = True
# Fresh commit of the complete answer after a failed finalize (#71643).
self._delivered_final_text = final_text.strip()
# Recorder-routed so a split turn records the unsplit ledger instead of
# a tail-only payload the gateway would read as stale (#78541).
self._record_turn_final_payload(final_text)
self._last_sent_text = final_text
self._fallback_prefix = ""
self._fallback_preserve_partial_messages = False
Expand Down Expand Up @@ -1901,6 +1943,15 @@ async def _try_fresh_final(self, text: str, *, is_turn_final: bool = True) -> bo
# current one plus any continuation fragments tracked while streaming
# (an oversized reply split across the platform's edit limit). All of
# them are replaced by the single fresh message below.
#
# That replacement is only sound while ``text`` holds the whole answer.
# On a multi-message split the head chunks were sealed and dropped out
# of ``_accumulated``, so ``text`` is just the tail — deleting the
# sealed heads would erase text the user already received and leave the
# complete reply nowhere on screen (#78541). Keep the sealed messages
# and take the normal edit path instead.
if self._turn_split_delivery:
return False
stale_ids = set(self._preview_message_ids)
if self._message_id and self._message_id != "__no_edit__":
stale_ids.add(self._message_id)
Expand Down Expand Up @@ -1987,6 +2038,7 @@ async def _suppress_silence_marker(self) -> None:
self._preview_message_ids = set()
self._message_id = None
self._accumulated = ""
self._stream_ledger = ""
self._last_sent_text = ""
self._already_sent = False
self._final_response_sent = False
Expand Down Expand Up @@ -2203,9 +2255,12 @@ async def _send_or_edit(
self._final_content_delivered = True
# ``text`` is already cleaned/fence-closed here and
# equals the visible prefix — the on-screen content
# IS this finalize payload (#71643).
if not self._turn_split_delivery:
self._delivered_final_text = text.strip()
# IS this finalize payload (#71643). Record it on
# split turns too: post-#78541 an unrecorded split
# reads as a mismatch and would re-send this
# already-visible answer, reintroducing the
# duplicate #45517 fixed (#36965 / #25349).
self._record_turn_final_payload(text)
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,
Expand Down
Loading
Loading