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

fix(cron): require positive evidence for live-adapter delivery confirmation

Browse Source 
A cron job fired, the scheduler logged "delivered to telegram:<chat> via
live adapter", and nothing reached Telegram (#77763). The log line was not
evidence of a send:

* the silence-narration filter returns {"success": True, "delivered": False}
  (a successful *drop*), and the dict-normalization branch read only
  "success", so a filtered message counted as delivered;
* an empty payload (no text, no media) skipped the send entirely and still
  fell into the "delivered" branch;
* the log line named the chat but not the lane, so a wrong-thread delivery
  and a phantom one are indistinguishable after the fact.

_confirm_adapter_delivery now inspects both result shapes: an explicit
`delivered: False` is a rejection even with a truthy `success`, and a
success with no message_id and no raw_response is accepted but logged as
UNVERIFIED. The empty-payload case fails closed into the existing
standalone/warn handling, and the delivered log carries thread= and
message_id=.

Failing closed on the live lane is only half the fix on a native target:
the standalone fallback sent the same empty payload, and the Telegram
adapter returns SendResult(success=True) for empty content without an API
call — a phantom live delivery became a phantom standalone one. Both
_send_to_platform call sites now sit behind one skip guard, so "empty
payload fails closed" holds on every lane (#77763).
This commit is contained in:
kaiomp
2026-09-01 01:52:10 +00:00
committed by Teknium
parent c83ea9bed7
commit dd72b42ba4
2 changed files with 353 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cron/scheduler.py
+95 -13
View File
@@ -2984,7 +2984,7 @@ def _send_media_via_adapter(
return errors
def _confirm_adapter_delivery(send_result) -> bool:
def _confirm_adapter_delivery(send_result, job_id: str = "?") -> bool:
"""Return True only if ``send_result`` unambiguously confirms delivery.
A live adapter that returns ``None`` (e.g. a swallowed exception, a busy
@@ -2993,16 +2993,52 @@ def _confirm_adapter_delivery(send_result) -> bool:
scheduler to log ``"delivered to <chat> via live adapter"`` while the
gateway never actually sees the message (#47056).
Likewise, an object missing a ``success`` attribute (e.g. a bare ``dict``
or a partial mock) is a contract violation: it does not actually tell us
whether the send succeeded. Require an explicit, truthy ``success``
attribute to count as confirmed.
Likewise, a result carrying no ``success`` at all (a partial mock, or a
``dict`` from a code path that never reached the adapter) is a contract
violation: it does not actually tell us whether the send succeeded.
Require an explicit, truthy ``success`` to count as confirmed.
Both shapes are inspected the same way, because ``_deliver_to_platform``
returns either a ``SendResult`` object or a plain ``dict``:
* ``delivered is False`` is a REJECTION even when ``success`` is truthy.
The silence-narration filter returns
``{"success": True, "delivered": False}`` — a successfully *dropped*
message, not a delivered one. Reading only ``success`` there is how a
cron brief was logged as delivered while the user got nothing (#77763).
* No ``message_id`` and no ``raw_response`` means we have no positive
evidence of a send. That is not proof of failure either (some adapters
legitimately return a bare success), so it is still accepted — but
logged at WARNING so an UNVERIFIED delivery is visible in the log
instead of masquerading as a confirmed one. Telegram ``SendResult``
objects carry ``message_id``; the dict-filter shape does not.
"""
if send_result is None:
return False
if not hasattr(send_result, "success"):
if isinstance(send_result, dict):
if "success" not in send_result:
return False
success = bool(send_result.get("success"))
delivered = send_result.get("delivered")
message_id = send_result.get("message_id")
raw_response = send_result.get("raw_response")
else:
if not hasattr(send_result, "success"):
return False
success = bool(getattr(send_result, "success"))
delivered = getattr(send_result, "delivered", None)
message_id = getattr(send_result, "message_id", None)
raw_response = getattr(send_result, "raw_response", None)
if not success or delivered is False:
return False
return bool(getattr(send_result, "success"))
if message_id is None and not raw_response:
logger.warning(
"Job '%s': live adapter reported success with no delivery evidence "
"(no message_id, no raw_response) — treating as delivered but "
"UNVERIFIED",
job_id,
)
return True
def _is_channel_dm_topic(
@@ -3530,7 +3566,20 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
adapter_ok = True
timed_out = False
delivered_message_id = None
if text_to_send:
if not text_to_send and not media_files:
# Nothing to hand the adapter at all. This used to fall
# straight through to the `if adapter_ok:` branch below and
# log "delivered to <chat> via live adapter" for a send that
# never happened (#77763). Fail closed so the run reports
# the empty payload instead.
msg = (
f"live adapter send skipped (empty text and no media) "
f"for {platform_name}:{chat_id}"
)
logger.warning("Job '%s': %s", job["id"], msg)
target_errors.append(msg)
adapter_ok = False
elif text_to_send:
from agent.async_utils import safe_schedule_threadsafe
router = DeliveryRouter(config, adapters)
@@ -3623,19 +3672,27 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
# {"success": True, "delivered": False, ...}.
# Normalize both shapes so a getattr default doesn't
# misread a dict, and so a None / success-less object
# is NOT counted as delivered (#47056).
# is NOT counted as delivered (#47056). The
# confirmation itself handles both shapes: a truthy
# `success` with `delivered: False` is a drop, not a
# delivery (#77763).
if isinstance(send_result, dict):
send_success = bool(send_result.get("success", False))
send_raw_response = send_result.get("raw_response")
delivered_message_id = send_result.get("message_id")
else:
send_success = _confirm_adapter_delivery(send_result)
send_raw_response = getattr(send_result, "raw_response", None)
delivered_message_id = getattr(send_result, "message_id", None)
send_success = _confirm_adapter_delivery(send_result, job["id"])
if not send_success:
if isinstance(send_result, dict):
err = send_result.get("error", "unknown")
# A filtered drop carries no "error" — name
# the filter instead of reporting "unknown".
err = (
send_result.get("error")
or send_result.get("filtered")
or "unknown"
)
shape = "dict"
elif send_result is not None:
err = getattr(send_result, "error", None)
@@ -3712,7 +3769,16 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
delivery_errors.append(msg)
if adapter_ok:
logger.info("Job '%s': delivered to %s:%s via live adapter", job["id"], platform_name, chat_id)
# Log WHERE it went, not just that it went: a ghost delivery
# that landed in the wrong lane (General topic instead of the
# routed thread) is indistinguishable from a real one without
# the routing identity (#77763).
logger.info(
"Job '%s': delivered to %s:%s via live adapter thread=%s message_id=%s",
job["id"], platform_name, chat_id,
route_thread_id if route_thread_id is not None else "-",
delivered_message_id if delivered_message_id is not None else "-",
)
delivered = True
# Seed the thread session only now that delivery into it
# succeeded (deferred from thread-open above).
@@ -3826,6 +3892,22 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
target_errors.append(msg)
delivery_errors.extend(target_errors)
continue
# The live lane already failed closed on an empty payload; the
# standalone senders do not. The Telegram adapter returns
# SendResult(success=True) for empty content WITHOUT an API call,
# so falling through here turns a phantom live delivery into a
# phantom standalone one and logs it as delivered (#77763). Both
# _send_to_platform call sites below are reached through this
# point, so one guard closes the lane.
if not cleaned_delivery_content.strip() and not media_files:
msg = (
f"standalone send skipped (empty text and no media) "
f"for {platform_name}:{chat_id}"
)
logger.warning("Job '%s': %s", job["id"], msg)
target_errors.append(msg)
delivery_errors.extend(target_errors)
continue
# Standalone path: run the async send in a fresh event loop (safe from any thread)
coro = _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)
try:
tests/cron/test_cron_live_delivery_confirmation.py
+258
View File
@@ -0,0 +1,258 @@
"""Live-adapter delivery confirmation for cron (#77763).
A ``no_agent`` job fired, the scheduler logged
``delivered to telegram:<chat> via live adapter``, and the user received
nothing — no message row, no delivery obligation. The log line was not
evidence of a send:
* the silence-narration filter returns ``{"success": True, "delivered": False}``
(a successful *drop*), and the normalization block read only ``success``;
* an empty payload skipped the send entirely and still fell into the
"delivered" branch;
* the log line named the chat but not the lane, so a wrong-thread delivery and
a phantom one look identical after the fact.
These tests pin the confirmation contract: positive evidence, honest logging,
and fail-closed on nothing-to-send.
"""
import asyncio
import logging
from concurrent.futures import Future
from unittest.mock import MagicMock, patch
import pytest
from cron import scheduler as sched
from cron.scheduler import _confirm_adapter_delivery, _deliver_result
from gateway.config import Platform, PlatformConfig
# ---------------------------------------------------------------------------
# _confirm_adapter_delivery: the contract in isolation
# ---------------------------------------------------------------------------
class _SendResult:
"""Minimal stand-in for an adapter SendResult."""
def __init__(self, success=True, message_id=None, raw_response=None, **extra):
self.success = success
self.message_id = message_id
self.raw_response = raw_response
for key, value in extra.items():
setattr(self, key, value)
class TestConfirmAdapterDelivery:
def test_none_is_not_delivered(self):
assert _confirm_adapter_delivery(None, "j1") is False
def test_missing_success_is_not_delivered(self):
assert _confirm_adapter_delivery(object(), "j1") is False
assert _confirm_adapter_delivery({"message_id": 7}, "j1") is False
def test_explicit_failure_is_not_delivered(self):
assert _confirm_adapter_delivery(_SendResult(success=False), "j1") is False
assert _confirm_adapter_delivery({"success": False}, "j1") is False
def test_filtered_dict_is_not_delivered(self):
"""The exact silence-filter shape: a successful DROP is not a delivery."""
filtered = {"success": True, "filtered": "silence_narration", "delivered": False}
assert _confirm_adapter_delivery(filtered, "j1") is False
def test_delivered_false_on_an_object_is_not_delivered(self):
result = _SendResult(success=True, message_id=42, delivered=False)
assert _confirm_adapter_delivery(result, "j1") is False
def test_positive_evidence_is_delivered_without_warning(self, caplog):
with caplog.at_level(logging.WARNING, logger="cron.scheduler"):
assert _confirm_adapter_delivery(_SendResult(message_id=1234), "j1") is True
assert "UNVERIFIED" not in caplog.text
def test_raw_response_alone_counts_as_evidence(self, caplog):
with caplog.at_level(logging.WARNING, logger="cron.scheduler"):
result = _SendResult(raw_response={"ok": True})
assert _confirm_adapter_delivery(result, "j1") is True
assert "UNVERIFIED" not in caplog.text
def test_evidence_free_success_is_accepted_but_warned(self, caplog):
"""Not proof of failure either — accept it, but say so in the log."""
with caplog.at_level(logging.WARNING, logger="cron.scheduler"):
assert _confirm_adapter_delivery(_SendResult(), "92e639af907f") is True
assert "UNVERIFIED" in caplog.text
assert "92e639af907f" in caplog.text
def test_evidence_free_success_dict_is_accepted_but_warned(self, caplog):
with caplog.at_level(logging.WARNING, logger="cron.scheduler"):
assert _confirm_adapter_delivery({"success": True}, "j1") is True
assert "UNVERIFIED" in caplog.text
# ---------------------------------------------------------------------------
# _deliver_result: the live lane end to end
# ---------------------------------------------------------------------------
CHAT_ID = "-1001234567890"
def _job(thread_id=None):
origin = {"platform": "telegram", "chat_id": CHAT_ID}
if thread_id is not None:
origin["thread_id"] = thread_id
return {
"id": "92e639af907f",
"name": "Ghost Delivery",
"deliver": "origin",
"origin": origin,
}
def _gateway_config(relay=False):
config = MagicMock()
platforms = {Platform.TELEGRAM: PlatformConfig(enabled=True)}
if relay:
platforms[Platform.RELAY] = PlatformConfig(enabled=True)
config.platforms = platforms
config.get_home_channel = lambda p: None
return config
def _adapters(relay=False):
adapter = MagicMock()
if relay:
adapter.fronts_platform = lambda p: p == Platform.TELEGRAM
return {Platform.RELAY: adapter}
return {Platform.TELEGRAM: adapter}
def _run(job, content, send_result, relay=False, standalone_result=None):
"""Drive ``_deliver_result`` over the live lane with a stubbed router.
Returns ``(error, router_calls, standalone_calls)``.
"""
loop = MagicMock()
loop.is_running.return_value = True
def fake_run_coro(coro, _loop):
future = Future()
try:
future.set_result(asyncio.run(coro))
except BaseException as e: # noqa: BLE001
future.set_exception(e)
return future
router_calls = []
standalone_calls = []
router = MagicMock()
async def _deliver_to_platform(target, text, metadata):
router_calls.append({"target": target, "text": text, "metadata": metadata})
return send_result
router._deliver_to_platform = _deliver_to_platform
async def _fake_send_to_platform(platform, pconfig, chat_id, text, **kwargs):
standalone_calls.append({"chat_id": chat_id, "text": text, "kwargs": kwargs})
return standalone_result if standalone_result is not None else {}
with patch("gateway.config.load_gateway_config", return_value=_gateway_config(relay)), \
patch("cron.scheduler.load_config",
return_value={"cron": {"wrap_response": False}}), \
patch("gateway.delivery.DeliveryRouter", return_value=router), \
patch("tools.send_message_tool._send_to_platform", _fake_send_to_platform), \
patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
error = _deliver_result(job, content, adapters=_adapters(relay), loop=loop)
return error, router_calls, standalone_calls
class TestFilteredResultIsNotDelivered:
FILTERED = {"success": True, "filtered": "silence_narration", "delivered": False}
def test_filtered_dict_does_not_log_a_live_delivery(self, caplog):
with caplog.at_level(logging.INFO, logger="cron.scheduler"):
_, router_calls, standalone_calls = _run(_job(), "...", self.FILTERED)
assert len(router_calls) == 1 # the live send was attempted
assert "via live adapter" not in caplog.text # but never claimed as delivered
assert len(standalone_calls) == 1 # fell back instead of lying
def test_filtered_dict_fails_closed_on_the_relay_lane(self):
"""Relay owns the destination, so there is no fallback — report it."""
error, _, standalone_calls = _run(_job(), "...", self.FILTERED, relay=True)
assert error is not None
assert "unconfirmed result" in error
assert "silence_narration" in error # names the filter, not "unknown"
assert standalone_calls == []
def test_confirmed_send_result_still_delivers(self, caplog):
with caplog.at_level(logging.INFO, logger="cron.scheduler"):
error, router_calls, standalone_calls = _run(
_job(), "Nightly report.", _SendResult(message_id=1234),
)
assert error is None
assert len(router_calls) == 1
assert standalone_calls == []
assert "via live adapter" in caplog.text
class TestEmptyPayloadFailsClosed:
def test_empty_payload_never_reaches_the_adapter(self, caplog):
with caplog.at_level(logging.INFO, logger="cron.scheduler"):
_, router_calls, _ = _run(_job(), " ", _SendResult(message_id=1))
assert router_calls == [] # nothing was sent
assert "via live adapter" not in caplog.text # and nothing was claimed
assert "empty text and no media" in caplog.text
def test_empty_payload_never_reaches_the_standalone_sender(self, caplog):
"""The native fallback must not re-open the hole the live lane closed.
Telegram's adapter returns ``SendResult(success=True)`` for empty
content without an API call, so an unguarded fallback would log a
standalone "delivered" for the same phantom payload (#77763).
"""
with caplog.at_level(logging.INFO, logger="cron.scheduler"):
error, router_calls, standalone_calls = _run(
_job(), " ", _SendResult(message_id=1),
)
assert router_calls == []
assert standalone_calls == [] # _send_to_platform never called
assert error is not None
assert "standalone send skipped (empty text and no media)" in error
assert "delivered to" not in caplog.text
def test_empty_payload_is_reported_on_the_relay_lane(self):
error, router_calls, _ = _run(_job(), "", _SendResult(message_id=1), relay=True)
assert router_calls == []
assert error is not None
assert "live adapter send skipped (empty text and no media)" in error
class TestDeliveredLogNamesTheLane:
def test_log_includes_thread_and_message_id(self, caplog):
with caplog.at_level(logging.INFO, logger="cron.scheduler"):
error, _, _ = _run(
_job(thread_id="99"), "Nightly report.", _SendResult(message_id=1234),
)
assert error is None
assert "via live adapter thread=99 message_id=1234" in caplog.text
def test_log_uses_a_dash_when_the_lane_is_unknown(self, caplog):
"""No thread and an evidence-free result must still be attributable."""
with caplog.at_level(logging.INFO, logger="cron.scheduler"):
error, _, _ = _run(_job(), "Nightly report.", _SendResult())
assert error is None
assert "via live adapter thread=- message_id=-" in caplog.text
assert "UNVERIFIED" in caplog.text
def test_scheduler_module_exposes_the_confirmation_helper():
"""Guard the import surface the delivery block depends on."""
assert callable(sched._confirm_adapter_delivery)