From 21589a2b785c6c42c98649b44269e30a06a18b8f Mon Sep 17 00:00:00 2001
From: beardthelion <beardthelion@users.noreply.github.com>
Date: Mon, 1 Jun 2026 18:12:53 -0500
Subject: [PATCH 1/8] feat(cron): surface cron deliveries in the next turn's
 system prompt

Push side of cron session-awareness. Cron deliveries don't enter the
interactive message history (the assistant-role mirror was removed in
#2313 because consecutive assistant turns break alternation, #2221), so
the agent was blind to what its own jobs sent. This buffers each delivery
and folds it into the SYSTEM PROMPT of the chat's next interactive turn,
which is alternation-safe (same vehicle as the auto-reset context note).

- cron/pending_notices.py: record()/drain(), single JSON store keyed by
  platform:chat_id, per-key cap, lock + atomic replace, fully best-effort
- cron/scheduler.py: _deliver_result records a notice after each successful
  target send, gated by cron.notify_session (default True); buffers the raw
  job output (MEDIA stripped), not the delivery wrapper
- gateway/run.py: _build_cron_delivery_note drains pending notices for the
  source chat and prepends a [System note: ...] block to context_prompt,
  then the buffer is cleared

Does NOT reintroduce the message-history mirror; test_no_mirror_to_session_call
still passes.

Tests: 18 new (10 pending_notices, 2 scheduler, 6 run.py note); 208 impacted
tests pass under scripts/run_tests.sh isolation.

(cherry picked from commit 4f2155e3fa2e88fe89d1c3660763c97e65c3a38b)
---
 cron/pending_notices.py                  | 124 +++++++++++++++++++++++
 cron/scheduler.py                        |  52 +++++++++-
 gateway/run.py                           |  53 +++++++++-
 tests/cron/test_pending_notices.py       |  61 +++++++++++
 tests/cron/test_scheduler.py             |  50 +++++++++
 tests/gateway/test_cron_delivery_note.py |  59 +++++++++++
 6 files changed, 397 insertions(+), 2 deletions(-)
 create mode 100644 cron/pending_notices.py
 create mode 100644 tests/cron/test_pending_notices.py
 create mode 100644 tests/gateway/test_cron_delivery_note.py

diff --git a/cron/pending_notices.py b/cron/pending_notices.py
new file mode 100644
index 0000000000000..d69b0e4d31750
--- /dev/null
+++ b/cron/pending_notices.py
@@ -0,0 +1,124 @@
+"""Pending cron delivery notices (push side of cron session-awareness).
+
+Cron deliveries do NOT enter the interactive conversation history. The
+assistant-role mirror that used to do that was removed in #2313 because two
+assistant turns in a row violate message alternation (#2221).
+
+Instead, each successful delivery is recorded here, keyed by destination, and
+the gateway message handler folds any pending notices into the SYSTEM PROMPT of
+that chat's next interactive turn (alternation-safe, mirroring the auto-reset
+context-note precedent) and then drains the buffer. The net effect: the agent
+becomes aware of what its crons sent, without polluting the message array.
+
+Standalone and best-effort: callable from the in-process scheduler tick, and
+every failure is swallowed so it can never break delivery (which has already
+happened by the time we record) or a user turn.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import threading
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# Scheduler ticks and gateway message handling run in the same process; a
+# module-level lock plus atomic replace is enough to keep the store consistent.
+_LOCK = threading.Lock()
+
+# Cap entries per destination so a chat that goes unread for a long time can't
+# grow the store (or the next turn's system prompt) without bound.
+_MAX_PER_KEY = 20
+
+
+def _store_path(base_dir: Optional[Path] = None) -> Path:
+    if base_dir is not None:
+        base = Path(base_dir)
+    else:
+        from hermes_constants import get_hermes_home
+        base = get_hermes_home() / "cron"
+    return base / "pending_notices.json"
+
+
+def _key(platform: str, chat_id) -> str:
+    return f"{str(platform).lower()}:{chat_id}"
+
+
+def _load(path: Path) -> Dict[str, List[dict]]:
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+        return data if isinstance(data, dict) else {}
+    except FileNotFoundError:
+        return {}
+    except Exception as e:
+        logger.debug("pending notices: unreadable store %s (%s); starting empty", path, e)
+        return {}
+
+
+def _save(path: Path, data: Dict[str, List[dict]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".json.tmp")
+    tmp.write_text(json.dumps(data), encoding="utf-8")
+    tmp.replace(path)
+
+
+def record(
+    platform: str,
+    chat_id,
+    job_name: str,
+    text: str,
+    thread_id: Optional[str] = None,
+    base_dir: Optional[Path] = None,
+) -> bool:
+    """Buffer one delivered cron message for ``platform``/``chat_id``.
+
+    Returns True if stored, False on empty input or any error.
+    """
+    text = (text or "").strip()
+    if not text or chat_id in (None, ""):
+        return False
+    try:
+        with _LOCK:
+            path = _store_path(base_dir)
+            data = _load(path)
+            key = _key(platform, chat_id)
+            entries = data.get(key, [])
+            entries.append({
+                "ts": datetime.now().isoformat(timespec="seconds"),
+                "job_name": job_name or "",
+                "thread_id": thread_id,
+                "text": text,
+            })
+            data[key] = entries[-_MAX_PER_KEY:]
+            _save(path, data)
+        return True
+    except Exception as e:
+        logger.debug("pending notice record failed for %s:%s: %s", platform, chat_id, e)
+        return False
+
+
+def drain(
+    platform: str,
+    chat_id,
+    base_dir: Optional[Path] = None,
+) -> List[dict]:
+    """Return and clear all pending notices for ``platform``/``chat_id``.
+
+    Returns an empty list when there is nothing pending or on any error.
+    """
+    try:
+        with _LOCK:
+            path = _store_path(base_dir)
+            data = _load(path)
+            key = _key(platform, chat_id)
+            entries = data.pop(key, [])
+            if entries:
+                _save(path, data)
+        return entries
+    except Exception as e:
+        logger.debug("pending notice drain failed for %s:%s: %s", platform, chat_id, e)
+        return []
diff --git a/cron/scheduler.py b/cron/scheduler.py
index a51ade8efe651..9ccc9dfe18dcb 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -640,10 +640,18 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
     # Optionally wrap the content with a header/footer so the user knows this
     # is a cron delivery.  Wrapping is on by default; set cron.wrap_response: false
     # in config.yaml for clean output.
+    #
+    # notify_session (on by default; set cron.notify_session: false to disable)
+    # buffers each delivery so the chat's next interactive turn surfaces it in
+    # the system prompt — see cron/pending_notices.py.  This does NOT inject
+    # into message history (that broke alternation, #2313/#2221).
     wrap_response = True
+    notify_session = True
     try:
         user_cfg = load_config()
-        wrap_response = user_cfg.get("cron", {}).get("wrap_response", True)
+        cron_cfg = user_cfg.get("cron", {})
+        wrap_response = cron_cfg.get("wrap_response", True)
+        notify_session = cron_cfg.get("notify_session", True)
     except Exception:
         pass
 
@@ -665,6 +673,11 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
     media_files, cleaned_delivery_content = BasePlatformAdapter.extract_media(delivery_content)
     media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files)
 
+    # For the session notice we want the raw job output (no wrapper/footer),
+    # with MEDIA tags stripped so the buffered text stays clean.
+    _, notice_text = BasePlatformAdapter.extract_media(content)
+    notice_text = (notice_text or "").strip()
+
     try:
         config = load_gateway_config()
     except Exception as e:
@@ -771,6 +784,10 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 if adapter_ok:
                     logger.info("Job '%s': delivered to %s:%s via live adapter", job["id"], platform_name, chat_id)
                     delivered = True
+                    _record_session_notice(
+                        notify_session, platform_name, chat_id,
+                        notice_text, thread_id, job,
+                    )
             except Exception as e:
                 logger.warning(
                     "Job '%s': live adapter delivery to %s:%s failed (%s), falling back to standalone",
@@ -804,12 +821,45 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 continue
 
             logger.info("Job '%s': delivered to %s:%s", job["id"], platform_name, chat_id)
+            _record_session_notice(
+                notify_session, platform_name, chat_id,
+                notice_text, thread_id, job,
+            )
 
     if delivery_errors:
         return "; ".join(delivery_errors)
     return None
 
 
+def _record_session_notice(
+    enabled: bool,
+    platform_name: str,
+    chat_id: str,
+    text: str,
+    thread_id: Optional[str],
+    job: dict,
+) -> None:
+    """Buffer a delivered cron message for the chat's next interactive turn.
+
+    No-ops when disabled or empty. Best-effort: delivery has already
+    succeeded, so a buffering failure must never surface as a delivery error.
+    """
+    if not enabled or not text:
+        return
+    try:
+        from cron.pending_notices import record
+
+        record(
+            platform_name,
+            str(chat_id),
+            job.get("name", job.get("id", "")),
+            text,
+            thread_id=thread_id,
+        )
+    except Exception as e:
+        logger.debug("Job '%s': session notice not buffered (%s)", job.get("id"), e)
+
+
 _DEFAULT_SCRIPT_TIMEOUT = 120  # seconds
 # Backward-compatible module override used by tests and emergency monkeypatches.
 _SCRIPT_TIMEOUT = _DEFAULT_SCRIPT_TIMEOUT
diff --git a/gateway/run.py b/gateway/run.py
index 64eb8eb560e1c..bb2cb10ea3964 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -7149,6 +7149,45 @@ def _get_unauthorized_dm_behavior(self, platform: Optional[Platform]) -> str:
 
         return "pair"
 
+    def _build_cron_delivery_note(self, source) -> Optional[str]:
+        """Drain buffered cron deliveries for ``source`` and format a system note.
+
+        Returns None when nothing is pending. The returned text is prepended to
+        the system prompt (not the message array) so the agent gains awareness of
+        what its crons sent without breaking user/assistant alternation.
+        """
+        platform = getattr(source, "platform", None)
+        chat_id = getattr(source, "chat_id", None)
+        if platform is None or chat_id in (None, ""):
+            return None
+        platform_key = getattr(platform, "value", None) or str(platform)
+
+        from cron.pending_notices import drain
+
+        entries = drain(platform_key, str(chat_id))
+        if not entries:
+            return None
+
+        _PER_ENTRY_CAP = 1500
+        lines = [
+            "[System note: Since your last reply, the following automated cron "
+            "job(s) delivered messages to THIS chat. The user has already seen "
+            "them, but they are NOT in your message history. Use them as context; "
+            "do not re-send unless the user asks.",
+            "",
+        ]
+        for e in entries:
+            name = (e.get("job_name") or "cron").strip()
+            ts = (e.get("ts") or "").strip()
+            text = (e.get("text") or "").strip()
+            if len(text) > _PER_ENTRY_CAP:
+                text = text[:_PER_ENTRY_CAP].rstrip() + " […truncated]"
+            header = f"• {name}" + (f" (delivered {ts})" if ts else "")
+            lines.append(f"{header}:\n{text}")
+            lines.append("")
+        lines.append("]")
+        return "\n".join(lines)
+
     async def _deliver_platform_notice(self, source, content: str) -> None:
         """Deliver a setup/operational notice using platform-specific privacy rules."""
         adapter = self.adapters.get(source.platform)
@@ -8754,7 +8793,19 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
 
         # Build the context prompt to inject
         context_prompt = build_session_context_prompt(context, redact_pii=_redact_pii)
-        
+
+        # Surface any cron jobs that delivered to this chat since the last turn.
+        # Cron deliveries don't enter message history (assistant-role mirroring
+        # broke alternation, #2313/#2221), so we fold them into the SYSTEM PROMPT
+        # here — alternation-safe, like the auto-reset note below — and drain the
+        # buffer once consumed.  See cron/pending_notices.py.
+        try:
+            cron_note = self._build_cron_delivery_note(source)
+            if cron_note:
+                context_prompt = cron_note + "\n\n" + context_prompt
+        except Exception as e:
+            logger.debug("Cron delivery note injection skipped (non-fatal): %s", e)
+
         # If the previous session expired and was auto-reset, prepend a notice
         # so the agent knows this is a fresh conversation (not an intentional /reset).
         if getattr(session_entry, 'was_auto_reset', False):
diff --git a/tests/cron/test_pending_notices.py b/tests/cron/test_pending_notices.py
new file mode 100644
index 0000000000000..0cade03204ed1
--- /dev/null
+++ b/tests/cron/test_pending_notices.py
@@ -0,0 +1,61 @@
+"""Tests for cron/pending_notices.py — the push-side buffer for cron
+session-awareness (record on delivery, drain on next interactive turn)."""
+
+from cron.pending_notices import record, drain, _MAX_PER_KEY
+
+
+class TestRecordDrain:
+    def test_record_then_drain_roundtrip(self, tmp_path):
+        assert record("telegram", "123", "PR Watch", "found 3 issues", base_dir=tmp_path)
+        got = drain("telegram", "123", base_dir=tmp_path)
+        assert len(got) == 1
+        assert got[0]["job_name"] == "PR Watch"
+        assert got[0]["text"] == "found 3 issues"
+
+    def test_drain_clears_buffer(self, tmp_path):
+        record("telegram", "123", "j", "x", base_dir=tmp_path)
+        drain("telegram", "123", base_dir=tmp_path)
+        assert drain("telegram", "123", base_dir=tmp_path) == []
+
+    def test_multiple_entries_preserve_order(self, tmp_path):
+        record("telegram", "123", "j1", "first", base_dir=tmp_path)
+        record("telegram", "123", "j2", "second", base_dir=tmp_path)
+        got = drain("telegram", "123", base_dir=tmp_path)
+        assert [e["text"] for e in got] == ["first", "second"]
+
+    def test_keys_are_isolated_by_platform_and_chat(self, tmp_path):
+        record("telegram", "123", "j", "tg", base_dir=tmp_path)
+        record("sendblue", "123", "j", "sb", base_dir=tmp_path)
+        record("telegram", "999", "j", "other", base_dir=tmp_path)
+        assert [e["text"] for e in drain("telegram", "123", base_dir=tmp_path)] == ["tg"]
+        assert [e["text"] for e in drain("sendblue", "123", base_dir=tmp_path)] == ["sb"]
+        assert [e["text"] for e in drain("telegram", "999", base_dir=tmp_path)] == ["other"]
+
+    def test_platform_key_is_case_insensitive(self, tmp_path):
+        record("Telegram", "123", "j", "x", base_dir=tmp_path)
+        assert len(drain("telegram", "123", base_dir=tmp_path)) == 1
+
+    def test_empty_text_not_recorded(self, tmp_path):
+        assert record("telegram", "123", "j", "   ", base_dir=tmp_path) is False
+        assert drain("telegram", "123", base_dir=tmp_path) == []
+
+    def test_missing_chat_id_not_recorded(self, tmp_path):
+        assert record("telegram", None, "j", "x", base_dir=tmp_path) is False
+        assert record("telegram", "", "j", "x", base_dir=tmp_path) is False
+
+    def test_entries_capped_at_max(self, tmp_path):
+        for i in range(_MAX_PER_KEY + 5):
+            record("telegram", "123", "j", f"n{i}", base_dir=tmp_path)
+        got = drain("telegram", "123", base_dir=tmp_path)
+        assert len(got) == _MAX_PER_KEY
+        # oldest dropped, newest kept
+        assert got[-1]["text"] == f"n{_MAX_PER_KEY + 4}"
+        assert got[0]["text"] == "n5"
+
+    def test_drain_unknown_key_is_empty(self, tmp_path):
+        assert drain("telegram", "nope", base_dir=tmp_path) == []
+
+    def test_thread_id_preserved(self, tmp_path):
+        record("telegram", "123", "j", "x", thread_id="42", base_dir=tmp_path)
+        got = drain("telegram", "123", base_dir=tmp_path)
+        assert got[0]["thread_id"] == "42"
diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py
index 38da3fe408758..b09368fb29198 100644
--- a/tests/cron/test_scheduler.py
+++ b/tests/cron/test_scheduler.py
@@ -809,6 +809,56 @@ def test_no_mirror_to_session_call(self):
 
         mirror_mock.assert_not_called()
 
+    def test_records_session_notice_on_delivery(self):
+        """Successful delivery buffers a pending session notice (push side)."""
+        from gateway.config import Platform
+
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("cron.pending_notices.record") as rec_mock:
+            job = {
+                "id": "test-job",
+                "name": "PR Watch",
+                "deliver": "origin",
+                "origin": {"platform": "telegram", "chat_id": "123"},
+            }
+            _deliver_result(job, "Hello!")
+
+        rec_mock.assert_called_once()
+        args, kwargs = rec_mock.call_args
+        assert args[0] == "telegram"
+        assert args[1] == "123"
+        assert args[2] == "PR Watch"
+        assert args[3] == "Hello!"
+
+    def test_notify_session_false_skips_notice(self):
+        """cron.notify_session: false disables the push buffer."""
+        from gateway.config import Platform
+
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"notify_session": False}}), \
+             patch("cron.pending_notices.record") as rec_mock:
+            job = {
+                "id": "test-job",
+                "deliver": "origin",
+                "origin": {"platform": "telegram", "chat_id": "123"},
+            }
+            _deliver_result(job, "Hello!")
+
+        rec_mock.assert_not_called()
+
     def test_origin_delivery_preserves_thread_id(self):
         """Origin delivery should forward thread_id to the send helper."""
         from gateway.config import Platform
diff --git a/tests/gateway/test_cron_delivery_note.py b/tests/gateway/test_cron_delivery_note.py
new file mode 100644
index 0000000000000..2d44c2c9961c3
--- /dev/null
+++ b/tests/gateway/test_cron_delivery_note.py
@@ -0,0 +1,59 @@
+"""Tests for GatewayRunner._build_cron_delivery_note — the push-side injection
+that folds buffered cron deliveries into the next turn's system prompt."""
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from gateway.run import GatewayRunner
+
+
+def _src(platform="telegram", chat_id="123"):
+    return SimpleNamespace(platform=SimpleNamespace(value=platform), chat_id=chat_id)
+
+
+def _note(source):
+    # method doesn't use self, so a dummy self is fine
+    return GatewayRunner._build_cron_delivery_note(None, source)
+
+
+class TestBuildCronDeliveryNote:
+    def test_none_when_no_entries(self):
+        with patch("cron.pending_notices.drain", return_value=[]):
+            assert _note(_src()) is None
+
+    def test_drains_with_platform_value_and_chat_id(self):
+        with patch("cron.pending_notices.drain", return_value=[]) as d:
+            _note(_src(platform="telegram", chat_id="999"))
+        d.assert_called_once_with("telegram", "999")
+
+    def test_formats_entries_into_system_note(self):
+        entries = [
+            {"job_name": "PR Watch", "ts": "2026-06-01T16:00:00", "text": "found 3 issues"},
+            {"job_name": "Nutrition", "ts": "2026-06-01T21:30:00", "text": "score 82"},
+        ]
+        with patch("cron.pending_notices.drain", return_value=entries):
+            note = _note(_src())
+        assert note is not None
+        assert note.startswith("[System note:")
+        assert note.rstrip().endswith("]")
+        assert "PR Watch" in note and "found 3 issues" in note
+        assert "Nutrition" in note and "score 82" in note
+        assert "NOT in your message history" in note
+
+    def test_long_text_truncated(self):
+        entries = [{"job_name": "j", "ts": "", "text": "x" * 5000}]
+        with patch("cron.pending_notices.drain", return_value=entries):
+            note = _note(_src())
+        assert "[…truncated]" in note
+        assert len(note) < 5000
+
+    def test_none_when_chat_id_missing(self):
+        with patch("cron.pending_notices.drain", return_value=[]) as d:
+            assert _note(_src(chat_id=None)) is None
+        d.assert_not_called()
+
+    def test_plain_string_platform_supported(self):
+        src = SimpleNamespace(platform="sendblue", chat_id="123")
+        with patch("cron.pending_notices.drain", return_value=[]) as d:
+            _note(src)
+        d.assert_called_once_with("sendblue", "123")

From 0df080d5649dc13547c22cc509bf83831dcd3bd7 Mon Sep 17 00:00:00 2001
From: beardthelion <beardthelion@users.noreply.github.com>
Date: Mon, 1 Jun 2026 19:24:35 -0500
Subject: [PATCH 2/8] docs(cron): document cron.notify_session
 session-awareness option

---
 website/docs/developer-guide/cron-internals.md |  2 ++
 website/docs/user-guide/features/cron.md       | 14 ++++++++++++++
 2 files changed, 16 insertions(+)

diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md
index bad59645dbc61..0824d76b26b7f 100644
--- a/website/docs/developer-guide/cron-internals.md
+++ b/website/docs/developer-guide/cron-internals.md
@@ -196,6 +196,8 @@ The `[SILENT]` prefix in a cron response suppresses delivery entirely — useful
 
 Cron deliveries are NOT mirrored into gateway session conversation history. They exist only in the cron job's own session. This prevents message alternation violations in the target chat's conversation.
 
+When `cron.notify_session` is enabled (the default), deliveries are additionally buffered (`~/.hermes/cron/pending_notices.json`, keyed by `platform:chat_id`) and folded into the system prompt of the target chat's next turn as a `[System note: ...]` block, then drained. This gives the agent awareness of its own cron output while keeping the message history (and thus alternation) untouched.
+
 ## Recursion Guard
 
 Cron-run sessions have the `cronjob` toolset disabled. This prevents:
diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md
index cbefde68a9e46..638c9c8240cc3 100644
--- a/website/docs/user-guide/features/cron.md
+++ b/website/docs/user-guide/features/cron.md
@@ -327,6 +327,20 @@ cron:
   wrap_response: false
 ```
 
+### Session awareness
+
+Cron deliveries are sent straight to the platform and are not added to the chat's conversation history, so by default the agent has no record of what a scheduled job sent. When `cron.notify_session` is enabled (the default), each delivery is buffered and surfaced to the agent on that chat's next message, as a `[System note: ...]` block in the system prompt. This keeps the agent aware of what its own jobs delivered without writing to the message history (which would otherwise break message alternation).
+
+The note is consumed once: it appears on the next turn after a delivery, then clears.
+
+To turn this off and keep deliveries fire-and-forget:
+
+```yaml
+# ~/.hermes/config.yaml
+cron:
+  notify_session: false
+```
+
 ### Silent suppression
 
 If the agent's final response starts with `[SILENT]`, delivery is suppressed entirely. The output is still saved locally for audit (in `~/.hermes/cron/output/`), but no message is sent to the delivery target.

From de590de2e4259b1e704df477faa70329bf281450 Mon Sep 17 00:00:00 2001
From: beardthelion <beardthelion@users.noreply.github.com>
Date: Mon, 1 Jun 2026 22:28:19 -0500
Subject: [PATCH 3/8] feat(cron): gate buffered notices with an inject flag for
 button mode

Extend cron/pending_notices.py so a delivery can be held until the user
opts it into context, the groundwork for inline accept/dismiss buttons.

- record() now stamps each entry with a short id (new_notice_id, sized
  for Telegram's 64-byte callback_data) and an inject flag, and returns
  the id so a caller can mint the button before recording.
- drain() returns and clears only injectable entries, leaving held ones
  (inject=False) in place; entries predating the flag default to
  injectable, so auto-mode behavior is unchanged.
- mark_accepted() flips a held entry to injectable (accept button);
  dismiss() drops it (dismiss button).

run.py needs no change: the system-prompt fold stays mode-agnostic
because the inject decision is made at record/accept time.
---
 cron/pending_notices.py            | 106 ++++++++++++++++++++++++++---
 tests/cron/test_pending_notices.py |  75 +++++++++++++++++++-
 2 files changed, 171 insertions(+), 10 deletions(-)

diff --git a/cron/pending_notices.py b/cron/pending_notices.py
index d69b0e4d31750..7ac8206e0c922 100644
--- a/cron/pending_notices.py
+++ b/cron/pending_notices.py
@@ -20,9 +20,10 @@
 import json
 import logging
 import threading
+import uuid
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Union
 
 logger = logging.getLogger(__name__)
 
@@ -48,6 +49,15 @@ def _key(platform: str, chat_id) -> str:
     return f"{str(platform).lower()}:{chat_id}"
 
 
+def new_notice_id() -> str:
+    """Short unique id for a buffered notice.
+
+    Kept short so it fits inside a Telegram inline-button ``callback_data``
+    (64-byte cap) as ``cron:accept:<id>``.
+    """
+    return uuid.uuid4().hex[:8]
+
+
 def _load(path: Path) -> Dict[str, List[dict]]:
     try:
         data = json.loads(path.read_text(encoding="utf-8"))
@@ -72,15 +82,24 @@ def record(
     job_name: str,
     text: str,
     thread_id: Optional[str] = None,
+    notice_id: Optional[str] = None,
+    inject: bool = True,
     base_dir: Optional[Path] = None,
-) -> bool:
+) -> Union[str, bool]:
     """Buffer one delivered cron message for ``platform``/``chat_id``.
 
-    Returns True if stored, False on empty input or any error.
+    ``inject`` controls how the next interactive turn treats the entry: True
+    (auto mode) folds it into the system prompt right away; False (button mode)
+    holds it until the user accepts it via an inline button, at which point
+    :func:`mark_accepted` flips it. ``notice_id`` lets the caller pre-mint the id
+    it put on the button so the two stay in sync; one is generated otherwise.
+
+    Returns the notice id (str) on success, or False on empty input / error.
     """
     text = (text or "").strip()
     if not text or chat_id in (None, ""):
         return False
+    nid = notice_id or new_notice_id()
     try:
         with _LOCK:
             path = _store_path(base_dir)
@@ -88,14 +107,16 @@ def record(
             key = _key(platform, chat_id)
             entries = data.get(key, [])
             entries.append({
+                "id": nid,
                 "ts": datetime.now().isoformat(timespec="seconds"),
                 "job_name": job_name or "",
                 "thread_id": thread_id,
                 "text": text,
+                "inject": bool(inject),
             })
             data[key] = entries[-_MAX_PER_KEY:]
             _save(path, data)
-        return True
+        return nid
     except Exception as e:
         logger.debug("pending notice record failed for %s:%s: %s", platform, chat_id, e)
         return False
@@ -106,19 +127,86 @@ def drain(
     chat_id,
     base_dir: Optional[Path] = None,
 ) -> List[dict]:
-    """Return and clear all pending notices for ``platform``/``chat_id``.
+    """Return and clear the injectable notices for ``platform``/``chat_id``.
+
+    Only entries with ``inject`` truthy are returned and removed; entries still
+    awaiting an accept button (``inject`` False) are left in place for a later
+    :func:`mark_accepted`. Entries written before the inject flag existed default
+    to injectable, preserving the original drain-everything behavior.
 
-    Returns an empty list when there is nothing pending or on any error.
+    Returns an empty list when nothing is injectable or on any error.
     """
     try:
         with _LOCK:
             path = _store_path(base_dir)
             data = _load(path)
             key = _key(platform, chat_id)
-            entries = data.pop(key, [])
-            if entries:
+            entries = data.get(key, [])
+            if not entries:
+                return []
+            injectable = [e for e in entries if e.get("inject", True)]
+            remaining = [e for e in entries if not e.get("inject", True)]
+            if injectable:
+                if remaining:
+                    data[key] = remaining
+                else:
+                    data.pop(key, None)
                 _save(path, data)
-        return entries
+            return injectable
     except Exception as e:
         logger.debug("pending notice drain failed for %s:%s: %s", platform, chat_id, e)
         return []
+
+
+def mark_accepted(
+    platform: str,
+    chat_id,
+    notice_id: str,
+    base_dir: Optional[Path] = None,
+) -> bool:
+    """Flip a held (button-mode) notice to injectable.
+
+    Called when the user taps the accept button. Returns True if a matching
+    entry was found, False otherwise (already drained, dismissed, or unknown id).
+    """
+    try:
+        with _LOCK:
+            path = _store_path(base_dir)
+            data = _load(path)
+            key = _key(platform, chat_id)
+            for e in data.get(key, []):
+                if e.get("id") == notice_id:
+                    e["inject"] = True
+                    _save(path, data)
+                    return True
+        return False
+    except Exception as e:
+        logger.debug("pending notice mark_accepted failed for %s:%s: %s", platform, chat_id, e)
+        return False
+
+
+def dismiss(
+    platform: str,
+    chat_id,
+    notice_id: str,
+    base_dir: Optional[Path] = None,
+) -> bool:
+    """Drop a held notice the user declined. Returns True if one was removed."""
+    try:
+        with _LOCK:
+            path = _store_path(base_dir)
+            data = _load(path)
+            key = _key(platform, chat_id)
+            entries = data.get(key, [])
+            kept = [e for e in entries if e.get("id") != notice_id]
+            if len(kept) == len(entries):
+                return False
+            if kept:
+                data[key] = kept
+            else:
+                data.pop(key, None)
+            _save(path, data)
+            return True
+    except Exception as e:
+        logger.debug("pending notice dismiss failed for %s:%s: %s", platform, chat_id, e)
+        return False
diff --git a/tests/cron/test_pending_notices.py b/tests/cron/test_pending_notices.py
index 0cade03204ed1..79fc2afdcc228 100644
--- a/tests/cron/test_pending_notices.py
+++ b/tests/cron/test_pending_notices.py
@@ -1,7 +1,14 @@
 """Tests for cron/pending_notices.py — the push-side buffer for cron
 session-awareness (record on delivery, drain on next interactive turn)."""
 
-from cron.pending_notices import record, drain, _MAX_PER_KEY
+from cron.pending_notices import (
+    record,
+    drain,
+    mark_accepted,
+    dismiss,
+    new_notice_id,
+    _MAX_PER_KEY,
+)
 
 
 class TestRecordDrain:
@@ -59,3 +66,69 @@ def test_thread_id_preserved(self, tmp_path):
         record("telegram", "123", "j", "x", thread_id="42", base_dir=tmp_path)
         got = drain("telegram", "123", base_dir=tmp_path)
         assert got[0]["thread_id"] == "42"
+
+
+class TestInjectGating:
+    """Button mode buffers entries with inject=False until the user accepts.
+
+    drain() returns only injectable entries and leaves the rest, so the
+    gateway's system-prompt fold (run.py) stays mode-agnostic.
+    """
+
+    def test_record_assigns_id_and_defaults_inject_true(self, tmp_path):
+        nid = record("telegram", "123", "j", "x", base_dir=tmp_path)
+        assert isinstance(nid, str) and nid
+        got = drain("telegram", "123", base_dir=tmp_path)
+        assert got[0]["id"] == nid
+        assert got[0]["inject"] is True
+
+    def test_record_with_explicit_id(self, tmp_path):
+        nid = record("telegram", "123", "j", "x", notice_id="abc123", base_dir=tmp_path)
+        assert nid == "abc123"
+        assert drain("telegram", "123", base_dir=tmp_path)[0]["id"] == "abc123"
+
+    def test_inject_false_held_until_accepted(self, tmp_path):
+        record("telegram", "123", "j", "held", notice_id="n1", inject=False, base_dir=tmp_path)
+        # not injected while pending, but not lost
+        assert drain("telegram", "123", base_dir=tmp_path) == []
+        assert mark_accepted("telegram", "123", "n1", base_dir=tmp_path) is True
+        got = drain("telegram", "123", base_dir=tmp_path)
+        assert [e["text"] for e in got] == ["held"]
+
+    def test_mark_accepted_unknown_id(self, tmp_path):
+        record("telegram", "123", "j", "x", notice_id="n1", inject=False, base_dir=tmp_path)
+        assert mark_accepted("telegram", "123", "nope", base_dir=tmp_path) is False
+
+    def test_dismiss_removes_entry(self, tmp_path):
+        record("telegram", "123", "j", "x", notice_id="n1", inject=False, base_dir=tmp_path)
+        assert dismiss("telegram", "123", "n1", base_dir=tmp_path) is True
+        assert drain("telegram", "123", base_dir=tmp_path) == []
+        # gone for good
+        assert mark_accepted("telegram", "123", "n1", base_dir=tmp_path) is False
+
+    def test_dismiss_unknown_id(self, tmp_path):
+        assert dismiss("telegram", "123", "nope", base_dir=tmp_path) is False
+
+    def test_drain_returns_injectable_leaves_pending(self, tmp_path):
+        record("telegram", "123", "j", "auto", notice_id="a", base_dir=tmp_path)
+        record("telegram", "123", "j", "held", notice_id="b", inject=False, base_dir=tmp_path)
+        assert [e["text"] for e in drain("telegram", "123", base_dir=tmp_path)] == ["auto"]
+        # the pending one survived the drain and can still be accepted later
+        assert mark_accepted("telegram", "123", "b", base_dir=tmp_path) is True
+        assert [e["text"] for e in drain("telegram", "123", base_dir=tmp_path)] == ["held"]
+
+    def test_legacy_entry_without_inject_is_drained(self, tmp_path):
+        # entries written by the pre-button record() have no inject/id field
+        import json
+        from cron.pending_notices import _store_path
+        p = _store_path(tmp_path)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(json.dumps(
+            {"telegram:123": [{"ts": "t", "job_name": "j", "thread_id": None, "text": "old"}]}
+        ))
+        assert [e["text"] for e in drain("telegram", "123", base_dir=tmp_path)] == ["old"]
+
+    def test_new_notice_id_unique_and_short(self):
+        ids = {new_notice_id() for _ in range(50)}
+        assert len(ids) == 50
+        assert all(isinstance(i, str) and 0 < len(i) <= 16 for i in ids)

From f419350cb669cbde7dafeb039ead04471c34f2e3 Mon Sep 17 00:00:00 2001
From: beardthelion <beardthelion@users.noreply.github.com>
Date: Mon, 1 Jun 2026 22:29:28 -0500
Subject: [PATCH 4/8] feat(cron): add notify_session mode normalizer
 (off/auto/button)

normalize_notify_mode() maps the cron.notify_session config value to one
of three modes while preserving the original boolean knob: True/on-ish
becomes auto, False/None/off-ish becomes off, "button" selects inline
accept/dismiss buttons. An unrecognized but present value stays on (auto),
matching the prior "any truthy value enabled it" behavior. Pure function,
unit-tested alongside the buffer.
---
 cron/pending_notices.py            | 20 +++++++++++++++++
 tests/cron/test_pending_notices.py | 36 ++++++++++++++++++++++++++++++
 2 files changed, 56 insertions(+)

diff --git a/cron/pending_notices.py b/cron/pending_notices.py
index 7ac8206e0c922..75510fc2865aa 100644
--- a/cron/pending_notices.py
+++ b/cron/pending_notices.py
@@ -58,6 +58,26 @@ def new_notice_id() -> str:
     return uuid.uuid4().hex[:8]
 
 
+def normalize_notify_mode(value) -> str:
+    """Normalize a ``cron.notify_session`` config value to off / auto / button.
+
+    Back-compatible with the original boolean knob: True (or any recognized
+    on-ish value) means auto, False / None / off-ish means off. "button" opts
+    into inline accept/dismiss buttons. An unrecognized but present value stays
+    on (auto), matching the old "any truthy config value enabled it" behavior.
+    """
+    if value is True:
+        return "auto"
+    if value is False or value is None:
+        return "off"
+    s = str(value).strip().lower()
+    if s in {"button", "buttons"}:
+        return "button"
+    if s in {"off", "no", "false", "0", "disabled", "none", ""}:
+        return "off"
+    return "auto"
+
+
 def _load(path: Path) -> Dict[str, List[dict]]:
     try:
         data = json.loads(path.read_text(encoding="utf-8"))
diff --git a/tests/cron/test_pending_notices.py b/tests/cron/test_pending_notices.py
index 79fc2afdcc228..0846e2ebd1e39 100644
--- a/tests/cron/test_pending_notices.py
+++ b/tests/cron/test_pending_notices.py
@@ -7,6 +7,7 @@
     mark_accepted,
     dismiss,
     new_notice_id,
+    normalize_notify_mode,
     _MAX_PER_KEY,
 )
 
@@ -132,3 +133,38 @@ def test_new_notice_id_unique_and_short(self):
         ids = {new_notice_id() for _ in range(50)}
         assert len(ids) == 50
         assert all(isinstance(i, str) and 0 < len(i) <= 16 for i in ids)
+
+
+class TestNotifyMode:
+    """cron.notify_session normalizes to off / auto / button, preserving the
+    legacy bool semantics (True == on == auto, False/None == off)."""
+
+    def test_true_is_auto(self):
+        assert normalize_notify_mode(True) == "auto"
+
+    def test_false_is_off(self):
+        assert normalize_notify_mode(False) == "off"
+
+    def test_none_is_off(self):
+        assert normalize_notify_mode(None) == "off"
+
+    def test_button_aliases(self):
+        assert normalize_notify_mode("button") == "button"
+        assert normalize_notify_mode("buttons") == "button"
+
+    def test_off_aliases(self):
+        for v in ("off", "no", "false", "disabled", ""):
+            assert normalize_notify_mode(v) == "off"
+
+    def test_auto_aliases(self):
+        for v in ("auto", "on", "yes", "true"):
+            assert normalize_notify_mode(v) == "auto"
+
+    def test_case_insensitive(self):
+        assert normalize_notify_mode("Button") == "button"
+        assert normalize_notify_mode("OFF") == "off"
+
+    def test_unknown_present_value_defaults_to_auto(self):
+        # a non-empty but unrecognized value stays on (matches the old
+        # "any truthy config value enabled it" behavior)
+        assert normalize_notify_mode("wat") == "auto"

From 865b7d56ade0de57815a6b038de867abe9c6c96e Mon Sep 17 00:00:00 2001
From: beardthelion <beardthelion@users.noreply.github.com>
Date: Mon, 1 Jun 2026 22:33:53 -0500
Subject: [PATCH 5/8] feat(telegram): add send_cron_notice accept/dismiss
 buttons

Button mode for cron deliveries. The cron message is sent normally, then
send_cron_notice posts a short prompt with two inline buttons whose
callback_data is cron:accept:<id> / cron:dismiss:<id>. The notice id is
the on-disk buffer key, so unlike the exec-approval in-memory counter the
buttons keep working after a gateway restart.

A SUPPORTS_CRON_BUTTONS capability flag (False on the base adapter, True
on Telegram) lets the scheduler fall back to automatic injection on
platforms without inline keyboards, so cron awareness is never lost.
---
 gateway/platforms/base.py                   |  7 ++
 gateway/platforms/telegram.py               | 60 ++++++++++++++
 tests/gateway/test_telegram_cron_buttons.py | 87 +++++++++++++++++++++
 3 files changed, 154 insertions(+)
 create mode 100644 tests/gateway/test_telegram_cron_buttons.py

diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py
index 761eba90e29d9..f8968cb7e70fa 100644
--- a/gateway/platforms/base.py
+++ b/gateway/platforms/base.py
@@ -2033,6 +2033,13 @@ async def send(
     # property) so the stream consumer knows not to short-circuit.
     REQUIRES_EDIT_FINALIZE: bool = False
 
+    # Default: the platform cannot render the cron-delivery accept/dismiss
+    # buttons used by ``cron.notify_session: button``.  Adapters with inline
+    # keyboards (e.g. Telegram) override this to True and implement
+    # ``send_cron_notice``; on platforms that leave it False, the scheduler
+    # falls back to automatic injection so cron awareness is never lost.
+    SUPPORTS_CRON_BUTTONS: bool = False
+
     async def create_handoff_thread(
         self,
         parent_chat_id: str,
diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py
index 14820c0fe7c57..c56997e0960f2 100644
--- a/gateway/platforms/telegram.py
+++ b/gateway/platforms/telegram.py
@@ -357,6 +357,10 @@ class TelegramAdapter(BasePlatformAdapter):
     # Fixes #25710.
     REQUIRES_EDIT_FINALIZE: bool = True
 
+    # Telegram renders the cron-delivery accept/dismiss buttons used by
+    # ``cron.notify_session: button`` (see send_cron_notice).
+    SUPPORTS_CRON_BUTTONS: bool = True
+
     # Adaptive text-batch ingress: short messages need a tighter delay so the
     # first token reaches the agent fast.  Numbers tuned for "feels instant":
     # ≤320 codepoints (one short paragraph) settles in ~180ms; ≤1024
@@ -2694,6 +2698,62 @@ async def send_exec_approval(
             logger.warning("[%s] send_exec_approval failed: %s", self.name, e)
             return SendResult(success=False, error=str(e))
 
+    async def send_cron_notice(
+        self,
+        chat_id: str,
+        notice_id: str,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Send an Add-to-context / Dismiss prompt beneath a cron delivery.
+
+        Button mode (``cron.notify_session: button``) delivers the cron message
+        normally and then this short prompt.  Accept folds the buffered delivery
+        into the chat's next interactive turn (pending_notices.mark_accepted);
+        dismiss drops it (pending_notices.dismiss).  No in-memory state is kept:
+        the on-disk notice buffer is the source of truth, keyed by ``notice_id``,
+        so the buttons keep working after a gateway restart (unlike the
+        exec-approval in-memory counter).
+        """
+        if not self._bot:
+            return SendResult(success=False, error="Not connected")
+
+        try:
+            text = (
+                "📥 <b>Cron delivery above</b>\n\n"
+                "It is not in my chat memory. Add it to this conversation's context?"
+            )
+            thread_id = self._metadata_thread_id(metadata)
+            keyboard = InlineKeyboardMarkup([
+                [
+                    InlineKeyboardButton("📥 Add to context", callback_data=f"cron:accept:{notice_id}"),
+                    InlineKeyboardButton("✕ Dismiss", callback_data=f"cron:dismiss:{notice_id}"),
+                ],
+            ])
+            kwargs: Dict[str, Any] = {
+                "chat_id": int(chat_id),
+                "text": text,
+                "parse_mode": ParseMode.HTML,
+                "reply_markup": keyboard,
+                **self._link_preview_kwargs(),
+            }
+            reply_to_id = self._reply_to_message_id_for_send(None, metadata, reply_to_mode=self._reply_to_mode)
+            kwargs["reply_to_message_id"] = reply_to_id
+            kwargs.update(
+                self._thread_kwargs_for_send(
+                    chat_id,
+                    thread_id,
+                    metadata,
+                    reply_to_message_id=reply_to_id,
+                    reply_to_mode=self._reply_to_mode,
+                )
+            )
+
+            msg = await self._send_message_with_thread_fallback(**kwargs)
+            return SendResult(success=True, message_id=str(msg.message_id))
+        except Exception as e:
+            logger.warning("[%s] send_cron_notice failed: %s", self.name, e)
+            return SendResult(success=False, error=str(e))
+
     async def send_slash_confirm(
         self, chat_id: str, title: str, message: str, session_key: str,
         confirm_id: str, metadata: Optional[Dict[str, Any]] = None,
diff --git a/tests/gateway/test_telegram_cron_buttons.py b/tests/gateway/test_telegram_cron_buttons.py
new file mode 100644
index 0000000000000..0bf8b70eb3553
--- /dev/null
+++ b/tests/gateway/test_telegram_cron_buttons.py
@@ -0,0 +1,87 @@
+"""Tests for Telegram cron-delivery accept/dismiss buttons (button mode)."""
+
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+_repo = str(Path(__file__).resolve().parents[2])
+if _repo not in sys.path:
+    sys.path.insert(0, _repo)
+
+
+def _ensure_telegram_mock():
+    if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
+        return
+    mod = MagicMock()
+    mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
+    mod.constants.ParseMode.MARKDOWN = "Markdown"
+    mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
+    mod.constants.ParseMode.HTML = "HTML"
+    mod.constants.ChatType.PRIVATE = "private"
+    mod.constants.ChatType.GROUP = "group"
+    mod.constants.ChatType.SUPERGROUP = "supergroup"
+    mod.constants.ChatType.CHANNEL = "channel"
+    mod.error.NetworkError = type("NetworkError", (OSError,), {})
+    mod.error.TimedOut = type("TimedOut", (OSError,), {})
+    mod.error.BadRequest = type("BadRequest", (Exception,), {})
+    for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
+        sys.modules.setdefault(name, mod)
+    sys.modules.setdefault("telegram.error", mod.error)
+
+
+_ensure_telegram_mock()
+
+from gateway.platforms import telegram as tg
+from gateway.platforms.telegram import TelegramAdapter
+from gateway.platforms.base import BasePlatformAdapter
+from gateway.config import PlatformConfig
+
+
+def _make_adapter():
+    config = PlatformConfig(enabled=True, token="test-token", extra={})
+    adapter = TelegramAdapter(config)
+    adapter._bot = AsyncMock()
+    adapter._app = MagicMock()
+    return adapter
+
+
+class TestTelegramCronNotice:
+    @pytest.mark.asyncio
+    async def test_sends_accept_dismiss_buttons(self):
+        adapter = _make_adapter()
+        mock_msg = MagicMock()
+        mock_msg.message_id = 77
+        adapter._bot.send_message = AsyncMock(return_value=mock_msg)
+        tg.InlineKeyboardButton.reset_mock()
+
+        result = await adapter.send_cron_notice(chat_id="12345", notice_id="ab12cd34")
+
+        assert result.success is True
+        assert result.message_id == "77"
+        adapter._bot.send_message.assert_called_once()
+        kwargs = adapter._bot.send_message.call_args[1]
+        assert kwargs["chat_id"] == 12345
+        assert kwargs["reply_markup"] is not None
+
+        callback_data = [
+            c.kwargs.get("callback_data") for c in tg.InlineKeyboardButton.call_args_list
+        ]
+        assert "cron:accept:ab12cd34" in callback_data
+        assert "cron:dismiss:ab12cd34" in callback_data
+
+    @pytest.mark.asyncio
+    async def test_not_connected_returns_failure(self):
+        adapter = _make_adapter()
+        adapter._bot = None
+        result = await adapter.send_cron_notice(chat_id="12345", notice_id="x")
+        assert result.success is False
+
+
+class TestCronButtonCapability:
+    def test_base_adapter_does_not_support_cron_buttons(self):
+        assert BasePlatformAdapter.SUPPORTS_CRON_BUTTONS is False
+
+    def test_telegram_supports_cron_buttons(self):
+        assert TelegramAdapter.SUPPORTS_CRON_BUTTONS is True

From 15e4a5c3c36a4526416b1f0e4191c2524a3cfae9 Mon Sep 17 00:00:00 2001
From: beardthelion <beardthelion@users.noreply.github.com>
Date: Mon, 1 Jun 2026 23:00:33 -0500
Subject: [PATCH 6/8] feat(telegram): route cron:accept/dismiss button
 callbacks

Add a cron: branch to _handle_callback_query mirroring the ea: exec-
approval flow: authorize the caller, then accept flips the buffered notice
to injectable (pending_notices.mark_accepted) and dismiss drops it
(pending_notices.dismiss), keyed by platform:chat_id from the query. The
button message is edited to show the outcome and its keyboard removed.
Unauthorized taps never touch the buffer.
---
 gateway/platforms/telegram.py               | 46 ++++++++++++++++
 tests/gateway/test_telegram_cron_buttons.py | 59 +++++++++++++++++++++
 2 files changed, 105 insertions(+)

diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py
index c56997e0960f2..285314dff60d9 100644
--- a/gateway/platforms/telegram.py
+++ b/gateway/platforms/telegram.py
@@ -3370,6 +3370,52 @@ async def _handle_callback_query(
                     self.resume_typing_for_chat(str(query_chat_id))
             return
 
+        # --- Cron delivery accept/dismiss callbacks (cron:choice:id) ---
+        if data.startswith("cron:"):
+            parts = data.split(":", 2)
+            if len(parts) == 3:
+                choice = parts[1]  # accept, dismiss
+                notice_id = parts[2]
+
+                # Only authorized users may resolve cron notices.
+                caller_id = str(getattr(query.from_user, "id", ""))
+                if not self._is_callback_user_authorized(
+                    caller_id,
+                    chat_id=query_chat_id,
+                    chat_type=str(query_chat_type) if query_chat_type is not None else None,
+                    thread_id=str(query_thread_id) if query_thread_id is not None else None,
+                    user_name=query_user_name,
+                ):
+                    await query.answer(text="⛔ You are not authorized.")
+                    return
+
+                # The on-disk notice buffer is the source of truth (keyed by
+                # platform:chat_id), so this survives a gateway restart.
+                from cron.pending_notices import dismiss as dismiss_notice
+                from cron.pending_notices import mark_accepted
+
+                chat_key = str(query_chat_id)
+                if choice == "accept":
+                    ok = mark_accepted("telegram", chat_key, notice_id)
+                    label = "📥 Added to context" if ok else "Already resolved"
+                elif choice == "dismiss":
+                    ok = dismiss_notice("telegram", chat_key, notice_id)
+                    label = "✕ Dismissed" if ok else "Already resolved"
+                else:
+                    await query.answer(text="Invalid action.")
+                    return
+
+                await query.answer(text=label)
+                try:
+                    await query.edit_message_text(
+                        text=self.format_message(label),
+                        parse_mode=ParseMode.MARKDOWN_V2,
+                        reply_markup=None,
+                    )
+                except Exception:
+                    pass  # non-fatal if the edit fails
+            return
+
         # --- Slash-confirm callbacks (sc:choice:confirm_id) ---
         if data.startswith("sc:"):
             parts = data.split(":", 2)
diff --git a/tests/gateway/test_telegram_cron_buttons.py b/tests/gateway/test_telegram_cron_buttons.py
index 0bf8b70eb3553..43e3cf8a10ea1 100644
--- a/tests/gateway/test_telegram_cron_buttons.py
+++ b/tests/gateway/test_telegram_cron_buttons.py
@@ -2,6 +2,7 @@
 
 import sys
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -85,3 +86,61 @@ def test_base_adapter_does_not_support_cron_buttons(self):
 
     def test_telegram_supports_cron_buttons(self):
         assert TelegramAdapter.SUPPORTS_CRON_BUTTONS is True
+
+
+def _make_callback_update(data, user_id=999, chat_id=12345):
+    query = MagicMock()
+    query.data = data
+    query.from_user = SimpleNamespace(id=user_id, first_name="Beardy")
+    query.message = SimpleNamespace(
+        chat_id=chat_id,
+        chat=SimpleNamespace(type="private"),
+        message_thread_id=None,
+    )
+    query.answer = AsyncMock()
+    query.edit_message_text = AsyncMock()
+    return SimpleNamespace(callback_query=query), query
+
+
+class TestCronCallback:
+    @pytest.mark.asyncio
+    async def test_accept_marks_notice_accepted(self, monkeypatch):
+        adapter = _make_adapter()
+        monkeypatch.setattr(adapter, "_is_callback_user_authorized", lambda *a, **k: True)
+        import cron.pending_notices as pn
+        accept_spy = MagicMock(return_value=True)
+        monkeypatch.setattr(pn, "mark_accepted", accept_spy)
+
+        update, query = _make_callback_update("cron:accept:ab12cd34", chat_id=12345)
+        await adapter._handle_callback_query(update, None)
+
+        accept_spy.assert_called_once_with("telegram", "12345", "ab12cd34")
+        query.edit_message_text.assert_awaited()
+        assert query.edit_message_text.call_args.kwargs.get("reply_markup") is None
+
+    @pytest.mark.asyncio
+    async def test_dismiss_drops_notice(self, monkeypatch):
+        adapter = _make_adapter()
+        monkeypatch.setattr(adapter, "_is_callback_user_authorized", lambda *a, **k: True)
+        import cron.pending_notices as pn
+        dismiss_spy = MagicMock(return_value=True)
+        monkeypatch.setattr(pn, "dismiss", dismiss_spy)
+
+        update, query = _make_callback_update("cron:dismiss:zz99", chat_id=-100777)
+        await adapter._handle_callback_query(update, None)
+
+        dismiss_spy.assert_called_once_with("telegram", "-100777", "zz99")
+
+    @pytest.mark.asyncio
+    async def test_unauthorized_does_not_touch_buffer(self, monkeypatch):
+        adapter = _make_adapter()
+        monkeypatch.setattr(adapter, "_is_callback_user_authorized", lambda *a, **k: False)
+        import cron.pending_notices as pn
+        accept_spy = MagicMock(return_value=True)
+        monkeypatch.setattr(pn, "mark_accepted", accept_spy)
+
+        update, query = _make_callback_update("cron:accept:ab12cd34")
+        await adapter._handle_callback_query(update, None)
+
+        accept_spy.assert_not_called()
+        query.answer.assert_awaited()

From a1c05f8258f72f2084c2d3387a73f2b20c1e9b7a Mon Sep 17 00:00:00 2001
From: beardthelion <beardthelion@users.noreply.github.com>
Date: Mon, 1 Jun 2026 23:05:03 -0500
Subject: [PATCH 7/8] feat(cron): wire scheduler delivery to button mode

_deliver_result now reads cron.notify_session as a three-way mode
(normalize_notify_mode) and threads it into _record_session_notice.

In button mode, when delivery used a live adapter that supports inline
buttons, the notice is buffered as held (inject=False) and an
accept/dismiss prompt is sent via adapter.send_cron_notice; if that send
fails the entry is auto-injected so awareness is never lost. Auto mode and
platforms without button support buffer as injectable, unchanged.
---
 cron/scheduler.py            | 75 +++++++++++++++++++++++++++++-------
 tests/cron/test_scheduler.py | 72 +++++++++++++++++++++++++++++++++-
 2 files changed, 132 insertions(+), 15 deletions(-)

diff --git a/cron/scheduler.py b/cron/scheduler.py
index 9ccc9dfe18dcb..d94968a4f952b 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -646,12 +646,13 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
     # the system prompt — see cron/pending_notices.py.  This does NOT inject
     # into message history (that broke alternation, #2313/#2221).
     wrap_response = True
-    notify_session = True
+    notify_mode = "auto"
     try:
+        from cron.pending_notices import normalize_notify_mode
         user_cfg = load_config()
         cron_cfg = user_cfg.get("cron", {})
         wrap_response = cron_cfg.get("wrap_response", True)
-        notify_session = cron_cfg.get("notify_session", True)
+        notify_mode = normalize_notify_mode(cron_cfg.get("notify_session", True))
     except Exception:
         pass
 
@@ -785,8 +786,9 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                     logger.info("Job '%s': delivered to %s:%s via live adapter", job["id"], platform_name, chat_id)
                     delivered = True
                     _record_session_notice(
-                        notify_session, platform_name, chat_id,
+                        notify_mode, platform_name, chat_id,
                         notice_text, thread_id, job,
+                        adapter=runtime_adapter, loop=loop, send_metadata=send_metadata,
                     )
             except Exception as e:
                 logger.warning(
@@ -822,7 +824,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
 
             logger.info("Job '%s': delivered to %s:%s", job["id"], platform_name, chat_id)
             _record_session_notice(
-                notify_session, platform_name, chat_id,
+                notify_mode, platform_name, chat_id,
                 notice_text, thread_id, job,
             )
 
@@ -832,30 +834,75 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
 
 
 def _record_session_notice(
-    enabled: bool,
+    mode: str,
     platform_name: str,
     chat_id: str,
     text: str,
     thread_id: Optional[str],
     job: dict,
+    adapter=None,
+    loop=None,
+    send_metadata=None,
 ) -> None:
     """Buffer a delivered cron message for the chat's next interactive turn.
 
-    No-ops when disabled or empty. Best-effort: delivery has already
-    succeeded, so a buffering failure must never surface as a delivery error.
+    ``mode`` is the normalized cron.notify_session value:
+
+      * ``off``    -> no-op.
+      * ``auto``   -> buffer as injectable; the next turn folds it into the
+        system prompt automatically.
+      * ``button`` -> on a live adapter that supports inline buttons, buffer the
+        entry as held (inject=False) and send an accept/dismiss prompt; the user
+        decides whether it reaches context. Platforms without button support (or
+        the standalone no-adapter path) fall back to ``auto`` so awareness is
+        never lost.
+
+    No-ops when off or empty. Best-effort: delivery has already succeeded, so a
+    buffering failure must never surface as a delivery error.
     """
-    if not enabled or not text:
+    text = (text or "").strip()
+    if mode == "off" or not text:
         return
     try:
-        from cron.pending_notices import record
+        from cron.pending_notices import new_notice_id, record
+
+        job_label = job.get("name", job.get("id", ""))
+        use_buttons = (
+            mode == "button"
+            and adapter is not None
+            and loop is not None
+            and getattr(adapter, "SUPPORTS_CRON_BUTTONS", False)
+        )
 
+        if not use_buttons:
+            record(platform_name, str(chat_id), job_label, text, thread_id=thread_id, inject=True)
+            return
+
+        # Button mode: hold the entry until the user accepts it, and send the
+        # accept/dismiss prompt beneath the delivery.
+        notice_id = new_notice_id()
         record(
-            platform_name,
-            str(chat_id),
-            job.get("name", job.get("id", "")),
-            text,
-            thread_id=thread_id,
+            platform_name, str(chat_id), job_label, text,
+            thread_id=thread_id, notice_id=notice_id, inject=False,
         )
+        try:
+            from agent.async_utils import safe_schedule_threadsafe
+
+            future = safe_schedule_threadsafe(
+                adapter.send_cron_notice(str(chat_id), notice_id, metadata=send_metadata),
+                loop,
+            )
+            if future is not None:
+                future.result(timeout=30)
+        except Exception as e:
+            # The prompt could not be sent; don't strand the notice. Make it
+            # injectable so the next turn still surfaces it (auto fallback).
+            logger.debug(
+                "Job '%s': cron notice button send failed (%s); auto-injecting",
+                job.get("id"), e,
+            )
+            from cron.pending_notices import mark_accepted
+            mark_accepted(platform_name, str(chat_id), notice_id)
     except Exception as e:
         logger.debug("Job '%s': session notice not buffered (%s)", job.get("id"), e)
 
diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py
index b09368fb29198..3817c836d42ab 100644
--- a/tests/cron/test_scheduler.py
+++ b/tests/cron/test_scheduler.py
@@ -3,11 +3,12 @@
 import json
 import logging
 import os
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch, MagicMock
 
 import pytest
 
-from cron.scheduler import _resolve_origin, _resolve_delivery_target, _deliver_result, _send_media_via_adapter, run_job, SILENT_MARKER, _build_job_prompt
+from cron.scheduler import _resolve_origin, _resolve_delivery_target, _deliver_result, _send_media_via_adapter, run_job, SILENT_MARKER, _build_job_prompt, _record_session_notice
 from tools.env_passthrough import clear_env_passthrough
 from tools.credential_files import clear_credential_files
 
@@ -887,6 +888,75 @@ def test_origin_delivery_preserves_thread_id(self):
         assert send_mock.call_args.kwargs["thread_id"] == "17585"
 
 
+class TestRecordSessionNoticeButton:
+    """_record_session_notice: auto records injectable, off skips, button holds
+    the entry (inject=False) and sends accept/dismiss buttons when the adapter
+    supports them, falling back to auto otherwise."""
+
+    def _job(self):
+        return {"id": "j1", "name": "PR Watch"}
+
+    def test_auto_mode_records_injectable(self, monkeypatch):
+        import cron.pending_notices as pn
+        rec = MagicMock(return_value="nid")
+        monkeypatch.setattr(pn, "record", rec)
+        _record_session_notice("auto", "telegram", "123", "Hello", None, self._job())
+        rec.assert_called_once()
+        assert rec.call_args.kwargs.get("inject") is True
+
+    def test_off_mode_records_nothing(self, monkeypatch):
+        import cron.pending_notices as pn
+        rec = MagicMock(return_value="nid")
+        monkeypatch.setattr(pn, "record", rec)
+        _record_session_notice("off", "telegram", "123", "Hello", None, self._job())
+        rec.assert_not_called()
+
+    def test_empty_text_records_nothing(self, monkeypatch):
+        import cron.pending_notices as pn
+        rec = MagicMock(return_value="nid")
+        monkeypatch.setattr(pn, "record", rec)
+        _record_session_notice("auto", "telegram", "123", "   ", None, self._job())
+        rec.assert_not_called()
+
+    def test_button_mode_with_support_holds_and_sends(self, monkeypatch):
+        import cron.pending_notices as pn
+        import agent.async_utils as au
+        rec = MagicMock(return_value="fixedid")
+        monkeypatch.setattr(pn, "record", rec)
+        monkeypatch.setattr(pn, "new_notice_id", lambda: "fixedid")
+        fake_future = MagicMock()
+        fake_future.result.return_value = SimpleNamespace(success=True)
+        monkeypatch.setattr(au, "safe_schedule_threadsafe", lambda coro, loop: fake_future)
+
+        adapter = MagicMock()
+        adapter.SUPPORTS_CRON_BUTTONS = True
+        loop = MagicMock()
+        _record_session_notice(
+            "button", "telegram", "123", "Hello", None, self._job(),
+            adapter=adapter, loop=loop, send_metadata={"thread_id": None},
+        )
+
+        rec.assert_called_once()
+        assert rec.call_args.kwargs.get("inject") is False
+        assert rec.call_args.kwargs.get("notice_id") == "fixedid"
+        adapter.send_cron_notice.assert_called_once_with("123", "fixedid", metadata={"thread_id": None})
+
+    def test_button_mode_without_support_falls_back_to_auto(self, monkeypatch):
+        import cron.pending_notices as pn
+        rec = MagicMock(return_value="nid")
+        monkeypatch.setattr(pn, "record", rec)
+        adapter = MagicMock()
+        adapter.SUPPORTS_CRON_BUTTONS = False
+        loop = MagicMock()
+        _record_session_notice(
+            "button", "telegram", "123", "Hello", None, self._job(),
+            adapter=adapter, loop=loop,
+        )
+        rec.assert_called_once()
+        assert rec.call_args.kwargs.get("inject") is True
+        adapter.send_cron_notice.assert_not_called()
+
+
 class TestDeliverResultErrorReturns:
     """Verify _deliver_result returns error strings on failure, None on success."""
 

From 920674b4894c1b15686b572e5bdc1417bde8d860 Mon Sep 17 00:00:00 2001
From: beardthelion <beardthelion@users.noreply.github.com>
Date: Mon, 1 Jun 2026 23:07:16 -0500
Subject: [PATCH 8/8] docs(cron): document notify_session button mode

Update the user guide and cron internals for the three-way
cron.notify_session knob (auto/button/off; the legacy bool still maps to
auto/off). Cover the inline Add-to-context / Dismiss buttons, the
SUPPORTS_CRON_BUTTONS platform fallback to auto, and the restart-durable
on-disk buffer the buttons resolve against.
---
 website/docs/developer-guide/cron-internals.md |  8 ++++++++
 website/docs/user-guide/features/cron.md       | 10 ++++++----
 2 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md
index 0824d76b26b7f..05a7b02e14741 100644
--- a/website/docs/developer-guide/cron-internals.md
+++ b/website/docs/developer-guide/cron-internals.md
@@ -198,6 +198,14 @@ Cron deliveries are NOT mirrored into gateway session conversation history. They
 
 When `cron.notify_session` is enabled (the default), deliveries are additionally buffered (`~/.hermes/cron/pending_notices.json`, keyed by `platform:chat_id`) and folded into the system prompt of the target chat's next turn as a `[System note: ...]` block, then drained. This gives the agent awareness of its own cron output while keeping the message history (and thus alternation) untouched.
 
+`cron.notify_session` normalizes to three modes via `normalize_notify_mode` (the legacy `true`/`false` map to `auto`/`off`):
+
+- `auto` buffers each entry as injectable, so the next turn's drain folds it in automatically.
+- `button` buffers the entry as held (`inject=False`) and, on a live adapter that sets `SUPPORTS_CRON_BUTTONS` (currently Telegram), sends an accept/dismiss prompt via `send_cron_notice`. Tapping **Add to context** calls `pending_notices.mark_accepted` (flipping the entry to injectable); **Dismiss** calls `pending_notices.dismiss`. Because the on-disk buffer is the source of truth, the buttons survive a gateway restart. Platforms without button support (or the standalone no-adapter delivery path) fall back to `auto`.
+- `off` skips buffering entirely.
+
+The drain in `gateway/run.py` stays mode-agnostic: it returns only entries with `inject` truthy and leaves held ones in place, so the gating decision lives entirely at record/accept time.
+
 ## Recursion Guard
 
 Cron-run sessions have the `cronjob` toolset disabled. This prevents:
diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md
index 638c9c8240cc3..ad018823ae291 100644
--- a/website/docs/user-guide/features/cron.md
+++ b/website/docs/user-guide/features/cron.md
@@ -329,16 +329,18 @@ cron:
 
 ### Session awareness
 
-Cron deliveries are sent straight to the platform and are not added to the chat's conversation history, so by default the agent has no record of what a scheduled job sent. When `cron.notify_session` is enabled (the default), each delivery is buffered and surfaced to the agent on that chat's next message, as a `[System note: ...]` block in the system prompt. This keeps the agent aware of what its own jobs delivered without writing to the message history (which would otherwise break message alternation).
+Cron deliveries are sent straight to the platform and are not added to the chat's conversation history, so on its own the agent has no record of what a scheduled job sent. The `cron.notify_session` setting controls how the agent is made aware, without ever writing to the message history (which would break message alternation):
 
-The note is consumed once: it appears on the next turn after a delivery, then clears.
+- `auto` (default): each delivery is buffered and surfaced to the agent on that chat's next message, as a `[System note: ...]` block in the system prompt. The note is consumed once, then clears.
+- `button`: the delivery is followed by an inline prompt with **Add to context** and **Dismiss** buttons (on platforms that support them, currently Telegram). The content reaches the agent's context only if you tap **Add**; **Dismiss** drops it. This is useful on smaller models where you want tight control over what enters the context window. Platforms without inline buttons fall back to `auto`.
+- `off`: deliveries are fire-and-forget and the agent is never notified.
 
-To turn this off and keep deliveries fire-and-forget:
+The legacy boolean values still work: `true` maps to `auto` and `false` maps to `off`.
 
 ```yaml
 # ~/.hermes/config.yaml
 cron:
-  notify_session: false
+  notify_session: button   # auto (default) | button | off
 ```
 
 ### Silent suppression