diff --git a/gateway/run_turn.py b/gateway/run_turn.py index 16b29133ff..e2c07b3e84 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -21,7 +21,7 @@ from contextlib import nullcontext, suppress from contextvars import copy_context from gateway.config import Platform from gateway.media_repair import repair_explicit_computer_use_media_paths -from gateway.platforms.base import BasePlatformAdapter +from gateway.platforms.base import BasePlatformAdapter, ProcessingOutcome from gateway.platforms.event import MessageEvent from gateway.session import ( SessionSource, _session_key_namespace, build_channel_continuity_note, @@ -3621,15 +3621,35 @@ class GatewayTurnMixin: # whole _run_agent chain unwinds — too late for the in-band follow-up. Use the same (session_key, # session_id) the recursive call runs under so the snapshot matches exactly what the follow-up's # guard will consult. Fail-safe in helper. - await self._refresh_agent_cache_message_count(session_key, session_id) + # Acknowledge the follow-up the way an idle-session message is: this in-band drain is the only + # place a queued/interrupting message ever runs, so base.py's hook site is never entered for it. + # Resolve the adapter from the follow-up's OWN source — a multiplexed gateway can route it to a + # different profile's adapter, and only that instance holds the per-message reaction state. + from gateway.run_turn_followup_ack import _followup_cancel_outcome, _run_followup_processing_hook + _hook_adapter = self._adapter_for_source(next_source) if pending_event is not None else None + await _run_followup_processing_hook(_hook_adapter, pending_event, "on_processing_start") + # The re-baseline sits inside the try: a /stop landing on its DB await must still close the marker + # (the helper's own ``except Exception`` does not catch cancellation). + try: + await self._refresh_agent_cache_message_count(session_key, session_id) - followup_result = await self._run_agent( - message=next_message, context_prompt=turn_ctx.context_prompt, history=updated_history, - source=next_source, session_id=session_id, session_key=next_session_key, - run_generation=run_generation, _interrupt_depth=_interrupt_depth + 1, - event_message_id=next_message_id, inbound_message_id=next_inbound_id, - channel_prompt=next_channel_prompt, message_type=next_message_type, - ) + followup_result = await self._run_agent( + message=next_message, context_prompt=turn_ctx.context_prompt, history=updated_history, + source=next_source, session_id=session_id, session_key=next_session_key, + run_generation=run_generation, _interrupt_depth=_interrupt_depth + 1, + event_message_id=next_message_id, inbound_message_id=next_inbound_id, + channel_prompt=next_channel_prompt, message_type=next_message_type, + ) + except asyncio.CancelledError: + await _run_followup_processing_hook( + _hook_adapter, pending_event, "on_processing_complete", _followup_cancel_outcome(_hook_adapter)) + raise + except BaseException: + await _run_followup_processing_hook( + _hook_adapter, pending_event, "on_processing_complete", ProcessingOutcome.FAILURE) + raise + await _run_followup_processing_hook( + _hook_adapter, pending_event, "on_processing_complete", ProcessingOutcome.SUCCESS) merged = _preserve_queued_followup_history_offset(result, followup_result) # The TERMINAL turn of the chain owns the ledger identity for the outer final send, which # the adapter brackets against the event that OPENED the chain. Without this the terminal diff --git a/gateway/run_turn_followup_ack.py b/gateway/run_turn_followup_ack.py new file mode 100644 index 0000000000..a49fff1e3a --- /dev/null +++ b/gateway/run_turn_followup_ack.py @@ -0,0 +1,60 @@ +"""Processing-lifecycle hooks for runner-drained queued follow-ups. + +A message that arrives mid-turn is parked in the adapter's pending slot and drained in-band by +``TurnRunner._run_agent_queued_followup``, never by ``BasePlatformAdapter._process_message_background`` +— the only other call site for ``on_processing_start`` / ``on_processing_complete``. Without firing +them here every adapter that renders a read-receipt reaction from the hooks silently skips queued, +interrupting and steer-demoted messages (#72502, salvage #72503). +""" + +from __future__ import annotations + +import asyncio + +from gateway.platforms.base import BasePlatformAdapter, MessageEvent, ProcessingOutcome + + +def _followup_processing_hooks_apply(adapter, event: MessageEvent | None) -> bool: + """Both conditions are necessary: a real inbound platform message to acknowledge (adapters key their + marker off ``message_id`` — Slack/Telegram/Feishu/Matrix/Photon — or off ``raw_message`` — Signal, + Discord; synthetic drains such as ``/goal`` continuations, wake-ups and startup auto-resume carry + neither and must stay silent), and an adapter that overrides ``on_processing_start``. We bracket, so + both halves must belong to us: a complete-only adapter (Google Chat reaps its typing card there, + webhook ends its per-delivery session) would otherwise be handed a completion for a turn whose reply + is delivered only after the drain chain unwinds back into ``_process_message_background``.""" + if adapter is None or event is None: + return False + if not (getattr(event, "message_id", None) or getattr(event, "raw_message", None)): + return False + start_hook = getattr(type(adapter), "on_processing_start", None) + return start_hook is not None and start_hook is not BasePlatformAdapter.on_processing_start + + +def _followup_cancel_outcome(adapter) -> ProcessingOutcome: + """Classify a cancelled follow-up exactly as ``_process_message_background`` does: only cancels the + adapter itself routed (``/stop``, ``/new``, ``/reset``, cleanup) are CANCELLED, anything else is a + failure. Signal and Matrix leave the in-progress marker in place on CANCELLED, so reporting an + unexpected cancellation as CANCELLED would strand it.""" + expected = getattr(adapter, "_expected_cancelled_tasks", None) + if expected is None: + return ProcessingOutcome.FAILURE + try: + current = asyncio.current_task() + except RuntimeError: + current = None + if current is None: + return ProcessingOutcome.FAILURE + try: + return ProcessingOutcome.CANCELLED if current in expected else ProcessingOutcome.FAILURE + except TypeError: + return ProcessingOutcome.FAILURE + + +async def _run_followup_processing_hook(adapter, event: MessageEvent | None, hook_name: str, *args) -> None: + """Fire one lifecycle hook for a runner-drained follow-up; no-op per ``_followup_processing_hooks_apply``.""" + if not _followup_processing_hooks_apply(adapter, event): + return + run_hook = getattr(adapter, "_run_processing_hook", None) + if not callable(run_hook): + return + await run_hook(hook_name, event, *args) diff --git a/tests/gateway/test_queued_followup_processing_hooks.py b/tests/gateway/test_queued_followup_processing_hooks.py new file mode 100644 index 0000000000..1356fd808b --- /dev/null +++ b/tests/gateway/test_queued_followup_processing_hooks.py @@ -0,0 +1,313 @@ +"""Processing-hook parity for queued follow-up turns. + +A message that arrives while a turn is already running is parked in the +adapter's ``_pending_messages`` slot and drained *in-band* by +``GatewayRunner._run_agent`` rather than by +``BasePlatformAdapter._process_message_background``. The runner-side drain +must still fire the ``on_processing_start`` / ``on_processing_complete`` +lifecycle hooks, otherwise every platform that renders a read-receipt +reaction from those hooks (Slack 👀, Discord, Telegram, Feishu, Matrix, +Signal, ...) silently skips the acknowledgement for mid-turn messages. +""" + +import importlib +import sys +import types +from types import SimpleNamespace + +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import ( + BasePlatformAdapter, + MessageEvent, + MessageType, + ProcessingOutcome, + SendResult, +) +from gateway.session import SessionSource + + +class HookRecordingAdapter(BasePlatformAdapter): + """Adapter that records the processing-hook lifecycle it is driven through.""" + + def __init__(self): + super().__init__(PlatformConfig(enabled=True, token="***"), Platform.TELEGRAM) + self.started: list = [] + self.completed: list = [] + + async def connect(self) -> bool: + return True + + async def disconnect(self) -> None: + return None + + async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult: + return SendResult(success=True, message_id="sent-1") + + async def send_typing(self, chat_id, metadata=None) -> None: + return None + + async def stop_typing(self, chat_id) -> None: + return None + + async def get_chat_info(self, chat_id: str): + return {"id": chat_id} + + async def on_processing_start(self, event: MessageEvent) -> None: + self.started.append(getattr(event, "message_id", None)) + + async def on_processing_complete(self, event, outcome) -> None: + self.completed.append((getattr(event, "message_id", None), outcome)) + + +class _TwoTurnAgent: + calls: list = [] + + def __init__(self, **kwargs): + self.tools = [] + + def run_conversation(self, message, conversation_history=None, task_id=None): + type(self).calls.append(message) + return { + "final_response": f"done-{len(type(self).calls)}", + "messages": [], + "api_calls": 1, + } + + +class _RaisingSecondTurnAgent: + calls: list = [] + + def __init__(self, **kwargs): + self.tools = [] + + def run_conversation(self, message, conversation_history=None, task_id=None): + type(self).calls.append(message) + if len(type(self).calls) >= 2: + raise RuntimeError("boom in the queued follow-up turn") + return { + "final_response": "done-1", + "messages": [], + "api_calls": 1, + } + + +def _make_runner(adapter): + gateway_run = importlib.import_module("gateway.run") + runner = object.__new__(gateway_run.GatewayRunner) + runner.adapters = {adapter.platform: adapter} + runner._voice_mode = {} + runner._prefill_messages = [] + runner._ephemeral_system_prompt = "" + runner._reasoning_config = None + runner._provider_routing = {} + runner._fallback_model = None + runner._session_db = None + runner._running_agents = {} + runner._session_run_generation = {} + runner.hooks = SimpleNamespace(loaded_hooks=False) + runner.config = SimpleNamespace( + thread_sessions_per_user=False, + group_sessions_per_user=False, + stt_enabled=False, + ) + runner._model = "openai/gpt-4.1-mini" + runner._base_url = None + return runner + + +def _install_fake_agent(monkeypatch, tmp_path, agent_cls): + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + + fake_run_agent = types.ModuleType("run_agent") + fake_run_agent.AIAgent = agent_cls + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + + gateway_run = importlib.import_module("gateway.run") + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr( + gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"} + ) + + +SESSION_KEY = "agent:main:telegram:dm:4242" + + +def _source(): + return SessionSource(platform=Platform.TELEGRAM, chat_id="4242", chat_type="dm") + + +@pytest.mark.asyncio +async def test_queued_followup_fires_processing_hooks(monkeypatch, tmp_path): + """The runner-drained follow-up gets the same start/complete hooks as a + message that arrives while the session is idle.""" + _TwoTurnAgent.calls = [] + _install_fake_agent(monkeypatch, tmp_path, _TwoTurnAgent) + + adapter = HookRecordingAdapter() + runner = _make_runner(adapter) + + adapter._pending_messages[SESSION_KEY] = MessageEvent( + text="the follow-up", + message_type=MessageType.TEXT, + source=_source(), + message_id="queued-1", + ) + + result = await runner._run_agent( + message="the first turn", + context_prompt="", + history=[], + source=_source(), + session_id="sess-hooks", + session_key=SESSION_KEY, + ) + + # The follow-up really did run in-band. + assert result["final_response"] == "done-2" + assert _TwoTurnAgent.calls == ["the first turn", "the follow-up"] + + # ...and it was acknowledged through the lifecycle hooks. + assert adapter.started == ["queued-1"] + assert adapter.completed == [("queued-1", ProcessingOutcome.SUCCESS)] + + +@pytest.mark.asyncio +async def test_queued_followup_failure_completes_the_hook(monkeypatch, tmp_path): + """A follow-up turn that blows up still closes its hook, so a platform + never strands a 'still working' marker on the user's message.""" + _RaisingSecondTurnAgent.calls = [] + _install_fake_agent(monkeypatch, tmp_path, _RaisingSecondTurnAgent) + + adapter = HookRecordingAdapter() + runner = _make_runner(adapter) + + adapter._pending_messages[SESSION_KEY] = MessageEvent( + text="the doomed follow-up", + message_type=MessageType.TEXT, + source=_source(), + message_id="queued-2", + ) + + with pytest.raises(RuntimeError): + await runner._run_agent( + message="the first turn", + context_prompt="", + history=[], + source=_source(), + session_id="sess-hooks-failure", + session_key=SESSION_KEY, + ) + + assert adapter.started == ["queued-2"] + assert adapter.completed == [("queued-2", ProcessingOutcome.FAILURE)] + + +@pytest.mark.asyncio +async def test_synthetic_followup_is_not_acknowledged(monkeypatch, tmp_path): + """Drains with no inbound platform message — /goal continuations, wake-ups, + CLI hand-offs — carry no message_id and must stay silent: there is nothing + on the platform to react to.""" + _TwoTurnAgent.calls = [] + _install_fake_agent(monkeypatch, tmp_path, _TwoTurnAgent) + + adapter = HookRecordingAdapter() + runner = _make_runner(adapter) + + adapter._pending_messages[SESSION_KEY] = MessageEvent( + text="synthetic continuation", + message_type=MessageType.TEXT, + source=_source(), + message_id=None, + ) + + result = await runner._run_agent( + message="the first turn", + context_prompt="", + history=[], + source=_source(), + session_id="sess-hooks-synthetic", + session_key=SESSION_KEY, + ) + + # It still ran — we only suppressed the acknowledgement, not the turn. + assert result["final_response"] == "done-2" + assert _TwoTurnAgent.calls == ["the first turn", "synthetic continuation"] + + assert adapter.started == [] + assert adapter.completed == [] + + +@pytest.mark.asyncio +async def test_raw_envelope_only_followup_is_acknowledged(monkeypatch, tmp_path): + """Signal never sets message_id — its hook keys off the raw envelope + (sender + timestamp_ms) — and Discord's reads raw_message. An event + carrying only a raw envelope is still a real inbound message.""" + _TwoTurnAgent.calls = [] + _install_fake_agent(monkeypatch, tmp_path, _TwoTurnAgent) + + adapter = HookRecordingAdapter() + runner = _make_runner(adapter) + + adapter._pending_messages[SESSION_KEY] = MessageEvent( + text="signal-shaped follow-up", + message_type=MessageType.TEXT, + source=_source(), + message_id=None, + raw_message={"sender": "+15550100", "timestamp_ms": 1700000000000}, + ) + + await runner._run_agent( + message="the first turn", + context_prompt="", + history=[], + source=_source(), + session_id="sess-hooks-raw", + session_key=SESSION_KEY, + ) + + assert adapter.started == [None] + assert adapter.completed == [(None, ProcessingOutcome.SUCCESS)] + + +class CompleteOnlyAdapter(HookRecordingAdapter): + """Google Chat and webhook implement on_processing_complete WITHOUT + on_processing_start; theirs is end-of-cycle teardown (reap the typing + card / end the delivery session), not a reaction.""" + + on_processing_start = BasePlatformAdapter.on_processing_start + + +@pytest.mark.asyncio +async def test_complete_only_adapter_is_left_alone(monkeypatch, tmp_path): + """We bracket, so both halves must belong to us. An adapter that only + implements the completion half must not be handed a completion here: at + this point the follow-up's reply has not been delivered yet, so its + teardown would fire against a live turn.""" + _TwoTurnAgent.calls = [] + _install_fake_agent(monkeypatch, tmp_path, _TwoTurnAgent) + + adapter = CompleteOnlyAdapter() + runner = _make_runner(adapter) + + adapter._pending_messages[SESSION_KEY] = MessageEvent( + text="the follow-up", + message_type=MessageType.TEXT, + source=_source(), + message_id="queued-3", + ) + + result = await runner._run_agent( + message="the first turn", + context_prompt="", + history=[], + source=_source(), + session_id="sess-hooks-complete-only", + session_key=SESSION_KEY, + ) + + assert result["final_response"] == "done-2" + assert adapter.completed == []