fix(cron): make cron push-notify configurable (cron.delivery.notify) and surface UNVERIFIED live deliveries in cron list/doctor

De-risking for the notify=True UX change: the marker is now driven by
cron.delivery.notify (config.yaml, default true = current behaviour), read
once per delivery and applied to both the text and media routes; a missing or
malformed section keeps the default.

An evidence-free live-adapter ack (bare SendResult(success=True) from
Slack/Matrix/Mattermost) is still accepted, but the target is recorded on the
job as last_delivery_unverified (cleared by the next evidenced delivery) so
the state shows up in 'hermes cron list' (⚠ Delivery UNVERIFIED), 'hermes cron
doctor', and the cronjob tool listing — not only in a WARNING log line.

Live repro (real _deliver_result + real 'hermes cron list' against a temp
HERMES_HOME, Slack target, SendResult(success=True)): before — list showed
nothing beyond the Deliver line and route metadata always carried
notify=true; after — list prints the UNVERIFIED line, and
cron.delivery.notify: false yields notify=false in the route metadata.

cron/jobs.py

@@ -2450,6 +2450,9 @@ def create_job(
         "last_status": None,
         "last_error": None,
         "last_delivery_error": None,
+        # Live-adapter targets whose last send was acked with no message_id /
+        # raw_response (accepted, but UNVERIFIED — surfaced by cron list/doctor).
+        "last_delivery_unverified": None,
         "failure_streak": 0,
         # Delivery configuration
         "deliver": deliver,

cron/scheduler.py

@@ -2984,7 +2984,7 @@ def _send_media_via_adapter(
     return errors
 
 
-def _confirm_adapter_delivery(send_result, job_id: str = "?") -> bool:
+def _confirm_adapter_delivery(send_result, job_id: str = "?", unverified: Optional[list] = None) -> bool:
     """Return True only if ``send_result`` unambiguously confirms delivery.
 
     A live adapter that returns ``None`` (e.g. a swallowed exception, a busy
@@ -3038,6 +3038,8 @@ def _confirm_adapter_delivery(send_result, job_id: str = "?") -> bool:
             "UNVERIFIED",
             job_id,
         )
+        if unverified is not None:
+            unverified.append(True)
     return True
 
 
@@ -3097,6 +3099,48 @@ def _is_channel_dm_topic(
     return is_channel
 
 
+def _cron_delivery_notify_enabled(cfg: Optional[dict]) -> bool:
+    """Resolve ``cron.delivery.notify`` (config.yaml). Default True.
+
+    Only an explicit boolean ``False`` (or a YAML ``false``/``off`` that parses
+    to it) disables the push notification; a missing/malformed section keeps
+    the default so a typo can never silently make cron briefs silent.
+    """
+    try:
+        cron_cfg = (cfg or {}).get("cron")
+        if not isinstance(cron_cfg, dict):
+            return True
+        delivery_cfg = cron_cfg.get("delivery")
+        if not isinstance(delivery_cfg, dict):
+            return True
+        return delivery_cfg.get("notify", True) is not False
+    except Exception:
+        return True
+
+
+def _record_delivery_verification(job: dict, unverified_targets: list) -> None:
+    """Persist the UNVERIFIED-delivery marker on the job record.
+
+    ``last_delivery_unverified`` is a list of ``platform:chat_id`` targets
+    whose live adapter acked the send with no message_id/raw_response, or
+    ``None`` once a run delivered with positive evidence (or to no live
+    target). Skips the write when nothing changed so the common verified
+    path costs no jobs.json save. Never raises — status bookkeeping must not
+    fail a delivery.
+    """
+    new_value = list(unverified_targets) or None
+    if (job.get("last_delivery_unverified") or None) == new_value:
+        return
+    try:
+        from cron.jobs import update_job
+
+        update_job(job["id"], {"last_delivery_unverified": new_value})
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.debug(
+            "Job '%s': could not record delivery verification: %s", job.get("id"), exc,
+        )
+
+
 def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Optional[str]:
     """
     Deliver job output to the configured target(s) (origin chat, specific platform, etc.).
@@ -3144,6 +3188,18 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
     except Exception:
         pass
 
+    # cron.delivery.notify (default True): mark live-adapter cron sends as
+    # FINAL notifications so the platform pushes them (Telegram's "important"
+    # mode otherwise sends with disable_notification=True). Configurable so
+    # operators who prefer silent briefs can opt back out.
+    notify_delivery = _cron_delivery_notify_enabled(user_cfg)
+    # Set when a live adapter acked a send with NO delivery evidence (no
+    # message_id / raw_response — the Slack/Matrix/Mattermost bare
+    # SendResult(success=True) shape). Persisted on the job as
+    # ``last_delivery_unverified`` so `hermes cron list` shows the state
+    # instead of it living only in a WARNING log line.
+    unverified_targets: list = []
+
     if wrap_response:
         task_name = job.get("name", job["id"])
         job_id = job.get("id", "")
@@ -3519,13 +3575,13 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 route_metadata = {
                     "direct_messages_topic_id": str(thread_id),
                     "job_id": job["id"],
-                    "notify": True,
+                    "notify": notify_delivery,
                 }
                 # Media metadata mirrors the text routing so attachments land in
                 # the same DM topic instead of the General lane (#22773).
                 media_metadata = {
                     "direct_messages_topic_id": str(thread_id),
-                    "notify": True,
+                    "notify": notify_delivery,
                 }
             else:
                 # Forum-style topic (private chat / supergroup) or non-topic
@@ -3537,10 +3593,10 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 # anchor, so the metadata key bypasses that check and lets the
                 # adapter route via a plain message_thread_id.
                 route_thread_id = str(thread_id) if thread_id is not None else None
-                route_metadata = {"job_id": job["id"], "notify": True}
+                route_metadata = {"job_id": job["id"], "notify": notify_delivery}
                 if route_thread_id:
                     route_metadata["thread_id"] = route_thread_id
-                media_metadata = {"notify": True}
+                media_metadata = {"notify": notify_delivery}
                 if thread_id:
                     media_metadata["thread_id"] = thread_id
 
@@ -3688,7 +3744,12 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                             else:
                                 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"])
+                            _evidence_gap: list = []
+                            send_success = _confirm_adapter_delivery(
+                                send_result, job["id"], _evidence_gap,
+                            )
+                            if send_success and _evidence_gap:
+                                unverified_targets.append(f"{platform_name}:{chat_id}")
 
                             if not send_success:
                                 if isinstance(send_result, dict):
@@ -4003,6 +4064,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
     if policy_drop_errors:
         # Filter-time drops apply to every target; report them once.
         delivery_errors.extend(policy_drop_errors)
+    _record_delivery_verification(job, unverified_targets)
     if delivery_errors:
         return "; ".join(delivery_errors)
     return None

hermes_cli/config_defaults.py

@@ -2814,6 +2814,15 @@ DEFAULT_CONFIG = {
         # Wrap delivered cron responses with a header (task name) and footer
         # ("The agent cannot see this message").  Set to false for clean output.
         "wrap_response": True,
+        # Delivery behaviour for cron output sent through a live gateway adapter.
+        "delivery": {
+            # Mark cron deliveries as FINAL notifications so the platform pushes
+            # them (Telegram's "important" notification mode otherwise sends
+            # every non-notify message with disable_notification=True, and users
+            # report the silent brief as "never delivered"). Set to false to
+            # restore silent (no-push) cron deliveries.
+            "notify": True,
+        },
         # Make cron deliveries CONTINUABLE: a user can reply to a cron brief
         # and the agent has it in context (no "what is Task #2?" amnesia).
         # Default False preserves the historical isolation guarantee (cron

hermes_cli/cron.py

@@ -292,6 +292,17 @@ def cron_list(show_all: bool = False):
         if delivery_err:
             print(f"    {color('⚠ Delivery failed:', Colors.YELLOW)} {delivery_err}")
 
+        # A live adapter acked the last send but returned no message_id /
+        # raw_response (Slack/Matrix/Mattermost shape): accepted as delivered,
+        # but say so here rather than only in a WARNING log line.
+        unverified = job.get("last_delivery_unverified")
+        if unverified:
+            targets = ", ".join(str(t) for t in unverified) if isinstance(unverified, list) else str(unverified)
+            print(
+                f"    {color('⚠ Delivery UNVERIFIED:', Colors.YELLOW)} "
+                f"adapter acked {targets} without message_id/raw_response"
+            )
+
         fire_err = job.get("last_fire_error")
         if isinstance(fire_err, dict) and fire_err.get("detail"):
             print(
@@ -705,6 +716,11 @@ def _cron_doctor_issues_for_job(job: Dict[str, Any]) -> List[str]:
     if delivery_err:
         issues.append(f"last delivery failed: {delivery_err}")
 
+    unverified = job.get("last_delivery_unverified")
+    if unverified:
+        targets = ", ".join(str(t) for t in unverified) if isinstance(unverified, list) else str(unverified)
+        issues.append(f"last delivery unverified (adapter acked without evidence): {targets}")
+
     if job.get("enabled", True) and job.get("state") not in {"paused", "completed"}:
         next_run = str(job.get("next_run_at") or "").strip()
         if not next_run:

tests/cron/test_cron_live_delivery_confirmation.py

@@ -125,10 +125,18 @@ def _adapters(relay=False):
     return {Platform.TELEGRAM: adapter}
 
 
-def _run(job, content, send_result, relay=False, standalone_result=None):
+RECORDED_VERIFICATION = []
+
+
+def _record_verification(job, unverified_targets):
+    RECORDED_VERIFICATION.append((job["id"], list(unverified_targets)))
+
+
+def _run(job, content, send_result, relay=False, standalone_result=None, cron_cfg=None):
     """Drive ``_deliver_result`` over the live lane with a stubbed router.
 
-    Returns ``(error, router_calls, standalone_calls)``.
+    Returns ``(error, router_calls, standalone_calls)``. ``cron_cfg`` extends
+    the ``cron:`` section handed to the scheduler (default: unwrapped output).
     """
     loop = MagicMock()
     loop.is_running.return_value = True
@@ -143,6 +151,7 @@ def _run(job, content, send_result, relay=False, standalone_result=None):
 
     router_calls = []
     standalone_calls = []
+    RECORDED_VERIFICATION.clear()
 
     router = MagicMock()
 
@@ -158,7 +167,8 @@ def _run(job, content, send_result, relay=False, standalone_result=None):
 
     with patch("gateway.config.load_gateway_config", return_value=_gateway_config(relay)), \
          patch("cron.scheduler.load_config",
-               return_value={"cron": {"wrap_response": False}}), \
+               return_value={"cron": {"wrap_response": False, **(cron_cfg or {})}}), \
+         patch("cron.scheduler._record_delivery_verification", side_effect=_record_verification), \
          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):
@@ -301,6 +311,94 @@ class TestLiveDeliveryIsAFinalNotification:
         assert sent[0]["metadata"]["notify"] is True
 
 
+class TestNotifyIsConfigurable:
+    """``cron.delivery.notify`` (config.yaml) gates the notify marker.
+
+    The current behaviour (push notification) stays the default; only an
+    explicit ``false`` restores silent deliveries. The knob rides both the
+    text route and the media route so the two never disagree.
+    """
+
+    def test_default_is_notify(self):
+        _, router_calls, _ = _run(_job(), "Nightly report.", _SendResult(message_id=1))
+        assert router_calls[0]["metadata"]["notify"] is True
+
+    def test_explicit_false_disables_notify_on_text_route(self):
+        _, router_calls, _ = _run(
+            _job(thread_id="99"), "Nightly report.", _SendResult(message_id=1),
+            cron_cfg={"delivery": {"notify": False}},
+        )
+        metadata = router_calls[0]["metadata"]
+        assert metadata["notify"] is False
+        assert metadata["thread_id"] == "99"  # routing untouched
+
+    def test_explicit_false_disables_notify_on_media_route(self, tmp_path):
+        media = tmp_path / "report.png"
+        media.write_bytes(b"\x89PNG\r\n\x1a\n")
+        sent = []
+
+        def fake_send_media(adapter, chat_id, media_files, metadata, loop, job, platform=None):
+            sent.append(metadata)
+            return []
+
+        with patch("cron.scheduler._send_media_via_adapter", side_effect=fake_send_media), \
+             patch("gateway.platforms.base.BasePlatformAdapter.filter_media_delivery_paths",
+                   side_effect=lambda files: files):
+            _run(
+                _job(), f"Nightly report.\nMEDIA:{media}", _SendResult(message_id=1),
+                cron_cfg={"delivery": {"notify": False}},
+            )
+        assert sent[0]["notify"] is False
+
+    @pytest.mark.parametrize("cron_cfg", [
+        {"delivery": None},            # `delivery:` with no body parses to null
+        {"delivery": "yes"},           # malformed scalar
+        {"delivery": {"notify": None}},  # `notify:` with no value
+    ])
+    def test_malformed_section_keeps_the_default(self, cron_cfg):
+        _, router_calls, _ = _run(_job(), "Nightly report.", _SendResult(message_id=1), cron_cfg=cron_cfg)
+        assert router_calls[0]["metadata"]["notify"] is True
+
+    def test_default_config_ships_notify_true(self):
+        from hermes_cli.config_defaults import DEFAULT_CONFIG
+
+        assert DEFAULT_CONFIG["cron"]["delivery"]["notify"] is True
+
+
+class TestUnverifiedDeliveryIsRecordedOnTheJob:
+    """An evidence-free ack is accepted, but the state must reach the job
+    record (and from there ``hermes cron list`` / ``cron doctor``), not only a
+    WARNING log line."""
+
+    def test_evidence_free_ack_records_the_target(self):
+        error, _, _ = _run(_job(), "Nightly report.", _SendResult())
+        assert error is None
+        assert RECORDED_VERIFICATION == [("92e639af907f", [f"telegram:{CHAT_ID}"])]
+
+    def test_positive_evidence_clears_the_marker(self):
+        error, _, _ = _run(_job(), "Nightly report.", _SendResult(message_id=1234))
+        assert error is None
+        assert RECORDED_VERIFICATION == [("92e639af907f", [])]
+
+    def test_recorder_skips_the_write_when_nothing_changed(self):
+        with patch("cron.jobs.update_job") as update_job:
+            sched._record_delivery_verification({"id": "j1", "last_delivery_unverified": None}, [])
+            update_job.assert_not_called()
+            sched._record_delivery_verification({"id": "j1", "last_delivery_unverified": None}, ["slack:C1"])
+            update_job.assert_called_once_with("j1", {"last_delivery_unverified": ["slack:C1"]})
+
+    def test_recorder_clears_a_stale_marker(self):
+        with patch("cron.jobs.update_job") as update_job:
+            sched._record_delivery_verification({"id": "j1", "last_delivery_unverified": ["slack:C1"]}, [])
+            update_job.assert_called_once_with("j1", {"last_delivery_unverified": None})
+
+    def test_tool_listing_exposes_the_field(self):
+        from tools.cronjob_tools import _format_job
+
+        assert _format_job({"id": "j1", "name": "n", "prompt": "p",
+                            "last_delivery_unverified": ["slack:C1"]})["last_delivery_unverified"] == ["slack:C1"]
+
+
 def test_scheduler_module_exposes_the_confirmation_helper():
     """Guard the import surface the delivery block depends on."""
     assert callable(sched._confirm_adapter_delivery)

tests/hermes_cli/test_cron.py

@@ -129,6 +129,44 @@ class TestCronCommandLifecycle:
         assert jobs[0]["name"] == "Skill combo"
 
 
+class TestUnverifiedDeliveryVisibility:
+    """An evidence-free live-adapter ack (Slack/Matrix/Mattermost bare
+    ``SendResult(success=True)``) is accepted as delivered, but the UNVERIFIED
+    state must be visible in ``hermes cron list`` and ``hermes cron doctor``,
+    not only in a WARNING log line."""
+
+    def _seed(self):
+        job = create_job(prompt="Nightly brief", schedule="every 1h", deliver="slack:C0123456")
+        jobs = load_jobs()
+        jobs[0]["last_status"] = "ok"
+        jobs[0]["last_delivery_unverified"] = ["slack:C0123456"]
+        save_jobs(jobs)
+        return job
+
+    def test_list_shows_unverified_delivery(self, tmp_cron_dir, capsys):
+        job = self._seed()
+        cron_command(Namespace(cron_command="list", all=True, json=False))
+        out = capsys.readouterr().out
+        assert job["id"] in out
+        assert "Delivery UNVERIFIED" in out
+        assert "slack:C0123456" in out
+        assert "without message_id/raw_response" in out
+
+    def test_list_is_quiet_when_delivery_was_verified(self, tmp_cron_dir, capsys):
+        create_job(prompt="Nightly brief", schedule="every 1h", deliver="slack:C0123456")
+        cron_command(Namespace(cron_command="list", all=True, json=False))
+        assert "UNVERIFIED" not in capsys.readouterr().out
+
+    def test_doctor_reports_unverified_delivery(self, tmp_cron_dir, capsys):
+        job = self._seed()
+        rc = cron_command(Namespace(cron_command="doctor"))
+        out = capsys.readouterr().out
+        assert rc == 1
+        assert job["id"] in out
+        assert "last delivery unverified" in out
+        assert "slack:C0123456" in out
+
+
 class TestCronDoctor:
     def test_doctor_reports_cron_health_issues(self, tmp_cron_dir, capsys):
         job = create_job(prompt="Daily digest", schedule="every 1h", script="missing.py")

tools/cronjob_tools.py

@@ -770,6 +770,7 @@ def _format_job(job: Dict[str, Any]) -> Dict[str, Any]:
         "last_run_at": job.get("last_run_at"),
         "last_status": job.get("last_status"),
         "last_delivery_error": job.get("last_delivery_error"),
+        "last_delivery_unverified": job.get("last_delivery_unverified"),
         "last_fire_error": job.get("last_fire_error"),
         "enabled": job.get("enabled", True),
         # Derive from enabled so half-paused records never render as paused.

website/docs/user-guide/features/cron.md

@@ -502,6 +502,43 @@ cron:
   wrap_response: false
 ```
 
+### Push notifications (`cron.delivery.notify`)
+
+Cron output is a *final* delivery, not a progress message, so by default it is
+sent with the platform's notification flag set — on Telegram this means the
+brief triggers a push even when the adapter's notification mode is `important`
+(which otherwise sends with `disable_notification=true`, and users report the
+silent brief as "never delivered"). To restore silent deliveries:
+
+```yaml
+# ~/.hermes/config.yaml
+cron:
+  delivery:
+    notify: false   # default: true
+```
+
+The flag rides both the text send and any media attachments, so a run never
+pushes for one and stays silent for the other.
+
+### Delivery confirmation and the `UNVERIFIED` state
+
+A live-adapter delivery is logged as delivered only on positive evidence from
+the adapter: an explicit `success` that is not a filtered drop
+(`delivered: false`), plus a `message_id` or `raw_response`. A result carrying
+`success` but neither piece of evidence — the shape Slack, Matrix and
+Mattermost adapters return — is still accepted (it is not proof of failure),
+but the run is recorded on the job as `last_delivery_unverified` and surfaces
+in `hermes cron list`:
+
+```
+⚠ Delivery UNVERIFIED: adapter acked slack:C0123456 without message_id/raw_response
+```
+
+and in `hermes cron doctor` as `last delivery unverified (...)`. The marker is
+cleared by the next run that delivers with evidence. An empty payload (no text
+and no media) is never handed to an adapter; it fails closed and is reported in
+`last_delivery_error` instead of being logged as delivered.
+
 ### Continuable jobs (reply to a cron delivery)
 
 By default a cron delivery is fire-and-forget: the message is sent, but it does