fix(gateway): stop payload-less split delivery from swallowing Telegram finals

gateway/run.py

@@ -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:
@@ -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

gateway/stream_consumer.py

@@ -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
@@ -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
@@ -405,20 +411,30 @@ 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.
 
         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).
+
+        For multi-message split delivery, prefer ``_stream_ledger`` (the
+        unsplit segment text) so reconciliation still works (#78541).  Older
+        code cleared the record on split and forced legacy trust; that let a
+        payload-less stale flag suppress later complete replies.
         """
-        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]:
@@ -432,22 +448,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
@@ -541,6 +559,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 = ""
@@ -666,7 +685,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:
@@ -678,7 +697,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
@@ -687,7 +706,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
@@ -732,7 +751,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:
@@ -939,11 +958,14 @@ 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._stream_ledger or "".join(chunks)
+                                )
                             return
                         if got_segment_break:
                             self._message_id = None
@@ -1987,6 +2009,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

tests/gateway/test_stale_finalize_suppression.py

@@ -257,6 +257,96 @@ async def test_equal_text_control_still_suppresses_duplicate_send(
     assert len(full_sends) <= 1, f"duplicate final delivery: {full_sends!r}"
 
 
+class _PayloadLessSplitConsumer(GatewayStreamConsumer):
+    """Force the #78541 shape after a normal stream drain.
+
+    Claims final delivery via the multi-message split path but leaves no
+    recorded payload — the pre-fix gateway treated matcher ``None`` as
+    legacy trust and swallowed the complete ``final_response``.
+    """
+
+    async def run(self):
+        await super().run()
+        self._final_response_sent = True
+        self._final_content_delivered = True
+        self._turn_split_delivery = True
+        self._delivered_final_text = None
+        self._stream_ledger = ""
+
+
+@pytest.mark.asyncio
+async def test_payload_less_split_does_not_suppress_complete_response(
+    monkeypatch, tmp_path
+):
+    """#78541 — payload-less split-delivery flags must not swallow the reply."""
+    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 = StalePrefixAgent
+    monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+
+    gateway_run = importlib.import_module("gateway.run")
+    stream_consumer_mod = importlib.import_module("gateway.stream_consumer")
+    # run.py imports GatewayStreamConsumer locally inside _run_agent — patch
+    # the defining module so the local import picks up the sabotage subclass.
+    monkeypatch.setattr(
+        stream_consumer_mod, "GatewayStreamConsumer", _PayloadLessSplitConsumer
+    )
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setattr(
+        gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
+    )
+
+    adapter = FinalizeCaptureAdapter()
+    runner = _make_runner(adapter)
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="-1004492624436",
+        chat_type="group",
+        thread_id="1",
+    )
+    result = await runner._run_agent(
+        message="describe this photo",
+        context_prompt="",
+        history=[],
+        source=source,
+        session_id="sess-78541-payload-less-split",
+        session_key="agent:main:telegram:group:-1004492624436:1",
+    )
+
+    assert result["final_response"] == FULL_RESPONSE
+    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}"
+    )
+    # Must not silently claim delivery without putting the complete text on
+    # the wire (the production ghost: already_sent with no full payload).
+    if result.get("already_sent"):
+        assert any(
+            FULL_RESPONSE in payload for payload in all_payloads
+        ), "already_sent=True but complete response never reached the platform"
+
+
 # ---------------------------------------------------------------------------
 # Consumer unit coverage: delivered_final_matches tri-state
 # ---------------------------------------------------------------------------
@@ -284,11 +374,27 @@ def test_stale_prefix_record_returns_false(self):
         consumer._record_turn_final_payload(STREAMED_PREFIX)
         assert consumer.delivered_final_matches(FULL_RESPONSE) is False
 
-    def test_split_delivery_returns_none(self):
+    def test_payload_less_split_delivery_returns_false(self):
+        """#78541 — payload-less split must not inherit legacy trust."""
         consumer = _consumer()
         consumer._turn_split_delivery = True
+        consumer._delivered_final_text = None
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is False
+
+    def test_split_delivery_with_matching_ledger_returns_true(self):
+        """Complete overflow split that recorded its ledger still suppresses."""
+        consumer = _consumer()
+        consumer._turn_split_delivery = True
+        consumer._stream_ledger = FULL_RESPONSE
+        consumer._record_turn_final_payload(STREAMED_PREFIX)  # tail only; ledger wins
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is True
+
+    def test_split_delivery_with_stale_ledger_returns_false(self):
+        consumer = _consumer()
+        consumer._turn_split_delivery = True
+        consumer._stream_ledger = STREAMED_PREFIX
         consumer._record_turn_final_payload(STREAMED_PREFIX)
-        assert consumer.delivered_final_matches(FULL_RESPONSE) is None
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is False
 
     def test_empty_final_text_returns_none(self):
         consumer = _consumer()
@@ -308,3 +414,4 @@ def test_reset_segment_state_clears_record(self):
         consumer._reset_segment_state()
         assert consumer._delivered_final_text is None
         assert consumer._turn_split_delivery is False
+        assert consumer._stream_ledger == ""