fix(gateway): include opted-in cron context in active-turn redirects

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

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(

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)

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

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"
+    )