ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

fix(gateway): key the queued-lane obligation on the raw inbound id

Browse Source 
The queued lane's ledger bracket used the reply anchor as the obligation's
message reference. The anchor is None wherever replies are not used (Telegram
forum topics, Slack reaction handoffs), so two turns in one topic answering
with the same text shared an obligation id and the second record overwrote
the first's outstanding row; the id also differed from the normal lane's.

The lane now runs the adapter's record / send-with-retry / finalize sequence
itself, keyed on turn_ctx.inbound_message_id like the normal lane, while the
anchor stays the reply target. Tests cover the forum-topic identity, the row
being `attempting` while the send is in flight, and the call site passing
both ids.
This commit is contained in:
Alexander Russell
2026-09-05 21:06:58 +03:00
committed by kshitij
parent c237c492cd
commit af09c4033e
3 changed files with 112 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/run_notifications.py
+32 -22
View File
@@ -317,12 +317,14 @@ class GatewayNotificationsMixin:
self, response: str, source: SessionSource, adapter,
metadata: Optional[Dict[str, Any]] = None, event_message_id: Optional[str] = None,
text_already_delivered: bool = False, deliver_media: bool = True, stream_consumer=None,
session_key: Optional[str] = None,
session_key: Optional[str] = None, inbound_message_id: Optional[str] = None,
) -> None:
"""Deliver a queued response using the normal text+attachment split.
``session_key`` lets the text send record a delivery-ledger obligation like the normal final
send does (see ``_send_queued_final_text``); without it the send stays unledgered."""
send does, keyed on ``inbound_message_id`` (the raw inbound id, distinct from the
``event_message_id`` reply anchor); see ``_send_queued_final_text``. Without a key the send
stays unledgered."""
from gateway.run import _strip_response_attachments_for_direct_send
if not text_already_delivered:
text_content = _strip_response_attachments_for_direct_send(response, adapter)
@@ -363,7 +365,8 @@ class GatewayNotificationsMixin:
logger.debug("Queued-lane reconcile edit failed (%s); falling back to send.", _qe)
if not _reconciled:
await self._send_queued_final_text(
adapter, source, text_content, metadata, event_message_id, session_key)
adapter, source, text_content, metadata, event_message_id, session_key,
inbound_message_id)
# Failed turns deliver their (normalized failure) text but must not upload attachments as if
# they succeeded — mirrors the ``not agent_result.get("failed")`` completed-turn guard.
if not deliver_media:
@@ -376,27 +379,34 @@ class GatewayNotificationsMixin:
async def _send_queued_final_text(
self, adapter, source: SessionSource, text_content: str, metadata: Optional[Dict[str, Any]],
event_message_id: Optional[str], session_key: Optional[str],
inbound_message_id: Optional[str] = None,
):
"""Send a queued-lane final through the adapter's ledger-bracketed final send.
"""Send a queued-lane final under the delivery-ledger bracket the normal final send uses.
This lane used to call ``adapter.send`` bare and discard the result. A final refused here
(flood control, a transport that just died) was gone for good: no delivery-ledger row was ever
written, so neither the boot sweep nor the runtime redelivery could see it, and the queued
follow-up ran as if the answer had landed. The normal final send records the obligation before
the send and finalizes it from the result; this routes the queued lane through that same method
so both lanes recover the same way. The event carries the inbound message id, so the obligation
id matches the one the normal lane would compute for the same turn (idempotent re-record), and
the reply is marked notify-worthy like every other user-visible final. Adapters without the
base contract (relay doubles, third-party plugins) and sends without a session key keep the
plain send. Returns the SendResult, or None when the adapter's send returned nothing
(queued-final-ledger)."""
bracketed = getattr(adapter, "_send_final_text", None)
if session_key and callable(bracketed):
results: list = []
await bracketed(
MessageEvent(text="", source=source, message_id=event_message_id), session_key,
text_content, _mark_notify_metadata(metadata), False, 0, results.append)
result = results[-1] if results else None
This lane used to call ``adapter.send`` bare and discard the result, so a final refused here
(flood control, a transport that had just died) left no ledger row and was gone for good.
Now the obligation is recorded before the send and finalized from the result, the same
record / send-with-retry / finalize sequence as ``_send_final_text``, so both lanes recover
the same way. The ledger id is keyed on the raw inbound message id, as the normal lane's is;
``event_message_id`` is only the reply anchor, which is None wherever replies are not used
(Telegram forum topics, Slack reaction handoffs) and so cannot identify the turn. Adapters
without the base contract and sends without a session key keep the plain send. Returns the
SendResult, or None when the adapter's send returned nothing."""
record = getattr(adapter, "_record_delivery_obligation", None)
finalize = getattr(adapter, "_finalize_delivery_obligation", None)
transport_for = getattr(adapter, "_final_delivery_adapter", None)
if session_key and callable(record) and callable(finalize) and callable(transport_for):
delivery_adapter = transport_for(source)
ledger_message_id = (
inbound_message_id if inbound_message_id is not None else event_message_id)
ledger_event = MessageEvent(text="", source=source, message_id=ledger_message_id)
obligation_id = await record(
ledger_event, session_key, text_content, delivery_adapter, False)
result = await delivery_adapter._send_with_retry(
chat_id=source.chat_id, content=text_content, reply_to=event_message_id,
metadata=_mark_notify_metadata(metadata))
if obligation_id is not None:
await finalize(obligation_id, result, ledger_event, delivery_adapter)
else:
result = await adapter.send(source.chat_id, text_content, metadata=metadata)
if not getattr(result, "success", False):
gateway/run_turn.py
+3 -2
View File
@@ -3383,8 +3383,9 @@ class GatewayTurnMixin:
metadata=turn_ctx._status_thread_metadata, event_message_id=turn_ctx.event_message_id,
text_already_delivered=_already_streamed,
deliver_media=not _delivery_result.get("failed"), stream_consumer=_sc,
# The text send records a delivery-ledger obligation under this key (queued-final-ledger).
session_key=session_key,
# The text send records a delivery-ledger obligation under this key, keyed on
# the raw inbound id (the anchor above is only the reply target).
session_key=session_key, inbound_message_id=turn_ctx.inbound_message_id,
)
except Exception as e:
logger.warning("Failed to send first response before queued message: %s", e)
tests/gateway/test_queued_final_ledger.py
+77 -28
View File
@@ -9,12 +9,14 @@ sweep nor the runtime redelivery could see it, and the follow-up ran as if the a
had landed. On 5 Sep 2026 two long replies were lost this way within an hour, both
refused by Telegram flood control while a delegation batch arrived the same second.
Now the queued lane routes its text send through the adapter's ``_send_final_text``,
the same ledger-bracketed method the normal lane uses: the obligation is recorded
before the send under the id the normal lane would compute for the same turn, the
result finalizes it, and the reply is marked notify-worthy like every other final.
The reconcile-by-edit path is unchanged. Adapters without the base contract and
sends without a session key keep the plain send.
Now the queued lane runs the same bracket as the normal lane: the obligation is
recorded before the send, the send goes through the adapter's retrying transport, and
the result finalizes the row. The obligation id is keyed on the raw inbound message id
exactly as the normal lane keys it; the reply anchor is only the reply target and is
None wherever replies are not used (Telegram forum topics), so it cannot identify the
turn. The reply is marked notify-worthy like every other final. The reconcile-by-edit
path is unchanged. Adapters without the base contract and sends without a session key
keep the plain send.
"""
from __future__ import annotations
@@ -31,6 +33,7 @@ from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, SendResult
SESSION_KEY = "agent:main:telegram:dm:5230977008"
TOPIC_SESSION_KEY = "agent:main:telegram:group:-1001:topic:7"
CHAT = "5230977008"
INBOUND_ID = "5301"
TEXT = "**Yes.** I would make room for one island trip and one Balkan trip."
@@ -46,17 +49,19 @@ def _fresh_db(tmp_path, monkeypatch):
yield
_COLUMNS = ("obligation_id", "session_key", "state", "attempts", "last_error", "content", "chat_id",
"platform")
def _rows():
with dl._connect() as conn:
cur = conn.execute(
"SELECT obligation_id, session_key, state, attempts, last_error, content, chat_id, platform"
" FROM delivery_obligations")
return [dict(zip(("obligation_id", "session_key", "state", "attempts", "last_error", "content",
"chat_id", "platform"), r)) for r in cur.fetchall()]
cur = conn.execute(f"SELECT {', '.join(_COLUMNS)} FROM delivery_obligations")
return [dict(zip(_COLUMNS, r)) for r in cur.fetchall()]
def _source():
return SimpleNamespace(platform=Platform.TELEGRAM, chat_id=CHAT, thread_id=None, chat_type="dm")
def _source(*, chat_id=CHAT, thread_id=None, chat_type="dm"):
return SimpleNamespace(platform=Platform.TELEGRAM, chat_id=chat_id, thread_id=thread_id,
chat_type=chat_type)
def _telegram_adapter(send_result=None):
@@ -69,7 +74,8 @@ def _telegram_adapter(send_result=None):
runner._schedule_flood_redelivery = MagicMock(return_value=187.0)
adapter.gateway_runner = runner
adapter.send = AsyncMock(return_value=send_result or SendResult(success=True, message_id="900"))
adapter.edit_message = AsyncMock(return_value=SendResult(success=False, error="message to edit not found"))
adapter.edit_message = AsyncMock(
return_value=SendResult(success=False, error="message to edit not found"))
return adapter
@@ -86,13 +92,14 @@ def _runner():
return object.__new__(GatewayRunner)
async def _deliver(adapter, *, session_key=SESSION_KEY, stream_consumer=None, metadata=None, text=TEXT):
async def _deliver(adapter, *, session_key=SESSION_KEY, stream_consumer=None, metadata=None,
text=TEXT, source=None, anchor=INBOUND_ID, inbound_id=INBOUND_ID):
from gateway.run import GatewayRunner
await GatewayRunner._deliver_queued_first_response(
_runner(), text, source=_source(), adapter=adapter, metadata=metadata, event_message_id=INBOUND_ID,
text_already_delivered=False, deliver_media=False, stream_consumer=stream_consumer,
session_key=session_key)
_runner(), text, source=source or _source(), adapter=adapter, metadata=metadata,
event_message_id=anchor, text_already_delivered=False, deliver_media=False,
stream_consumer=stream_consumer, session_key=session_key, inbound_message_id=inbound_id)
# ---------------------------------------------------------------------------
@@ -114,6 +121,23 @@ async def test_a_delivered_queued_final_is_recorded_and_marked_delivered():
adapter.send.assert_awaited_once()
@pytest.mark.asyncio
async def test_the_row_is_already_attempting_while_the_send_is_in_flight():
"""Record BEFORE the send: a crash mid-send must leave a row the next boot can redeliver."""
adapter = _telegram_adapter()
seen: list = []
async def _send(**kwargs):
seen.append([(r["state"], r["content"]) for r in _rows()])
return SendResult(success=True, message_id="900")
adapter.send = AsyncMock(side_effect=_send)
await _deliver(adapter)
assert seen == [[("attempting", TEXT)]]
@pytest.mark.asyncio
async def test_the_obligation_id_is_the_one_the_normal_lane_would_use():
"""Same turn, same text: re-recording from either lane is idempotent, never a second row."""
@@ -124,9 +148,29 @@ async def test_the_obligation_id_is_the_one_the_normal_lane_would_use():
assert _rows()[0]["obligation_id"] == dl.compute_obligation_id(SESSION_KEY, INBOUND_ID, TEXT)
@pytest.mark.asyncio
async def test_forum_topic_turns_are_identified_by_their_inbound_id_not_the_reply_anchor():
"""Telegram forum topics route by topic metadata and never reply, so the anchor is None for
every message in the topic. Two turns answering with the same text must still get two rows,
or the second would overwrite the first's outstanding obligation and retry state."""
adapter = _telegram_adapter()
topic = _source(chat_id="-1001", thread_id="7", chat_type="supergroup")
await _deliver(adapter, session_key=TOPIC_SESSION_KEY, source=topic, anchor=None,
inbound_id="5301")
await _deliver(adapter, session_key=TOPIC_SESSION_KEY, source=topic, anchor=None,
inbound_id="5302")
ids = sorted(r["obligation_id"] for r in _rows())
assert ids == sorted([dl.compute_obligation_id(TOPIC_SESSION_KEY, "5301", TEXT),
dl.compute_obligation_id(TOPIC_SESSION_KEY, "5302", TEXT)])
assert [c.kwargs["reply_to"] for c in adapter.send.await_args_list] == [None, None]
@pytest.mark.asyncio
async def test_a_flood_refused_queued_final_stays_in_the_ledger_as_failed(caplog):
"""The row survives the refusal, so the sweeps (and a flood timer, where present) can redeliver."""
"""The row survives the refusal, so the sweeps (and a flood timer, where present) can redeliver
it."""
adapter = _telegram_adapter(SendResult(success=False, error="flood_control:185.0"))
with caplog.at_level(logging.WARNING, logger="gateway.run"):
@@ -136,13 +180,14 @@ async def test_a_flood_refused_queued_final_stays_in_the_ledger_as_failed(caplog
assert len(rows) == 1
assert rows[0]["state"] == "failed"
assert rows[0]["last_error"] == "flood_control:185.0"
assert any("Queued-lane final send" in r.getMessage() and "flood_control:185.0" in r.getMessage()
for r in caplog.records)
assert any("Queued-lane final send" in r.getMessage()
and "flood_control:185.0" in r.getMessage() for r in caplog.records)
@pytest.mark.asyncio
async def test_any_other_refusal_is_recorded_as_failed_too():
adapter = _telegram_adapter(SendResult(success=False, error="Forbidden: bot was blocked by the user"))
adapter = _telegram_adapter(
SendResult(success=False, error="Forbidden: bot was blocked by the user"))
await _deliver(adapter)
@@ -228,11 +273,11 @@ async def test_a_plain_send_failure_is_still_logged(caplog):
# ---------------------------------------------------------------------------
# The call site hands the session key over.
# The call site hands the session key and the raw inbound id over.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_the_queued_lane_receives_the_session_key_from_the_turn():
async def test_the_queued_lane_receives_the_session_key_and_inbound_id_from_the_turn():
from gateway.run import GatewayRunner
runner = _runner()
@@ -240,9 +285,12 @@ async def test_the_queued_lane_receives_the_session_key_from_the_turn():
runner._is_intentional_silence = MagicMock(return_value=False)
runner._pop_post_delivery_callback = MagicMock(return_value=None)
runner._deliver_queued_first_response = AsyncMock()
# A forum-topic turn: no reply anchor, but a raw inbound id the ledger must be keyed on.
turn_ctx = SimpleNamespace(
session_key=SESSION_KEY, stream_consumer_holder=[None], source=_source(),
_status_thread_metadata={"thread_id": None}, event_message_id=INBOUND_ID, run_generation=3)
session_key=TOPIC_SESSION_KEY, stream_consumer_holder=[None],
source=_source(chat_id="-1001", thread_id="7", chat_type="supergroup"),
_status_thread_metadata={"thread_id": "7"}, event_message_id=None,
inbound_message_id=INBOUND_ID, run_generation=3)
adapter = _telegram_adapter()
await GatewayRunner._run_agent_deliver_first_response(
@@ -250,6 +298,7 @@ async def test_the_queued_lane_receives_the_session_key_from_the_turn():
runner._deliver_queued_first_response.assert_awaited_once()
kwargs = runner._deliver_queued_first_response.await_args.kwargs
assert kwargs["session_key"] == SESSION_KEY
assert kwargs["event_message_id"] == INBOUND_ID
assert kwargs["session_key"] == TOPIC_SESSION_KEY
assert kwargs["inbound_message_id"] == INBOUND_ID
assert kwargs["event_message_id"] is None
assert kwargs["text_already_delivered"] is False