diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index b88367a7963de..21ac94093d446 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -336,6 +336,15 @@ def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text checkpoint_parts += ["Visible response before the interruption:", visible] checkpoint = "\n\n".join(checkpoint_parts) correction = f"[Context from the interrupted assistant response]\n{checkpoint}\n\n{text}" + try: + from gateway.cron_delivery_handoff import ( + drain_agent_delivery_references, format_delivery_reference_block, + ) + reference_block = format_delivery_reference_block(drain_agent_delivery_references(agent)) + if reference_block: + correction = f"{reference_block}\n\n{correction}" + except Exception: + logger.debug("cron delivery reference context drain failed", exc_info=True) # The live tail is normally user or tool, so an assistant placeholder + correction # keeps strict alternation; if the tail is already assistant, the checkpoint is folded diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py index c5dba0935113a..035d0bf37aa3a 100644 --- a/cron/scheduler_delivery.py +++ b/cron/scheduler_delivery.py @@ -204,6 +204,36 @@ def _cron_mirror_message(job: dict, text: str) -> str: return f"[Cron delivery: {_cron_display_name(job)}]\n{text}" +def _record_opted_in_cron_delivery_handoff( + job: dict, *, brief_text: str, platform_name: str, chat_id: str, + thread_id: Optional[str] = None, user_id: Optional[str] = None, + session_id: Optional[str] = None, +) -> None: + """Best-effort scoped handoff after a successful opted-in persist. Never raises. + + Delivery must never be reported failed because this bookkeeping broke — same + rule as the existing mirror. Only ``attach_to_session is True`` records; the + idle follow-up path already reloads SQLite and must stay fail-open otherwise. + """ + try: + if job.get("attach_to_session") is not True: + return + if not session_id: + from gateway.mirror import _find_session_id + session_id = _find_session_id( + platform_name, str(chat_id), thread_id=thread_id, user_id=user_id, + ) + from gateway.cron_delivery_handoff import record_opted_in_cron_handoff + record_opted_in_cron_handoff( + session_id=session_id, job=job, brief_text=brief_text, provenance="cron", + ) + except Exception as e: + logger.debug( + "Job '%s': cron delivery handoff record failed for %s:%s: %s", + job.get("id", "?"), platform_name, chat_id, e, + ) + + def _maybe_mirror_cron_delivery( job: dict, platform_name: str, chat_id: str, mirror_text: str, thread_id: Optional[str] = None, user_id: Optional[str] = None, *, enabled: bool = False, @@ -234,6 +264,10 @@ def _maybe_mirror_cron_delivery( logger.info( "Job '%s': mirrored delivery into %s:%s session transcript", job.get("id", "?"), platform_name, chat_id) + _record_opted_in_cron_delivery_handoff( + job, brief_text=text, platform_name=platform_name, chat_id=chat_id, + thread_id=thread_id, user_id=user_id, + ) else: logger.debug( "Job '%s': delivery mirror skipped for %s:%s " @@ -311,11 +345,17 @@ def _seed_cron_session( # bails on populated chats. _entry = session_store.get_or_create_session(dest_source) seeded_session_id = getattr(_entry, "session_id", None) - return mirror_to_session( + ok = mirror_to_session( platform_name, str(chat_id), _cron_mirror_message(job, text), source_label="cron", thread_id=thread_id, user_id=user_id, role="user", session_id=seeded_session_id, ) + if ok: + _record_opted_in_cron_delivery_handoff( + job, brief_text=text, platform_name=platform_name, chat_id=chat_id, + thread_id=thread_id, user_id=user_id, session_id=seeded_session_id, + ) + return ok def _seed_cron_thread_session( diff --git a/gateway/cron_delivery_handoff.py b/gateway/cron_delivery_handoff.py new file mode 100644 index 0000000000000..5f1b5a6bab289 --- /dev/null +++ b/gateway/cron_delivery_handoff.py @@ -0,0 +1,117 @@ +"""Scoped opted-in cron delivery handoff for an already-running turn. + +Records trusted provenance at the moment a successful attach/mirror/seed +lands, keyed by exact ``session_id``. An accepted busy redirect moves the +pending brief onto the running agent as provider-facing reference context +(not a new user instruction). Fail-open: never raise. +""" + +from __future__ import annotations + +import logging +import threading +from typing import Any, Iterable, List, Optional + +logger = logging.getLogger(__name__) + +_lock = threading.Lock() +_pending: dict[str, list[dict[str, Any]]] = {} + +_AGENT_SLOT = "_pending_cron_delivery_references" +_REFERENCE_MARKER = "[Reference context from a cron delivery attached to this session]" + + +def record_opted_in_cron_handoff( + *, + session_id: Optional[str], + job: Optional[dict] = None, + job_id: Optional[str] = None, + brief_text: Optional[str] = None, + provenance: str = "cron", +) -> None: + """Record a scoped handoff after a successful opted-in persist. Never raises.""" + try: + if not session_id: + return + if job is not None and job.get("attach_to_session") is not True: + return + text = (brief_text or "").strip() + if not text: + return + resolved_job_id = job_id + if not resolved_job_id and isinstance(job, dict): + resolved_job_id = job.get("id") + entry = { + "job_id": str(resolved_job_id or "cron"), + "brief_text": text, + "provenance": provenance, + "session_id": str(session_id), + } + key = str(session_id) + with _lock: + bucket = _pending.setdefault(key, []) + if any( + item.get("job_id") == entry["job_id"] and item.get("brief_text") == entry["brief_text"] + for item in bucket + ): + return + bucket.append(entry) + except Exception: + logger.debug("cron delivery handoff record failed", exc_info=True) + + +def take_opted_in_cron_handoffs(session_id: Optional[str]) -> List[dict[str, Any]]: + """Consume pending handoffs for this exact session_id. Empty on any error.""" + try: + if not session_id: + return [] + with _lock: + return list(_pending.pop(str(session_id), []) or []) + except Exception: + logger.debug("cron delivery handoff take failed", exc_info=True) + return [] + + +def attach_handoffs_to_agent(agent: Any, handoffs: Iterable[dict[str, Any]]) -> None: + """Move consumed session handoffs onto the running agent. Fail-open.""" + try: + items = [ + h for h in (handoffs or []) + if isinstance(h, dict) and (h.get("brief_text") or "").strip() + ] + if agent is None or not items: + return + existing = getattr(agent, _AGENT_SLOT, None) + if not isinstance(existing, list): + existing = [] + setattr(agent, _AGENT_SLOT, existing) + existing.extend(items) + except Exception: + logger.debug("cron delivery handoff attach failed", exc_info=True) + + +def drain_agent_delivery_references(agent: Any) -> List[dict[str, Any]]: + """Consume agent-scoped delivery reference context. Empty on any error.""" + try: + items = getattr(agent, _AGENT_SLOT, None) + setattr(agent, _AGENT_SLOT, []) + if not isinstance(items, list): + return [] + return [h for h in items if isinstance(h, dict)] + except Exception: + logger.debug("cron delivery handoff drain failed", exc_info=True) + return [] + + +def format_delivery_reference_block(handoffs: Iterable[dict[str, Any]]) -> str: + """Labeled provider-facing reference block; never a user instruction.""" + parts: List[str] = [] + for handoff in handoffs or []: + if not isinstance(handoff, dict): + continue + brief = (handoff.get("brief_text") or "").strip() + if not brief: + continue + job_id = handoff.get("job_id") or "cron" + parts.append(f"{_REFERENCE_MARKER}\njob_id={job_id}\n{brief}") + return "\n\n".join(parts) diff --git a/gateway/run_busy.py b/gateway/run_busy.py index 89cadb02a82e8..e7d1bc4e0d6b5 100644 --- a/gateway/run_busy.py +++ b/gateway/run_busy.py @@ -639,6 +639,34 @@ def _demote_interrupt(session_key: str, why: str) -> str: logger.info("Demoting busy_input_mode 'interrupt' to 'queue' for session %s because %s", session_key, why) return "queue" + def _attach_cron_delivery_handoff_on_redirect( + self, running_agent, event: Optional[MessageEvent], session_key: str, + ) -> None: + """Move a same-session opted-in cron brief onto the agent after an accepted redirect. + + Quoted follow-ups, missing session identity, and unsupported runtimes are no-ops. + Fail-open: never raise, never rewrite ``event.text`` / redirect content. + """ + try: + if running_agent is None or event is None: + return + if getattr(event, "reply_to_message_id", None) or getattr(event, "reply_to_text", None): + return + if getattr(running_agent, "_supports_active_turn_redirect", False) is not True: + return + store = getattr(self, "session_store", None) + if store is None or not session_key: + return + session_id = store.peek_session_id(session_key) + if not session_id: + return + from gateway.cron_delivery_handoff import ( + attach_handoffs_to_agent, take_opted_in_cron_handoffs, + ) + attach_handoffs_to_agent(running_agent, take_opted_in_cron_handoffs(session_id)) + except Exception: + logger.debug("cron delivery handoff attach on redirect failed", exc_info=True) + def _try_agent_verb( self, running_agent, verb: str, text: str, session_key: str, *, event: Optional[MessageEvent] = None ) -> bool: @@ -647,7 +675,10 @@ def _try_agent_verb( call_text = self._steer_text_with_origin(text, event) if event else text if verb == "steer": return self._steer_running_agent(running_agent, call_text) - return bool(getattr(running_agent, verb)(call_text)) + accepted = bool(getattr(running_agent, verb)(call_text)) + if accepted and verb == "redirect": + self._attach_cron_delivery_handoff_on_redirect(running_agent, event, session_key) + return accepted except Exception as exc: logger.warning("Gateway %s failed for session %s: %s", verb, session_key, exc) return False diff --git a/tests/cron/test_discord_cron_followup_context.py b/tests/cron/test_discord_cron_followup_context.py new file mode 100644 index 0000000000000..8365cdddac507 --- /dev/null +++ b/tests/cron/test_discord_cron_followup_context.py @@ -0,0 +1,135 @@ +"""Offline reproduction: delivered cron context versus an active Discord turn. + +Real scheduler, Discord send, SQLite session store, gateway busy entry points, +redirect buffer and redirect application. Only Discord's network boundary is +replaced; no model or Discord service is contacted. The live messages are a +snapshot taken before delivery, as in GatewayRunner._hmwa_prepare_turn. +""" + +import asyncio +import copy +import threading +from dataclasses import asdict +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from cron.scheduler_delivery import _deliver_result +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.platforms.event import MessageEvent +from gateway.run import GatewayRunner +from gateway.session import SessionSource +from hermes_constants import get_hermes_home +from plugins.platforms.discord.adapter import DiscordAdapter +from run_agent import AIAgent + +BRIEF = "Scheduled briefing: widget validation is broken; open an issue for it." +FOLLOWUP = "Aren't you opening an issue?" + + +@pytest.fixture(params=["dm", "thread"]) +def conversation(tmp_path, monkeypatch, request): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr("gateway.run._hermes_home", tmp_path) + (tmp_path / "config.yaml").write_text( + "cron:\n wrap_response: false\n" + "platforms:\n discord:\n enabled: true\n token: offline-test\n", + encoding="utf-8", + ) + config = GatewayConfig() + config.platforms[Platform.DISCORD] = PlatformConfig(enabled=True, token="offline-test") + runner = GatewayRunner(config=config) + source = SessionSource( + platform=Platform.DISCORD, chat_id="1001", chat_type=request.param, user_id="2001", + thread_id="1001" if request.param == "thread" else None, + ) + session = runner.session_store.get_or_create_session(source) + history = [ + {"role": "user", "content": "Review the README pull request."}, + {"role": "assistant", "content": "The README pull request only needs a wording change."}, + ] + for message in history: + runner.session_store.append_to_transcript(session.session_id, message) + + # DiscordAdapter.send and create_handoff_thread remain real. DMs cannot + # host a child thread; an explicitly targeted existing thread is retained. + import discord + + channel = MagicMock(spec=discord.DMChannel if request.param == "dm" else discord.Thread) + channel.id = int(source.chat_id) + channel.send = AsyncMock(return_value=SimpleNamespace(id=3001)) + adapter = DiscordAdapter(config.platforms[Platform.DISCORD]) + adapter._client = SimpleNamespace(get_channel=lambda _id: channel) + adapter.set_session_store(runner.session_store) + runner.adapters[Platform.DISCORD] = adapter + assert get_hermes_home() == tmp_path + return runner, source, session, adapter, channel, history + + +async def deliver(conversation, attach): + runner, source, session, adapter, channel, history = conversation + job = {"id": "offline-brief", "name": "Widget briefing", "deliver": "origin", "origin": asdict(source)} + job["origin"]["platform"] = source.platform.value + if attach is not None: + job["attach_to_session"] = attach + error = await asyncio.to_thread( + _deliver_result, job, BRIEF, {Platform.DISCORD: adapter}, asyncio.get_running_loop(), + ) + assert error is None + channel.send.assert_awaited_once() + assert channel.send.await_args.kwargs["content"] == BRIEF + reply_session = runner.session_store.get_or_create_session(source) + assert reply_session.session_id == session.session_id + return runner.session_store.load_transcript(reply_session.session_id) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("attach", [None, False, True]) +async def test_idle_followup_observes_only_opted_in_delivery(conversation, attach): + transcript = await deliver(conversation, attach) + assert (BRIEF in str(transcript)) is (attach is True) + assert [{key: message[key] for key in ("role", "content")} for message in transcript[:2]] == conversation[-1] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ["normal", "priority"]) +async def test_active_followup_observes_opted_in_delivery(conversation, route): + from agent.conversation_loop import _apply_active_turn_redirect + + runner, source, session, adapter, channel, history = conversation + runner.session_store.append_to_transcript( + session.session_id, + {"role": "user", "content": "Continue reviewing that README pull request."}, + ) + live_messages = copy.deepcopy(runner.session_store.load_transcript(session.session_id)) + cached_prefix = copy.deepcopy(live_messages) + transcript = await deliver(conversation, True) + assert BRIEF in str(transcript), "The opted-in brief must first reach the exact reply-facing session" + assert BRIEF not in str(live_messages), "Delivery must not mutate a running request" + + # Run the real AIAgent redirect implementation without constructing a + # provider client. No model call is made; the request-active event is the + # deterministic pause point for the in-flight snapshot. + agent = AIAgent.__new__(AIAgent) + agent._supports_active_turn_redirect = True + agent._model_request_active = threading.Event() + agent._model_request_active.set() + agent._interrupt_requested = False + agent._current_streamed_assistant_text = "The README wording can be improved." + event = MessageEvent(text=FOLLOWUP, source=source, message_id="4001") + assert event.reply_to_message_id is None and event.reply_to_text is None + if route == "normal": + outcome = await runner._resolve_busy_steer_or_redirect(event, session.session_key, "interrupt", agent) + assert outcome.redirected + else: + await runner._hm_busy_interrupt(event, source, agent, session.session_key) + correction = agent._drain_pending_redirect() + assert correction and FOLLOWUP in correction + _apply_active_turn_redirect(agent, live_messages, correction) + assert live_messages[:len(cached_prefix)] == cached_prefix + assert all(a["role"] != b["role"] for a, b in zip(live_messages, live_messages[1:])) + assert BRIEF in str(live_messages), ( + "Successful opted-in Discord delivery is durable, but the accepted active follow-up " + "only carries the README snapshot and the user's bare question, not the cron brief" + )