From 30878411b8e36f43484f8bb07f6ac48be3bebcea Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 31 Jul 2026 23:46:37 -0700 Subject: [PATCH] fix(gateway): stop stale streamed finalize from suppressing the complete Telegram response MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A successful finalize edit can carry only the last streamed preview snapshot: deltas generated between the last preview edit and stream completion never reach any Bot API call, yet final_response_sent / final_content_delivered were set from the call's success and suppressed the gateway's normal final send — losing the tail permanently. The stream consumer now records the exact cleaned payload of every turn-final delivery (delivered_final_matches tri-state), and gateway/run.py reconciles that record against the completed final_response before trusting either suppression flag. On a demonstrable mismatch it edits the streamed message up to the complete response, falling back to the normal final send if the edit fails. Multi-message split deliveries and legacy paths without a record keep the existing flag-trusting behavior, so overflow splits and the failed-finalize handling (#51828/#33793) are untouched. Fixes #71643 --- gateway/run.py | 76 +++++ gateway/stream_consumer.py | 97 ++++++ .../test_stale_finalize_suppression.py | 310 ++++++++++++++++++ 3 files changed, 483 insertions(+) create mode 100644 tests/gateway/test_stale_finalize_suppression.py diff --git a/gateway/run.py b/gateway/run.py index c28f695530987..9e6b2e745cb43 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -24017,6 +24017,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if consumer is None: return False if getattr(consumer, "final_response_sent", False): + # A successful finalize call is not proof the *content* was + # final: the edit may have carried only the last preview + # snapshot while the tail generated between that snapshot and + # 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. + matcher = getattr(consumer, "delivered_final_matches", None) + if callable(matcher): + try: + if matcher(final_text) is False: + return False + except Exception: + pass return True if previewed: has_delivered_text = getattr(consumer, "has_delivered_text", None) @@ -24703,6 +24719,27 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _content_delivered = bool( _sc and getattr(_sc, "final_content_delivered", False) ) + # #71643: a *successful* finalize edit can still carry only the + # last preview snapshot — deltas generated between that edit and + # stream completion never reach any API call, and both suppression + # flags are set from the call's success rather than its content. + # 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 + # (#51828 / #33793) is unaffected because those paths leave the + # flags False or record the complete fallback payload. + _stale_finalized = False + if _content_delivered and not _is_empty_sentinel: + _matcher = getattr(_sc, "delivered_final_matches", None) + if callable(_matcher): + try: + _stale_finalized = _matcher(_final) is False + except Exception: + _stale_finalized = False + if _stale_finalized: + _content_delivered = False # Plugin hooks (e.g. transform_llm_output) may have appended content # after streaming finished — when the response was transformed, always # send the final version so the appended content reaches the client. @@ -24727,6 +24764,45 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _content_delivered, ) response["already_sent"] = True + elif not _is_empty_sentinel and not _transformed and _stale_finalized and _sc is not None: + # Stale finalize (#71643): the streamed message holds only the + # last preview snapshot. Prefer editing it up to the complete + # response (same shape as the transformed branch below) so the + # user gets one corrected message; on edit failure fall through + # with already_sent unset so the normal final send delivers the + # complete text. + _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: + try: + _reconcile_res = await _sc_adapter.edit_message( + chat_id=source.chat_id, + message_id=_sc_msg_id, + content=_final, + finalize=True, + ) + if getattr(_reconcile_res, "success", True): + response["already_sent"] = True + logger.info( + "Reconciled stale streamed finalize for session %s: edited message %s with the complete response (#71643).", + session_key or "?", _sc_msg_id, + ) + else: + logger.warning( + "Stale-finalize reconciliation edit failed for session %s (%s); sending complete response via normal final send.", + session_key or "?", + getattr(_reconcile_res, "error", None), + ) + except Exception as _edit_err: + logger.warning( + "Stale-finalize reconciliation edit failed for session %s: %s; sending complete response via normal final send.", + session_key or "?", _edit_err, + ) + else: + logger.info( + "Stale streamed finalize detected for session %s with no editable message; delivering complete response via normal final send (#71643).", + session_key or "?", + ) elif not _is_empty_sentinel and _transformed and _sc is not None: # Plugin hooks transformed the response after streaming — edit the # existing streamed message instead of sending a duplicate. diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index 0e601be801606..58c0aebdaad48 100644 --- a/gateway/stream_consumer.py +++ b/gateway/stream_consumer.py @@ -261,6 +261,19 @@ class GatewayStreamConsumer: # streaming, even if the final edit (cursor removal etc.) # subsequently failed. self._final_content_delivered = False + # Exact cleaned payload of the turn-final delivery that set the flags + # above. The gateway compares this against the completed + # ``final_response`` before trusting the flags: a *successful* finalize + # edit that carried only a stale preview snapshot must not suppress the + # complete send (#71643). ``None`` means "no record" — legacy trust, + # so paths that predate the record keep their behavior. + 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. + self._turn_split_delivery = False self._delivered_commentary_texts: list[str] = [] # Retains the finalized visible text of each streaming segment so # ``has_delivered_text`` can still match after ``_reset_segment_state`` @@ -392,6 +405,57 @@ class GatewayStreamConsumer: pass return await self.adapter.edit_message(**kwargs) + def _record_turn_final_payload(self, text: str) -> None: + """Record the exact cleaned payload of a turn-final delivery. + + 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``. + """ + if self._turn_split_delivery: + self._delivered_final_text = None + return + self._delivered_final_text = ensure_closed_code_fences( + self._clean_for_display(text or "") + ).strip() + + def delivered_final_matches(self, final_text: str) -> Optional[bool]: + """Reconcile the recorded turn-final payload against ``final_text``. + + Returns a tri-state verdict for the gateway's suppression decision + (#71643 — a *successful* finalize edit can still carry only a stale + preview snapshot, so call success alone must not confirm delivery): + + - ``True`` — the recorded turn-final payload (or a previously + 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. + """ + 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.strip() == target: + return True + # A segment break / commentary may have delivered the final text + # earlier in the turn under a different record. + if self.has_delivered_text(final_text): + return True + return False + def has_delivered_text(self, text: str) -> bool: """Return True if *text* was already delivered as visible chat content.""" target = self._clean_for_display(text or "").strip() @@ -488,6 +552,8 @@ class GatewayStreamConsumer: # run.py reads these only after the consumer task exits. self._final_response_sent = False self._final_content_delivered = False + self._delivered_final_text = None + self._turn_split_delivery = False # Native draft streaming: bump the draft_id so the next text segment # animates as a fresh preview below the tool-progress bubbles, not # over the prior segment's already-finalized draft. This is how @@ -873,6 +939,11 @@ class GatewayStreamConsumer: 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). + self._turn_split_delivery = True + self._delivered_final_text = None return if got_segment_break: self._message_id = None @@ -926,6 +997,9 @@ class GatewayStreamConsumer: self._accumulated = self._accumulated[split_at:].lstrip("\n") self._message_id = None self._last_sent_text = "" + # Sealed head chunk delivered — this turn is now a + # multi-message delivery (#71643 record semantics). + self._turn_split_delivery = True display_text = self._accumulated if not got_done and not got_segment_break and commentary_text is None: @@ -964,6 +1038,7 @@ class GatewayStreamConsumer: # edit here would duplicate the message / re-delete, # so just record delivery and stop. self._final_content_delivered = True + self._record_turn_final_payload(self._accumulated) elif ( current_update_visible and ( @@ -983,6 +1058,7 @@ class GatewayStreamConsumer: # on screen. self._final_response_sent = True self._final_content_delivered = True + self._record_turn_final_payload(self._accumulated) elif self._message_id: # Either the mid-stream edit didn't run (no # visible update this tick) OR the adapter needs @@ -992,6 +1068,7 @@ class GatewayStreamConsumer: ) if self._final_response_sent: self._final_content_delivered = True + self._record_turn_final_payload(self._accumulated) elif self._fallback_final_send: # The final edit attempt itself may be the one # that exhausts flood-control strikes and @@ -1005,6 +1082,7 @@ class GatewayStreamConsumer: self._final_response_sent = await self._send_or_edit(self._accumulated) if self._final_response_sent: self._final_content_delivered = True + self._record_turn_final_payload(self._accumulated) return if commentary_text is not None: @@ -1082,6 +1160,7 @@ class GatewayStreamConsumer: if _best_effort_ok and not self._final_response_sent: self._final_response_sent = True self._final_content_delivered = True + self._record_turn_final_payload(self._accumulated) except Exception as e: logger.error("Stream consumer error: %s", e) finally: @@ -1320,6 +1399,8 @@ class GatewayStreamConsumer: self._already_sent = True 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() return raw_limit = getattr(self.adapter, "MAX_MESSAGE_LENGTH", 4096) @@ -1422,6 +1503,10 @@ class GatewayStreamConsumer: self._already_sent = True self._final_response_sent = True self._final_content_delivered = True + # 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() self._last_sent_text = chunks[-1] self._fallback_prefix = "" self._fallback_preserve_partial_messages = False @@ -1492,6 +1577,8 @@ class GatewayStreamConsumer: self._already_sent = True 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() self._last_sent_text = final_text self._fallback_prefix = "" self._fallback_preserve_partial_messages = False @@ -1904,6 +1991,8 @@ class GatewayStreamConsumer: self._already_sent = False self._final_response_sent = False self._final_content_delivered = False + self._delivered_final_text = None + self._turn_split_delivery = False logger.info( "Suppressed streamed intentional-silence marker (chat=%s)", self.chat_id, @@ -2082,6 +2171,9 @@ class GatewayStreamConsumer: and result.message_id != self._message_id ): self._last_edit_overflowed = True + # Adapter adopted continuation messages — this + # turn is a multi-message delivery (#71643). + self._turn_split_delivery = True self._message_id = str(result.message_id) self._message_created_ts = time.monotonic() self._last_sent_text = "" @@ -2109,6 +2201,11 @@ class GatewayStreamConsumer: # when Telegram/Discord rate-limit this cosmetic # final edit (#36965, #25349). 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() 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, diff --git a/tests/gateway/test_stale_finalize_suppression.py b/tests/gateway/test_stale_finalize_suppression.py new file mode 100644 index 0000000000000..e1244a7d85654 --- /dev/null +++ b/tests/gateway/test_stale_finalize_suppression.py @@ -0,0 +1,310 @@ +"""Regression coverage for #71643 — stale streamed finalize suppression. + +A *successful* Telegram finalize edit can carry only the last streamed +preview snapshot: deltas generated between the last preview edit and stream +completion never reach any Bot API call, yet ``final_response_sent`` / +``final_content_delivered`` are set from the call's success and suppress the +gateway's normal final send. The missing tail is then lost with no retry. + +These tests exercise the real gateway boundary (``GatewayRunner._run_agent`` +with a live ``GatewayStreamConsumer``), per the review guidance on #71643: + +1. fake agent emits a visible prefix through ``stream_delta_callback``; +2. the consumer successfully finalizes that prefix; +3. the agent returns a longer ``final_response`` containing a missing tail; +4. the result must NOT silently suppress — the complete final response must + reach the platform (reconciliation edit or normal final send); +5. control: when the streamed text exactly equals the final text, the + suppression still occurs (no duplicate delivery). + +Plus unit coverage for ``GatewayStreamConsumer.delivered_final_matches``. +""" + +import importlib +import sys +import types +from types import SimpleNamespace + +import pytest + +from gateway.config import Platform, PlatformConfig, StreamingConfig +from gateway.platforms.base import BasePlatformAdapter, SendResult +from gateway.session import SessionSource +from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig + + +# --------------------------------------------------------------------------- +# Boundary-test fakes +# --------------------------------------------------------------------------- + + +class FinalizeCaptureAdapter(BasePlatformAdapter): + """Adapter that records every send/edit with its finalize flag.""" + + def __init__(self, platform=Platform.TELEGRAM): + super().__init__(PlatformConfig(enabled=True, token="***"), platform) + self.sent = [] + self.edits = [] + self._next_id = 0 + + async def connect(self, *, is_reconnect: bool = False) -> bool: + return True + + async def disconnect(self) -> None: + return None + + def _mint_id(self) -> str: + self._next_id += 1 + return f"m-{self._next_id}" + + async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult: + self.sent.append({"chat_id": chat_id, "content": content, "metadata": metadata}) + return SendResult(success=True, message_id=self._mint_id()) + + async def edit_message( + self, chat_id, message_id, content, *, finalize: bool = False, metadata=None + ) -> SendResult: + self.edits.append( + { + "chat_id": chat_id, + "message_id": message_id, + "content": content, + "finalize": finalize, + } + ) + return SendResult(success=True, message_id=message_id) + + async def send_typing(self, chat_id, metadata=None) -> None: + return None + + async def stop_typing(self, chat_id) -> None: + return None + + async def get_chat_info(self, chat_id: str): + return {"id": chat_id} + + +STREAMED_PREFIX = "The photo shows a dog on a beach" +MISSING_TAIL = " with a red frisbee in its mouth, mid-leap over the surf." +FULL_RESPONSE = STREAMED_PREFIX + MISSING_TAIL + + +class StalePrefixAgent: + """Streams only a prefix; the completed response carries a longer tail. + + Models the #71643 incident shape: the tail generated between the last + preview edit and stream completion never reaches the stream callback, so + the consumer's successful finalize edit carries stale preview text while + ``final_response`` holds the complete answer. + """ + + def __init__(self, **kwargs): + self.stream_delta_callback = kwargs.get("stream_delta_callback") + self.tools = [] + + def run_conversation(self, message, conversation_history=None, task_id=None): + if self.stream_delta_callback: + self.stream_delta_callback(STREAMED_PREFIX) + return { + "final_response": FULL_RESPONSE, + "response_previewed": False, + "messages": [], + "api_calls": 1, + } + + +class CompleteStreamAgent: + """Control: the streamed text exactly equals the final response.""" + + def __init__(self, **kwargs): + self.stream_delta_callback = kwargs.get("stream_delta_callback") + self.tools = [] + + def run_conversation(self, message, conversation_history=None, task_id=None): + if self.stream_delta_callback: + self.stream_delta_callback(FULL_RESPONSE) + return { + "final_response": FULL_RESPONSE, + "response_previewed": False, + "messages": [], + "api_calls": 1, + } + + +def _make_runner(adapter): + gateway_run = importlib.import_module("gateway.run") + runner = object.__new__(gateway_run.GatewayRunner) + runner.adapters = {adapter.platform: adapter} + runner._voice_mode = {} + runner._prefill_messages = [] + runner._ephemeral_system_prompt = "" + runner._reasoning_config = None + runner._provider_routing = {} + runner._fallback_model = None + runner._session_db = None + runner._running_agents = {} + runner._session_run_generation = {} + runner.session_store = SimpleNamespace(_entries={}, _save=lambda: None) + runner.hooks = SimpleNamespace(loaded_hooks=False) + runner.config = SimpleNamespace( + thread_sessions_per_user=False, + group_sessions_per_user=False, + stt_enabled=False, + streaming=StreamingConfig.from_dict( + {"enabled": True, "edit_interval": 0.01, "buffer_threshold": 1} + ), + ) + return runner + + +async def _run_streaming_turn(monkeypatch, tmp_path, agent_cls, session_id): + import yaml + + (tmp_path / "config.yaml").write_text( + yaml.dump( + { + "display": {"tool_progress": "off", "interim_assistant_messages": False}, + "streaming": { + "enabled": True, + "edit_interval": 0.01, + "buffer_threshold": 1, + }, + } + ), + encoding="utf-8", + ) + + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + + fake_run_agent = types.ModuleType("run_agent") + fake_run_agent.AIAgent = agent_cls + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + + adapter = FinalizeCaptureAdapter() + runner = _make_runner(adapter) + gateway_run = importlib.import_module("gateway.run") + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr( + gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"} + ) + + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="-1001", + chat_type="group", + ) + result = await runner._run_agent( + message="describe this photo", + context_prompt="", + history=[], + source=source, + session_id=session_id, + session_key="agent:main:telegram:group:-1001", + ) + return adapter, result + + +# --------------------------------------------------------------------------- +# Gateway-boundary regression (#71643) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_stale_finalize_does_not_suppress_complete_response( + monkeypatch, tmp_path +): + """The complete response must reach the platform even when the finalize + edit succeeded with only the stale preview snapshot.""" + adapter, result = await _run_streaming_turn( + monkeypatch, tmp_path, StalePrefixAgent, "sess-71643-stale-finalize" + ) + + assert result["final_response"] == FULL_RESPONSE + # The missing tail must appear in at least one platform call — either the + # reconciliation edit or the normal final send. On the buggy path it + # appears in NO call at all (message loss). + all_payloads = [c["content"] for c in adapter.sent] + [ + e["content"] for e in adapter.edits + ] + assert any(FULL_RESPONSE in payload for payload in all_payloads), ( + f"complete response never reached the platform; payloads: {all_payloads!r}" + ) + # The preferred recovery is an in-place reconciliation edit of the + # streamed message (single corrected message, no duplicate). + if result.get("already_sent"): + assert any( + e["content"] == FULL_RESPONSE and e["finalize"] for e in adapter.edits + ), "already_sent=True but no edit carried the complete response" + + +@pytest.mark.asyncio +async def test_equal_text_control_still_suppresses_duplicate_send( + monkeypatch, tmp_path +): + """When the streamed text equals the final response, suppression must + keep working — no duplicate full-response send.""" + adapter, result = await _run_streaming_turn( + monkeypatch, tmp_path, CompleteStreamAgent, "sess-71643-control-equal" + ) + + assert result["final_response"] == FULL_RESPONSE + assert result.get("already_sent") is True + # Exactly one platform message holds the answer: the streamed message + # (created by one send, then edited). No duplicate full send. + full_sends = [c for c in adapter.sent if FULL_RESPONSE in c["content"]] + assert len(full_sends) <= 1, f"duplicate final delivery: {full_sends!r}" + + +# --------------------------------------------------------------------------- +# Consumer unit coverage: delivered_final_matches tri-state +# --------------------------------------------------------------------------- + + +def _consumer(): + adapter = FinalizeCaptureAdapter() + return GatewayStreamConsumer( + adapter, "chat-1", StreamConsumerConfig(cursor=" ▉") + ) + + +class TestDeliveredFinalMatches: + def test_no_record_returns_none(self): + consumer = _consumer() + assert consumer.delivered_final_matches("anything") is None + + def test_matching_record_returns_true(self): + consumer = _consumer() + consumer._record_turn_final_payload(FULL_RESPONSE) + assert consumer.delivered_final_matches(FULL_RESPONSE) is True + + def test_stale_prefix_record_returns_false(self): + consumer = _consumer() + consumer._record_turn_final_payload(STREAMED_PREFIX) + assert consumer.delivered_final_matches(FULL_RESPONSE) is False + + def test_split_delivery_returns_none(self): + consumer = _consumer() + consumer._turn_split_delivery = True + consumer._record_turn_final_payload(STREAMED_PREFIX) + assert consumer.delivered_final_matches(FULL_RESPONSE) is None + + def test_empty_final_text_returns_none(self): + consumer = _consumer() + consumer._record_turn_final_payload(STREAMED_PREFIX) + assert consumer.delivered_final_matches("") is None + + def test_segment_delivered_text_still_matches(self): + consumer = _consumer() + consumer._record_turn_final_payload(STREAMED_PREFIX) + # A prior segment delivered the exact final text. + consumer._delivered_segment_texts.append(FULL_RESPONSE) + assert consumer.delivered_final_matches(FULL_RESPONSE) is True + + def test_reset_segment_state_clears_record(self): + consumer = _consumer() + consumer._record_turn_final_payload(STREAMED_PREFIX) + consumer._reset_segment_state() + assert consumer._delivered_final_text is None + assert consumer._turn_split_delivery is False