feat(cron): surface cron deliveries via system-prompt note or accept/dismiss buttons

cron/pending_notices.py

@@ -0,0 +1,232 @@
+"""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
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional, Union
+
+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 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 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"))
+        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,
+    notice_id: Optional[str] = None,
+    inject: bool = True,
+    base_dir: Optional[Path] = None,
+) -> Union[str, bool]:
+    """Buffer one delivered cron message for ``platform``/``chat_id``.
+
+    ``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)
+            data = _load(path)
+            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 nid
+    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 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 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.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 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

cron/scheduler.py

@@ -640,10 +640,19 @@ 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_mode = "auto"
     try:
+        from cron.pending_notices import normalize_notify_mode
         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_mode = normalize_notify_mode(cron_cfg.get("notify_session", True))
     except Exception:
         pass
 
@@ -665,6 +674,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 +785,11 @@ 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_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(
                     "Job '%s': live adapter delivery to %s:%s failed (%s), falling back to standalone",
@@ -804,12 +823,90 @@ 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_mode, platform_name, chat_id,
+                notice_text, thread_id, job,
+            )
 
     if delivery_errors:
         return "; ".join(delivery_errors)
     return None
 
 
+def _record_session_notice(
+    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.
+
+    ``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.
+    """
+    text = (text or "").strip()
+    if mode == "off" or not text:
+        return
+    try:
+        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_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)
+
+
 _DEFAULT_SCRIPT_TIMEOUT = 120  # seconds
 # Backward-compatible module override used by tests and emergency monkeypatches.
 _SCRIPT_TIMEOUT = _DEFAULT_SCRIPT_TIMEOUT

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,