RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(gateway): stop stale streamed finalize from suppressing the complete Telegram response 

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
This commit is contained in:
Teknium 2026-07-31 23:46:37 -07:00
parent c05f0bb81d
commit 30878411b8
3 changed files with 483 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
gateway/run.py
View File

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

97
gateway/stream_consumer.py
View File

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

310
tests/gateway/test_stale_finalize_suppression.py Normal file
View File

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