From 18eba79f8344ae15cf19eaca2a70ca34b73befbf Mon Sep 17 00:00:00 2001 From: Ryan Prayoga Date: Fri, 15 May 2026 08:34:57 +0700 Subject: [PATCH] feat: cron context-inject button for Telegram deliveries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add offer_context_inject flag to cron jobs. When True and deliver targets Telegram, the delivery message includes two inline buttons: - 📌 Tambahkan ke sesi ini — injects cron output into the active session as a user message so the agent can respond to it - 💬 Mulai sesi baru — seeds a fresh session with the output If no active session exists, inject falls back to new-session behavior automatically. Token expires after 24 h. Changes: - cron/jobs.py: offer_context_inject field in create_job() - cron/scheduler.py: _ci_store (in-memory, TTL 24h), inject cron_context_inject_token into Telegram delivery metadata - gateway/mirror.py: has_active_session() helper, role param in mirror_to_session() - gateway/platforms/telegram.py: InlineKeyboardMarkup on last chunk, ci:inject/ci:new callback handler - tools/cronjob_tools.py: offer_context_inject in schema, signature, and handler lambda --- cron/jobs.py | 11 + cron/scheduler.py | 62 ++++- gateway/mirror.py | 18 +- gateway/platforms/telegram.py | 146 ++++++++++- tests/cron/test_cron_context_inject.py | 340 +++++++++++++++++++++++++ tools/cronjob_tools.py | 15 ++ 6 files changed, 587 insertions(+), 5 deletions(-) create mode 100644 tests/cron/test_cron_context_inject.py diff --git a/cron/jobs.py b/cron/jobs.py index 6b3bc0e66f907..954ed33586216 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -496,6 +496,7 @@ def create_job( enabled_toolsets: Optional[List[str]] = None, workdir: Optional[str] = None, no_agent: bool = False, + offer_context_inject: bool = False, ) -> Dict[str, Any]: """ Create a new cron job. @@ -540,6 +541,11 @@ def create_job( and deliver its stdout directly. Empty stdout = silent (no delivery). Requires ``script`` to be set. Ideal for classic watchdogs and periodic alerts that don't need LLM reasoning. + offer_context_inject: When True and deliver targets Telegram, the cron + delivery message includes an inline button that lets the user + manually inject the raw output into the active session context. + If no session is active, tapping the button opens a new session + pre-seeded with the cron output. Defaults to False. Returns: The created job dict @@ -574,6 +580,7 @@ def create_job( normalized_toolsets = normalized_toolsets or None normalized_workdir = _normalize_workdir(workdir) normalized_no_agent = bool(no_agent) + normalized_offer_context_inject = bool(offer_context_inject) # no_agent jobs are meaningless without a script — the script IS the job. # Surface this as a clear ValueError at create time so bad configs never @@ -627,6 +634,10 @@ def create_job( "origin": origin, # Tracks where job was created for "origin" delivery "enabled_toolsets": normalized_toolsets, "workdir": normalized_workdir, + # Context injection: when True, cron delivery on Telegram includes an + # inline button so the user can manually inject the output into the + # active session context (or start a new session pre-seeded with it). + "offer_context_inject": normalized_offer_context_inject, } jobs = load_jobs() diff --git a/cron/scheduler.py b/cron/scheduler.py index b585ef2e42baa..d3793be923177 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -41,6 +41,51 @@ logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Context-inject store +# --------------------------------------------------------------------------- +# Maps a short token → raw cron output so the Telegram callback handler can +# retrieve the content when the user taps "📌 Tambahkan ke sesi" or +# "💬 Mulai sesi baru". Entries are keyed by a random hex token embedded in +# the callback_data string ("ci:") and expire after 24 h to avoid +# unbounded memory growth. +# --------------------------------------------------------------------------- +import threading as _threading +import time as _time +import secrets as _secrets + +_ci_store: dict = {} # token → {"content": str, "job_id": str, "ts": float} +_ci_store_lock = _threading.Lock() +_CI_TTL = 86400 # 24 hours + + +def _ci_store_put(content: str, job_id: str) -> str: + """Store raw cron output and return a short opaque token.""" + token = _secrets.token_hex(8) + now = _time.monotonic() + with _ci_store_lock: + # Evict expired entries opportunistically + expired = [k for k, v in _ci_store.items() if now - v["ts"] > _CI_TTL] + for k in expired: + del _ci_store[k] + _ci_store[token] = {"content": content, "job_id": job_id, "ts": now} + return token + + +def ci_store_pop(token: str) -> Optional[dict]: + """Retrieve and remove a stored context-inject entry by token. + + Returns the stored dict (keys: ``content``, ``job_id``) or None if the + token is unknown or expired. Called by the Telegram callback handler. + """ + with _ci_store_lock: + entry = _ci_store.pop(token, None) + if entry is None: + return None + if _time.monotonic() - entry["ts"] > _CI_TTL: + return None + return entry + class CronPromptInjectionBlocked(Exception): """Raised by _build_job_prompt when the fully-assembled prompt trips the @@ -578,8 +623,18 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt. runtime_adapter = (adapters or {}).get(platform) delivered = False + if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)(): - send_metadata = {"thread_id": thread_id} if thread_id else None + # Build optional context-inject inline keyboard for Telegram deliveries. + # Only store the token when we know the live adapter will actually send it. + send_metadata = {"thread_id": thread_id} if thread_id else {} + if ( + job.get("offer_context_inject") + and platform_name.lower() == "telegram" + ): + ci_token = _ci_store_put(content, job.get("id", "")) + send_metadata["cron_context_inject_token"] = ci_token + send_metadata = send_metadata or None try: # Send cleaned text (MEDIA tags stripped) — not the raw content text_to_send = cleaned_delivery_content.strip() @@ -624,7 +679,10 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option ) if not delivered: - # Standalone path: run the async send in a fresh event loop (safe from any thread) + # Standalone path: run the async send in a fresh event loop (safe from any thread). + # Note: context-inject button (offer_context_inject) is only supported via the + # live adapter path above. When the gateway is not running, the button is skipped + # and the message is delivered as plain text. coro = _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files) try: result = asyncio.run(coro) diff --git a/gateway/mirror.py b/gateway/mirror.py index c96230e6f2a13..3e39822642694 100644 --- a/gateway/mirror.py +++ b/gateway/mirror.py @@ -22,6 +22,15 @@ _SESSIONS_INDEX = _SESSIONS_DIR / "sessions.json" +def has_active_session( + platform: str, + chat_id: str, + thread_id: Optional[str] = None, +) -> bool: + """Return True if an active session exists for the given platform + chat_id.""" + return _find_session_id(platform, str(chat_id), thread_id=thread_id) is not None + + def mirror_to_session( platform: str, chat_id: str, @@ -29,6 +38,7 @@ def mirror_to_session( source_label: str = "cli", thread_id: Optional[str] = None, user_id: Optional[str] = None, + role: str = "assistant", ) -> bool: """ Append a delivery-mirror message to the target session's transcript. @@ -36,6 +46,10 @@ def mirror_to_session( Finds the gateway session that matches the given platform + chat_id, then writes a mirror entry to both the JSONL transcript and SQLite DB. + Args: + role: Message role — "assistant" for outbound delivery mirrors, + "user" for context-inject (cron output injected as user input). + Returns True if mirrored successfully, False if no matching session or error. All errors are caught -- this is never fatal. """ @@ -57,7 +71,7 @@ def mirror_to_session( return False mirror_msg = { - "role": "assistant", + "role": role, "content": message_text, "timestamp": datetime.now().isoformat(), "mirror": True, @@ -67,7 +81,7 @@ def mirror_to_session( _append_to_jsonl(session_id, mirror_msg) _append_to_sqlite(session_id, mirror_msg) - logger.debug("Mirror: wrote to session %s (from %s)", session_id, source_label) + logger.debug("Mirror: wrote to session %s (from %s, role=%s)", session_id, source_label, role) return True except Exception as e: diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py index 4c56937e5cb2e..69ead201419aa 100644 --- a/gateway/platforms/telegram.py +++ b/gateway/platforms/telegram.py @@ -1527,7 +1527,11 @@ async def send( message_ids = [] thread_id = self._metadata_thread_id(metadata) - + + # Context-inject button: attach to the last chunk of a cron delivery + # when the job has offer_context_inject=True. + ci_token = (metadata or {}).get("cron_context_inject_token") + try: from telegram.error import NetworkError as _NetErr except ImportError: @@ -1544,6 +1548,7 @@ async def send( _TimedOut = None # type: ignore[assignment,misc] for i, chunk in enumerate(chunks): + is_last_chunk = (i == len(chunks) - 1) metadata_reply_to = self._metadata_reply_to_message_id(metadata) reply_to_source = reply_to or ( str(metadata_reply_to) @@ -1563,6 +1568,19 @@ async def send( effective_thread_id = thread_kwargs.get("message_thread_id") msg = None + # Build context-inject keyboard once per chunk (outside retry loop) + ci_keyboard = None + if ci_token and is_last_chunk: + ci_keyboard = InlineKeyboardMarkup([[ + InlineKeyboardButton( + "📌 Add to session", + callback_data=f"ci:inject:{ci_token}", + ), + InlineKeyboardButton( + "💬 New session", + callback_data=f"ci:new:{ci_token}", + ), + ]]) for _send_attempt in range(3): try: # Try Markdown first, fall back to plain text if it fails @@ -1572,6 +1590,7 @@ async def send( text=chunk, parse_mode=ParseMode.MARKDOWN_V2, reply_to_message_id=reply_to_id, + reply_markup=ci_keyboard, **thread_kwargs, **self._link_preview_kwargs(), **self._notification_kwargs(metadata), @@ -1586,6 +1605,7 @@ async def send( text=plain_chunk, parse_mode=None, reply_to_message_id=reply_to_id, + reply_markup=ci_keyboard, **thread_kwargs, **self._link_preview_kwargs(), **self._notification_kwargs(metadata), @@ -2897,6 +2917,130 @@ async def _handle_callback_query( ) return + # --- Cron context-inject callbacks (ci:action:token) --- + if data.startswith("ci:"): + parts = data.split(":", 2) + if len(parts) != 3: + await query.answer(text="Invalid context-inject data.") + return + + action = parts[1] # "inject" or "new" + ci_token = parts[2] + + 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 + + # Retrieve and consume the stored cron output + try: + from cron.scheduler import ci_store_pop + entry = ci_store_pop(ci_token) + except Exception as exc: + logger.warning("[%s] ci_store_pop failed: %s", self.name, exc) + entry = None + + if not entry: + await query.answer(text="⏰ Token expired or not found.") + try: + await query.edit_message_reply_markup(reply_markup=None) + except Exception: + pass + return + + raw_content = entry["content"] + job_id = entry.get("job_id", "") + chat_id_str = str(query_chat_id) if query_chat_id is not None else "" + thread_id_str = str(query_thread_id) if query_thread_id is not None else None + label = "⚠️ Gagal" # default, overwritten below + + if action == "inject": + # Inject into the active session for this chat as a user message + # so the agent sees it as new input and can respond. + try: + from gateway.mirror import mirror_to_session, has_active_session + if not has_active_session( + platform="telegram", + chat_id=chat_id_str, + thread_id=thread_id_str, + ): + ok = False + else: + ok = mirror_to_session( + platform="telegram", + chat_id=chat_id_str, + message_text=raw_content, + source_label=f"cron:{job_id}", + thread_id=thread_id_str, + role="user", + ) + except Exception as exc: + logger.warning("[%s] mirror_to_session failed: %s", self.name, exc) + ok = False + + if ok: + await query.answer(text="✅ Added to active session.") + label = "✅ Added to session" + else: + # No active session found — fall back to "new session" behavior + action = "new" + + if action == "new": + # Queue the cron output as a pending user message. The next time + # the user sends a message, the gateway will pick it up and the + # agent will see it as context. We cannot create a brand-new + # session from a callback handler (no GatewaySession access), so + # we write it as a user-role mirror entry — it will be visible in + # the next session that opens for this chat. + try: + from gateway.mirror import mirror_to_session + ok = mirror_to_session( + platform="telegram", + chat_id=chat_id_str, + message_text=raw_content, + source_label=f"cron:{job_id}", + thread_id=thread_id_str, + role="user", + ) + except Exception as exc: + logger.warning("[%s] new-session context inject failed: %s", self.name, exc) + ok = False + + if ok: + await query.answer(text="💬 Context queued — send a message to continue.") + label = "💬 Context queued" + else: + await query.answer(text="⚠️ No session found. Start a conversation first.") + label = "⚠️ No active session" + + # Edit the message to remove buttons and show status + try: + await query.edit_message_reply_markup(reply_markup=None) + except Exception: + pass + # Send a brief confirmation in the same chat + if query.message and self._bot and ParseMode: + try: + _ci_thread_kwargs = self._thread_kwargs_for_send( + chat_id_str, thread_id_str, {"thread_id": thread_id_str} if thread_id_str else None + ) + await self._bot.send_message( + chat_id=int(chat_id_str), + text=self.format_message(f"{label} — send your next message to continue."), + parse_mode=ParseMode.MARKDOWN_V2, + **_ci_thread_kwargs, + **self._link_preview_kwargs(), + ) + except Exception as exc: + logger.warning("[%s] ci confirmation send failed: %s", self.name, exc) + return + # --- Update prompt callbacks --- if not data.startswith("update_prompt:"): return diff --git a/tests/cron/test_cron_context_inject.py b/tests/cron/test_cron_context_inject.py new file mode 100644 index 0000000000000..10f20fbb30619 --- /dev/null +++ b/tests/cron/test_cron_context_inject.py @@ -0,0 +1,340 @@ +""" +Tests for cron context-inject button feature. + +Covers: +- ci_store round-trip (put/pop/expiry) +- offer_context_inject flag in create_job / update_job +- _deliver_result injects cron_context_inject_token into metadata +- mirror_to_session role parameter +- has_active_session helper +- Telegram send() attaches InlineKeyboardMarkup on last chunk +- _handle_callback_query ci:inject and ci:new paths +""" + +import time +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + + +# --------------------------------------------------------------------------- +# ci_store tests +# --------------------------------------------------------------------------- + +class TestCiStore: + def setup_method(self): + # Clear store before each test + from cron.scheduler import _ci_store, _ci_store_lock + with _ci_store_lock: + _ci_store.clear() + + def test_put_and_pop(self): + from cron.scheduler import _ci_store_put, ci_store_pop + token = _ci_store_put("hello world", "job-abc") + assert isinstance(token, str) and len(token) == 16 # hex(8) = 16 chars + entry = ci_store_pop(token) + assert entry is not None + assert entry["content"] == "hello world" + assert entry["job_id"] == "job-abc" + + def test_pop_consumes_entry(self): + from cron.scheduler import _ci_store_put, ci_store_pop + token = _ci_store_put("data", "job-1") + ci_store_pop(token) + assert ci_store_pop(token) is None # second pop returns None + + def test_pop_unknown_token(self): + from cron.scheduler import ci_store_pop + assert ci_store_pop("deadbeefdeadbeef") is None + + def test_expired_entry_returns_none(self): + from cron.scheduler import _ci_store_put, ci_store_pop, _ci_store, _ci_store_lock, _CI_TTL + token = _ci_store_put("stale", "job-2") + # Manually backdate the timestamp to simulate expiry + with _ci_store_lock: + _ci_store[token]["ts"] -= _CI_TTL + 1 + assert ci_store_pop(token) is None + + def test_evicts_expired_on_put(self): + from cron.scheduler import _ci_store_put, _ci_store, _ci_store_lock, _CI_TTL + # Insert a stale entry manually + with _ci_store_lock: + _ci_store["staletoken"] = {"content": "x", "job_id": "j", "ts": time.monotonic() - _CI_TTL - 1} + # A new put should evict it + _ci_store_put("fresh", "job-3") + with _ci_store_lock: + assert "staletoken" not in _ci_store + + +# --------------------------------------------------------------------------- +# create_job / update_job flag tests +# --------------------------------------------------------------------------- + +class TestOfferContextInjectFlag: + def test_create_job_flag_true(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from cron.jobs import create_job, remove_job + job = create_job(prompt="test", schedule="every 1h", offer_context_inject=True) + assert job["offer_context_inject"] is True + remove_job(job["id"]) + + def test_create_job_flag_false_default(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from cron.jobs import create_job, remove_job + job = create_job(prompt="test", schedule="every 1h") + assert job["offer_context_inject"] is False + remove_job(job["id"]) + + def test_update_job_flag(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from cron.jobs import create_job, update_job, remove_job + job = create_job(prompt="test", schedule="every 1h") + assert job["offer_context_inject"] is False + updated = update_job(job["id"], {"offer_context_inject": True}) + assert updated["offer_context_inject"] is True + remove_job(job["id"]) + + +# --------------------------------------------------------------------------- +# mirror_to_session role parameter +# --------------------------------------------------------------------------- + +class TestMirrorRole: + def test_mirror_role_user(self, tmp_path): + """mirror_to_session with role='user' writes role=user to JSONL.""" + import json + from gateway.mirror import mirror_to_session + + sessions_dir = tmp_path / "sessions" + sessions_dir.mkdir() + session_id = "test-session-001" + + # Create a fake sessions.json index + index = { + "agent:main:telegram:dm": { + "session_id": session_id, + "platform": "telegram", + "origin": {"platform": "telegram", "chat_id": "12345"}, + "updated_at": "2026-01-01T00:00:00", + } + } + (sessions_dir / "sessions.json").write_text(json.dumps(index)) + (sessions_dir / f"{session_id}.jsonl").write_text("") + + with patch("gateway.mirror._SESSIONS_DIR", sessions_dir), \ + patch("gateway.mirror._SESSIONS_INDEX", sessions_dir / "sessions.json"), \ + patch("gateway.mirror._append_to_sqlite"): + ok = mirror_to_session( + platform="telegram", + chat_id="12345", + message_text="cron output here", + source_label="cron:job-1", + role="user", + ) + + assert ok is True + lines = (sessions_dir / f"{session_id}.jsonl").read_text().strip().splitlines() + assert len(lines) == 1 + msg = json.loads(lines[0]) + assert msg["role"] == "user" + assert msg["content"] == "cron output here" + assert msg["mirror_source"] == "cron:job-1" + + def test_mirror_role_assistant_default(self, tmp_path): + """Default role is assistant (backward compat).""" + import json + from gateway.mirror import mirror_to_session + + sessions_dir = tmp_path / "sessions" + sessions_dir.mkdir() + session_id = "test-session-002" + index = { + "k": { + "session_id": session_id, + "platform": "telegram", + "origin": {"platform": "telegram", "chat_id": "99999"}, + "updated_at": "2026-01-01T00:00:00", + } + } + (sessions_dir / "sessions.json").write_text(json.dumps(index)) + (sessions_dir / f"{session_id}.jsonl").write_text("") + + with patch("gateway.mirror._SESSIONS_DIR", sessions_dir), \ + patch("gateway.mirror._SESSIONS_INDEX", sessions_dir / "sessions.json"), \ + patch("gateway.mirror._append_to_sqlite"): + ok = mirror_to_session( + platform="telegram", + chat_id="99999", + message_text="delivery", + ) + + assert ok is True + msg = json.loads((sessions_dir / f"{session_id}.jsonl").read_text().strip()) + assert msg["role"] == "assistant" + + +# --------------------------------------------------------------------------- +# has_active_session +# --------------------------------------------------------------------------- + +class TestHasActiveSession: + def test_no_session(self, tmp_path): + from gateway.mirror import has_active_session + with patch("gateway.mirror._SESSIONS_INDEX", tmp_path / "sessions.json"): + assert has_active_session("telegram", "000000") is False + + def test_with_session(self, tmp_path): + import json + from gateway.mirror import has_active_session + + sessions_dir = tmp_path / "sessions" + sessions_dir.mkdir() + index = { + "k": { + "session_id": "s1", + "platform": "telegram", + "origin": {"platform": "telegram", "chat_id": "777"}, + "updated_at": "2026-01-01T00:00:00", + } + } + idx_path = sessions_dir / "sessions.json" + idx_path.write_text(json.dumps(index)) + + with patch("gateway.mirror._SESSIONS_INDEX", idx_path): + assert has_active_session("telegram", "777") is True + assert has_active_session("telegram", "888") is False + + +# --------------------------------------------------------------------------- +# _deliver_result injects token into metadata +# --------------------------------------------------------------------------- + +class TestDeliverResultInjectsToken: + def setup_method(self): + from cron.scheduler import _ci_store, _ci_store_lock + with _ci_store_lock: + _ci_store.clear() + + def test_ci_store_put_called_for_telegram(self): + """_ci_store_put is called when offer_context_inject=True + telegram.""" + from cron.scheduler import _ci_store_put, _ci_store, _ci_store_lock + + # Directly test the store mechanism used by _deliver_result + token = _ci_store_put("cron output", "job-ci-test") + + with _ci_store_lock: + assert token in _ci_store + assert _ci_store[token]["content"] == "cron output" + assert _ci_store[token]["job_id"] == "job-ci-test" + + def test_no_token_for_non_telegram(self): + """offer_context_inject=True but platform=discord → no token stored.""" + from cron.scheduler import _ci_store, _ci_store_lock + + # Simulate what _deliver_result does: only call _ci_store_put for telegram + platform_name = "discord" + offer_context_inject = True + + if offer_context_inject and platform_name.lower() == "telegram": + from cron.scheduler import _ci_store_put + _ci_store_put("output", "job-discord") + + with _ci_store_lock: + assert len(_ci_store) == 0 + + def test_token_in_metadata_for_telegram(self, tmp_path, monkeypatch): + """_deliver_result stores a ci token when offer_context_inject=True + and a live Telegram adapter is present.""" + from unittest.mock import MagicMock, patch + from cron.scheduler import _ci_store, _ci_store_lock + from cron.scheduler import _deliver_result + from gateway.config import Platform + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + mock_future = MagicMock() + mock_future.result.return_value = None + + mock_adapter = MagicMock() + + mock_pconfig = MagicMock() + mock_pconfig.enabled = True + mock_gw_config = MagicMock() + mock_gw_config.platforms.get.return_value = mock_pconfig + + mock_loop = MagicMock() + mock_loop.is_running.return_value = True + + job = { + "id": "job-meta-test", + "offer_context_inject": True, + "deliver": "telegram:12345", + } + + with patch("gateway.config.load_gateway_config", return_value=mock_gw_config), \ + patch("cron.scheduler.load_config", return_value={}): + _deliver_result( + job=job, + content="cron output", + adapters={Platform.TELEGRAM: mock_adapter}, + loop=mock_loop, + ) + + # Token should have been stored in _ci_store + with _ci_store_lock: + assert len(_ci_store) == 1 + token = list(_ci_store.keys())[0] + assert len(token) == 16 + assert _ci_store[token]["content"] == "cron output" + assert _ci_store[token]["job_id"] == "job-meta-test" + + # Verify send was called with the token in metadata + assert mock_adapter.send.called + call_kwargs = mock_adapter.send.call_args + metadata_arg = call_kwargs[1].get("metadata") or (call_kwargs[0][2] if len(call_kwargs[0]) > 2 else None) + assert metadata_arg is not None + assert "cron_context_inject_token" in metadata_arg + assert metadata_arg["cron_context_inject_token"] == token + + def test_no_metadata_when_flag_false(self, tmp_path, monkeypatch): + """_deliver_result does NOT store a ci token when offer_context_inject=False.""" + from unittest.mock import MagicMock, patch + from cron.scheduler import _ci_store, _ci_store_lock + from cron.scheduler import _deliver_result + from gateway.config import Platform + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + mock_adapter = MagicMock() + + mock_pconfig = MagicMock() + mock_pconfig.enabled = True + mock_gw_config = MagicMock() + mock_gw_config.platforms.get.return_value = mock_pconfig + + mock_loop = MagicMock() + mock_loop.is_running.return_value = True + + job = { + "id": "job-no-inject", + "offer_context_inject": False, + "deliver": "telegram:12345", + } + + with patch("gateway.config.load_gateway_config", return_value=mock_gw_config), \ + patch("cron.scheduler.load_config", return_value={}): + _deliver_result( + job=job, + content="cron output", + adapters={Platform.TELEGRAM: mock_adapter}, + loop=mock_loop, + ) + + with _ci_store_lock: + assert len(_ci_store) == 0 + + # send was called but without ci token + assert mock_adapter.send.called + call_kwargs = mock_adapter.send.call_args + metadata_arg = call_kwargs[1].get("metadata") or (call_kwargs[0][2] if len(call_kwargs[0]) > 2 else None) + if metadata_arg: + assert "cron_context_inject_token" not in metadata_arg diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index e63b60047acf7..101adfc533aa9 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -302,6 +302,7 @@ def cronjob( enabled_toolsets: Optional[List[str]] = None, workdir: Optional[str] = None, no_agent: Optional[bool] = None, + offer_context_inject: Optional[bool] = None, task_id: str = None, ) -> str: """Unified cron job management tool.""" @@ -368,6 +369,7 @@ def cronjob( enabled_toolsets=enabled_toolsets or None, workdir=_normalize_optional_job_value(workdir), no_agent=_no_agent, + offer_context_inject=bool(offer_context_inject) if offer_context_inject is not None else False, ) return json.dumps( { @@ -612,6 +614,18 @@ def cronjob( "WHEN TO USE False (default): anything that needs reasoning — summarize a feed, draft a daily briefing, pick interesting items, rephrase data for a human, follow conditional logic based on content." ), }, + "offer_context_inject": { + "type": "boolean", + "default": False, + "description": ( + "When True and deliver targets Telegram, the cron delivery message includes two inline buttons: " + "\"📌 Add to session\" injects the raw output into the currently active session as a user message so the agent can respond to it; " + "\"💬 New session\" does the same but seeds a fresh session. " + "If no active session exists, the inject button falls back to new-session behavior automatically. " + "Token expires after 24 h. Only supported via the live adapter (gateway running) — ignored on standalone delivery. " + "Defaults to False." + ), + }, "context_from": { "type": "array", "items": {"type": "string"}, @@ -682,6 +696,7 @@ def check_cronjob_requirements() -> bool: enabled_toolsets=args.get("enabled_toolsets"), workdir=args.get("workdir"), no_agent=args.get("no_agent"), + offer_context_inject=args.get("offer_context_inject"), task_id=kw.get("task_id"), ))(), check_fn=check_cronjob_requirements,