78b8032301
_continuation_text backed the cut up to the previous space even when the visible
preview already ended on a whole word, so the continuation re-sent that word at
the seam ("the answer" + "answer to your question."). Back up only when the cut
lands inside a word (#116312 behaviour kept).
Found by the platforms E2E suite (#121409).
416 lines
22 KiB
Python
416 lines
22 KiB
Python
"""Fallback delivery for GatewayStreamConsumer: continuation sends after edits
|
|
stop working, chunking, cursor cleanup, commentary and silence retraction."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
from typing import Any, Callable, Optional
|
|
|
|
from gateway.platforms.base import BasePlatformAdapter as _BasePlatformAdapter
|
|
from gateway.stream_consumer_fences import ensure_closed_code_fences
|
|
|
|
logger = logging.getLogger("gateway.stream_consumer")
|
|
|
|
|
|
class StreamFallbackMixin:
|
|
"""Non-streaming delivery paths used once progressive edits fail or the turn ends oddly."""
|
|
|
|
async def _send_new_chunk(self, text: str, reply_to_id: Optional[str], *,
|
|
final: bool = False) -> Optional[str]:
|
|
"""Send a new chunk threaded to ``reply_to_id``; returns the new message_id."""
|
|
text = self._clean_for_display(text)
|
|
if not text.strip():
|
|
return reply_to_id
|
|
try:
|
|
result = await self.adapter.send(
|
|
chat_id=self.chat_id, content=text, reply_to=reply_to_id,
|
|
metadata=self._metadata_for_send(final=final, expect_edits=not final))
|
|
if not (result.success and result.message_id):
|
|
self._edit_supported = False
|
|
return reply_to_id
|
|
self._message_id = str(result.message_id)
|
|
self._track_preview_ids_from_result(result)
|
|
self._already_sent = True
|
|
self._last_sent_text = text
|
|
self._notify_new_message()
|
|
return str(result.message_id)
|
|
except Exception as e:
|
|
logger.error("Stream send chunk error: %s", e)
|
|
return reply_to_id
|
|
|
|
def _visible_prefix(self) -> str:
|
|
"""Return the visible text already shown in the streamed message."""
|
|
prefix = self._last_sent_text or ""
|
|
if self.cfg.cursor and prefix.endswith(self.cfg.cursor):
|
|
prefix = prefix[:-len(self.cfg.cursor)]
|
|
return self._clean_for_display(prefix)
|
|
|
|
def _continuation_text(self, final_text: str) -> str:
|
|
"""Return only the part of final_text the user has not already seen."""
|
|
prefix = self._fallback_prefix or self._visible_prefix()
|
|
if prefix and final_text.startswith(prefix):
|
|
cut = len(prefix)
|
|
# ``prefix`` is whatever the last successful edit put on screen. Edits
|
|
# fire on a throttle tick, not at a word boundary, so that prefix can
|
|
# end inside a word. Back the cut up to the last space or newline so
|
|
# the continuation re-sends the broken word's tail and reads as an
|
|
# ordinary continuation. A prefix with no boundary (one very long
|
|
# token) keeps the original cut rather than re-sending the whole reply.
|
|
# A prefix that already ends on a whole word needs no back-up: doing it
|
|
# re-sent that word at the seam.
|
|
if cut < len(final_text) and not final_text[cut].isspace() and not final_text[cut - 1].isspace():
|
|
boundary = max(
|
|
final_text.rfind(" ", 0, cut),
|
|
final_text.rfind("\n", 0, cut),
|
|
)
|
|
if boundary >= 0:
|
|
cut = boundary + 1
|
|
return final_text[cut:].lstrip()
|
|
return final_text
|
|
|
|
@staticmethod
|
|
def _split_text_chunks(text: str, limit: int, len_fn: "Callable[[str], int]" = len,
|
|
) -> list[str]:
|
|
"""Split text for fallback sends: newline-preferred, fence-balanced across chunks."""
|
|
from gateway.platforms.helpers import split_text_fence_aware
|
|
return split_text_fence_aware(text, limit, len_fn, prefer_paragraphs=False,
|
|
balance_fences=True)
|
|
|
|
def _truncate_for_stream(self, text: str, limit: int, len_fn: "Callable[[str], int]",
|
|
) -> list[str]:
|
|
"""Split via the adapter's canonical truncate_message (platform-specific rules);
|
|
non-base test doubles / legacy adapters keep the 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(c, str) for c 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 (one flood retry
|
|
per chunk)."""
|
|
if getattr(self, "_egress_declined", False):
|
|
# The connector refused this destination earlier in the run. The
|
|
# whole point of this path is to deliver the unseen tail as a NEW
|
|
# message, which is exactly the re-addressing the egress guard
|
|
# exists to stop.
|
|
logger.warning(
|
|
"suppressing the fallback continuation: the connector already "
|
|
"declined this destination for this run"
|
|
)
|
|
return
|
|
# Balance fences BEFORE computing the continuation so the closing fence
|
|
# reaches the user even when only the tail is delivered.
|
|
final_text = ensure_closed_code_fences(self._clean_for_display(text))
|
|
continuation = self._continuation_text(final_text)
|
|
self._fallback_final_send = False
|
|
if not continuation.strip():
|
|
continuation = await self._fallback_when_nothing_unseen(final_text)
|
|
if continuation is None:
|
|
return
|
|
|
|
_len_fn, raw_limit = self._fallback_len_budget()
|
|
chunks = self._split_text_chunks(continuation, max(500, raw_limit - 100), len_fn=_len_fn)
|
|
|
|
stale_message_id = self._message_id # partial message to clean up
|
|
last_message_id: Optional[str] = None
|
|
last_successful_chunk = ""
|
|
sent_any_chunk = False
|
|
# Thread only a FULL resend (it replaces the preview); a tail continuation
|
|
# keeps its existing unthreaded delivery on every platform.
|
|
anchor = self._initial_reply_to_id if continuation == final_text else None
|
|
for chunk in chunks:
|
|
result = await self._send_with_flood_retry(
|
|
content=chunk, reply_to=None if sent_any_chunk else anchor,
|
|
retry_log="Flood control on fallback send, retrying in %.1fs")
|
|
if not result or not result.success:
|
|
# Partial continuation landed: do NOT set _final_response_sent (the
|
|
# gateway must still deliver the full answer); _already_sent only
|
|
# prevents a duplicate of the partial. Nothing landed: let the
|
|
# gateway final send try once more.
|
|
self._already_sent = sent_any_chunk
|
|
self._message_id = last_message_id
|
|
self._last_sent_text = last_successful_chunk
|
|
self._fallback_prefix = ""
|
|
return
|
|
sent_any_chunk = True
|
|
last_successful_chunk = chunk
|
|
last_message_id = result.message_id or last_message_id
|
|
self._notify_new_message()
|
|
|
|
# Best-effort delete of the frozen partial — ONLY when the FULL final was
|
|
# re-sent. If only the missing tail went out, the partial IS the head of
|
|
# the answer ("sent only the second half" symptom).
|
|
if (stale_message_id and stale_message_id != last_message_id
|
|
and not self._fallback_preserve_partial_messages and continuation == final_text):
|
|
await self._delete_previews([stale_message_id], label="Fallback partial",
|
|
skip_sentinel=False)
|
|
|
|
self._message_id = last_message_id
|
|
self._already_sent = True
|
|
# Recorder substitutes the unsplit ledger on a split turn.
|
|
self._mark_final_delivered(record=final_text)
|
|
self._last_sent_text = chunks[-1]
|
|
self._fallback_prefix = ""
|
|
self._fallback_preserve_partial_messages = False
|
|
|
|
async def _fallback_when_nothing_unseen(self, final_text: str) -> Optional[str]:
|
|
"""Fallback entered but the visible prefix already covers ``final_text``: returns the
|
|
continuation to send (the whole final when the prefix is from a *previous* segment)
|
|
or None when the turn is settled here."""
|
|
visible = self._visible_prefix()
|
|
# Telegram clients can lose (part of) a streamed preview after a failed
|
|
# final edit, so opt-in adapters commit a fresh final send.
|
|
if (final_text.strip() and final_text == visible
|
|
and getattr(self.adapter, "RESEND_FINAL_ON_EMPTY_STREAM_FALLBACK", False) is True):
|
|
delivery = await self._send_empty_fallback_final(final_text)
|
|
if delivery == "delivered":
|
|
return None
|
|
self._already_sent = True
|
|
self._fallback_prefix = ""
|
|
self._fallback_preserve_partial_messages = False
|
|
# "ambiguous" (timeout: Telegram may have accepted) and "preview" (flood:
|
|
# the complete ACKed preview is authoritative) keep dup suppression;
|
|
# "failed" lets the gateway perform its normal final send.
|
|
self._final_content_delivered = delivery in {"ambiguous", "preview"}
|
|
if delivery == "preview":
|
|
# This branch is only reached when the ACKed preview already shows the complete final text
|
|
# (final_text == _visible_prefix()), so record it as the turn-final payload: the gateway's
|
|
# reconciliation then confirms delivery instead of re-sending a second bubble next to the
|
|
# never-deleted preview (#71047 Problem B).
|
|
self._record_turn_final_payload(final_text)
|
|
elif delivery == "ambiguous":
|
|
self._delivery_ambiguous = True
|
|
else:
|
|
self._final_response_sent = False
|
|
return None
|
|
# The prefix may be from a *previous* segment (before a tool boundary),
|
|
# wrongly reading as "already shown" — send final_text as-is.
|
|
if final_text.strip() and final_text != visible:
|
|
return final_text
|
|
# Best-effort strip of a cursor left stuck by the edit failure.
|
|
if (self._message_id and self._last_sent_text and self.cfg.cursor
|
|
and self._last_sent_text.endswith(self.cfg.cursor)):
|
|
clean_text = self._last_sent_text[:-len(self.cfg.cursor)]
|
|
with contextlib.suppress(Exception):
|
|
result = await self._edit_message(message_id=self._message_id, content=clean_text)
|
|
if result.success:
|
|
self._last_sent_text = clean_text
|
|
self._already_sent = True
|
|
# Recorder substitutes the full ledger on a split turn.
|
|
self._mark_final_delivered(record=final_text)
|
|
return None
|
|
|
|
def _fallback_len_budget(self) -> "tuple[Callable[[str], int], int]":
|
|
"""(len_fn, raw_limit) for fallback chunking — per-chat cap/unit on base adapters."""
|
|
raw_limit = getattr(self.adapter, "MAX_MESSAGE_LENGTH", 4096)
|
|
_len_fn: "Callable[[str], int]" = len
|
|
if isinstance(self.adapter, _BasePlatformAdapter):
|
|
_len_fn = self.adapter.message_len_fn
|
|
try: # per-chat cap/unit (relay adapter fronting N platforms)
|
|
raw_limit = self.adapter.max_message_length_for_chat(self.chat_id)
|
|
_len_fn = self.adapter.message_len_fn_for_chat(self.chat_id)
|
|
except Exception as e:
|
|
logger.debug("per-chat limit resolution failed: %s", e)
|
|
return _len_fn, raw_limit
|
|
|
|
async def _send_with_flood_retry(self, *, content: str, retry_log: str, reply_to=None):
|
|
"""adapter.send(final metadata) with ONE bounded flood retry; returns the last
|
|
SendResult. Exceptions propagate (callers decide whether a raise is "ambiguous")."""
|
|
kwargs = dict(chat_id=self.chat_id, content=content,
|
|
metadata=self._metadata_for_send(final=True))
|
|
if reply_to is not None:
|
|
kwargs["reply_to"] = reply_to
|
|
result = None
|
|
for attempt in range(2):
|
|
result = await self.adapter.send(**kwargs)
|
|
if getattr(result, "success", False):
|
|
break
|
|
retry_delay = self._fallback_flood_retry_delay(result)
|
|
if attempt or retry_delay is None:
|
|
break # non-flood error, long flood wait, or second failure
|
|
raw = getattr(result, "raw_response", None)
|
|
if isinstance(raw, dict) and raw.get("partial_overflow"):
|
|
break # split head already on screen: re-sending the whole content duplicates it
|
|
logger.debug(retry_log, retry_delay)
|
|
await asyncio.sleep(retry_delay)
|
|
return result
|
|
|
|
async def _send_empty_fallback_final(self, final_text: str) -> str:
|
|
"""Commit a completed answer after Telegram finalization fails: "delivered", "failed"
|
|
(gateway may retry), "ambiguous" (a timeout may have landed) or "preview" (flood
|
|
control; the complete preview is authoritative)."""
|
|
# Segment-scoped only: never delete an earlier finalized preamble.
|
|
stale_ids = self._stale_preview_ids(segment_only=True)
|
|
try:
|
|
result = await self._send_with_flood_retry(
|
|
content=final_text, reply_to=self._initial_reply_to_id,
|
|
retry_log="Flood control on empty fallback final send; retrying in %.1fs")
|
|
except Exception as exc:
|
|
logger.debug("Empty fallback final send failed: %s", exc)
|
|
return "ambiguous" if self._send_failure_may_have_delivered(exc) else "failed"
|
|
if not getattr(result, "success", False):
|
|
if self._is_flood_error(result):
|
|
return "preview"
|
|
return "ambiguous" if self._send_failure_may_have_delivered(result) else "failed"
|
|
|
|
new_message_id = getattr(result, "message_id", None)
|
|
# Telegram reports delete failure by returning False; the flood window that
|
|
# broke the finalize can reject this too — one bounded retry.
|
|
await self._delete_previews(stale_ids, skip=new_message_id, label="Empty fallback",
|
|
retry_on_false=True)
|
|
self._segment_preview_message_ids = set()
|
|
self._message_id = new_message_id or "__no_edit__"
|
|
self._already_sent = True
|
|
self._mark_final_delivered()
|
|
# Record VERBATIM, not via _record_turn_final_payload: the sealed previews
|
|
# were just deleted, so the ledger (still holding sealed heads) would claim
|
|
# delivery for text this path removed.
|
|
self._delivered_final_text = self._display_payload(final_text)
|
|
self._last_sent_text = final_text
|
|
self._fallback_prefix = ""
|
|
self._fallback_preserve_partial_messages = False
|
|
self._notify_new_message()
|
|
return "delivered"
|
|
|
|
@staticmethod
|
|
def _send_failure_may_have_delivered(result_or_exc: Any) -> bool:
|
|
"""Return True for timeout failures where retrying may duplicate."""
|
|
if getattr(result_or_exc, "retryable", None) is True:
|
|
return False
|
|
error = str(getattr(result_or_exc, "error", None) or result_or_exc).lower()
|
|
name = result_or_exc.__class__.__name__.lower()
|
|
return "timeout" in error or "timed out" in error or "timeout" in name
|
|
|
|
def _fallback_flood_retry_delay(self, result: Any) -> float | None:
|
|
"""Return a bounded retry delay for a fallback send, if safe to retry."""
|
|
if not self._is_flood_error(result):
|
|
return None
|
|
try:
|
|
delay = float(getattr(result, "retry_after", None) or 3.0)
|
|
except (TypeError, ValueError):
|
|
delay = 3.0
|
|
if delay > self._max_fallback_flood_retry_seconds:
|
|
logger.debug("Flood control requests %.1fs; leaving final delivery to the gateway",
|
|
delay)
|
|
return None
|
|
return max(0.0, delay)
|
|
|
|
def _is_flood_error(self, result) -> bool:
|
|
"""Check if a SendResult failure is due to flood control / rate limiting."""
|
|
err_lower = (getattr(result, "error", "") or "").lower()
|
|
return "flood" in err_lower or "retry after" in err_lower or "rate" in err_lower
|
|
|
|
async def _flush_segment_tail_on_edit_failure(self) -> None:
|
|
"""Before a segment reset, send the unseen tail as a new message (and best-effort
|
|
strip the stuck cursor from the partial)."""
|
|
if getattr(self, "_egress_declined", False):
|
|
return # a new message is exactly the re-addressing the egress guard refused
|
|
if not self._fallback_final_send:
|
|
await self._try_strip_cursor()
|
|
visible = self._fallback_prefix or self._visible_prefix()
|
|
tail = self._accumulated
|
|
if visible and tail.startswith(visible):
|
|
tail = tail[len(visible):].lstrip()
|
|
tail = self._clean_for_display(tail)
|
|
if not tail.strip():
|
|
return
|
|
try:
|
|
# Interim: must never seal a native stream (see _send_commentary).
|
|
_md = dict(self.metadata) if self.metadata else {}
|
|
_md["_interim_send"] = True
|
|
result = await self.adapter.send(chat_id=self.chat_id, content=tail, metadata=_md)
|
|
if result.success:
|
|
self._already_sent = True
|
|
except Exception as e:
|
|
logger.error("Segment-break tail flush error: %s", e)
|
|
|
|
async def _try_strip_cursor(self) -> None:
|
|
"""Best-effort edit removing a stuck cursor when entering fallback mode."""
|
|
prefix = self._visible_prefix()
|
|
if not self._has_real_preview() or not prefix.strip():
|
|
return
|
|
with contextlib.suppress(Exception): # never block the fallback path
|
|
result = await self._edit_message(message_id=self._message_id, content=prefix)
|
|
if getattr(result, "success", False):
|
|
self._last_sent_text = prefix
|
|
|
|
async def _send_commentary(self, text: str) -> bool:
|
|
"""Send a completed interim assistant commentary message."""
|
|
text = self._clean_for_display(text)
|
|
if not text.strip():
|
|
return False
|
|
try:
|
|
# Interim: a stream-is-the-message adapter's seal-interception must not
|
|
# turn this into draft(final=true), which would seal the live stream
|
|
# with interim text and orphan the true final.
|
|
_md = self._metadata_for_send(final=False) or {}
|
|
_md["_interim_send"] = True
|
|
# reply_to only for reply-anchored threading; Discord/Telegram use
|
|
# thread_id metadata and reply_to on every commentary is spam.
|
|
_plat = getattr(getattr(self.adapter, "platform", None), "value", None)
|
|
_platform_name = str(_plat or getattr(self.adapter, "name", "")).lower()
|
|
_needs_reply_anchor = _platform_name in ("buzz", "slack", "mattermost", "feishu")
|
|
result = await self.adapter.send(
|
|
chat_id=self.chat_id, content=text,
|
|
reply_to=self._initial_reply_to_id if _needs_reply_anchor else None, metadata=_md)
|
|
# Do NOT set _already_sent: commentary is interim, and the flag would
|
|
# suppress the real final after multiple tool calls.
|
|
if result.success:
|
|
self._notify_new_message()
|
|
# Lets run.py confirm whether an interim send carried the final.
|
|
# Record the exact delivered text so run.py can confirm whether an interim "preview"
|
|
# actually carried the final response, vs. unrelated commentary delivered during a session
|
|
# split (#14238).
|
|
self._delivered_commentary_texts.append(text)
|
|
return result.success
|
|
except Exception as e:
|
|
logger.error("Commentary send error: %s", e)
|
|
return False
|
|
|
|
def _raw_message_limit(self) -> int:
|
|
"""Per-chat length budget (``message_len_fn`` units) before overflow splits; rich
|
|
adapters may raise it via ``streaming_overflow_limit`` so a reply that fits one rich
|
|
message isn't fragmented at the edit limit."""
|
|
base = getattr(self.adapter, "MAX_MESSAGE_LENGTH", 4096)
|
|
# isinstance gate keeps MagicMock adapters (mock attrs, not ints) on base.
|
|
if isinstance(self.adapter, _BasePlatformAdapter):
|
|
try:
|
|
base = self.adapter.max_message_length_for_chat(self.chat_id)
|
|
except Exception as e:
|
|
logger.debug("max_message_length_for_chat failed: %s", e)
|
|
try:
|
|
cap = self.adapter.streaming_overflow_limit()
|
|
except Exception as e:
|
|
logger.debug("streaming_overflow_limit check failed: %s", e)
|
|
cap = None
|
|
if isinstance(cap, int) and cap > base:
|
|
return cap
|
|
return base
|
|
|
|
# Fresh send carried exactly ``text`` — record it so the gateway can reconcile the flag against the
|
|
# completed response (#71643/#95382 content-vs-flag contract).
|
|
async def _suppress_silence_marker(self) -> None:
|
|
"""Retract any streamed preview when the final reply is a bare silence marker. Flags
|
|
stay False so the gateway's whole-response filter owns what goes out next: "" for a
|
|
machinery turn, the visible fallback for a human one."""
|
|
# A native-stream bubble isn't a deletable message — close an open one
|
|
# (e.g. from an eager re-seed) with an empty finalize so it doesn't hang.
|
|
if self._native_stream_opened:
|
|
await self._close_empty_native_bubble("Silence-marker native stream close failed: %s")
|
|
|
|
await self._delete_previews(self._stale_preview_ids(), label="Silence-marker")
|
|
self._preview_message_ids = set()
|
|
self._message_id = None
|
|
self._accumulated = self._stream_ledger = self._last_sent_text = ""
|
|
self._already_sent = False
|
|
self._clear_turn_final_flags()
|
|
logger.info("Suppressed streamed intentional-silence marker (chat=%s)", self.chat_id)
|