diff --git a/gateway/run.py b/gateway/run.py index f11686ccd360b..3766d7f880291 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -9281,6 +9281,8 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g "chat_id": source.chat_id or "", "session_id": session_entry.session_id, "message": message_text[:500], + "trigger": "message", + "interrupt_depth": 0, } await self.hooks.emit("agent:start", hook_ctx) @@ -18793,6 +18795,26 @@ async def _notify_long_running(): except Exception: pass + # Emit agent:start for the drained follow-up turn — the main + # dispatch fires this before its _run_agent call, but the + # interrupt/queue drain path historically did not, so hooks + # that observe turn starts (SessionStart-style integrations, + # activity loggers, visualizers) silently missed follow-ups. + # Mirror the main-path payload, built from this turn's source + # and the final (already-transcribed) next_message. Build it + # once and reuse it for the paired agent:end below — exactly as + # the main path reuses `hook_ctx` for both start and end. + followup_hook_ctx = { + "platform": next_source.platform.value if next_source.platform else "", + "user_id": next_source.user_id, + "chat_id": next_source.chat_id or "", + "session_id": session_id, + "message": next_message[:500], + "trigger": "interrupt", + "interrupt_depth": _interrupt_depth + 1, + } + await self.hooks.emit("agent:start", followup_hook_ctx) + followup_result = await self._run_agent( message=next_message, context_prompt=context_prompt, @@ -18805,6 +18827,22 @@ async def _notify_long_running(): event_message_id=next_message_id, channel_prompt=next_channel_prompt, ) + + # Pair the drained follow-up's agent:start with an agent:end so + # start/end-pairing hooks stay balanced on interrupted turns. + # The main dispatch only emits its agent:end (9428) for the + # outermost turn; the recursive follow-up re-enters _run_agent, + # not _handle_message_with_agent, so its end must be emitted + # here. Mirror the main-path end: {**hook_ctx, "response": …}. + _followup_response = ( + followup_result.get("final_response", "") + if isinstance(followup_result, dict) + else "" + ) + await self.hooks.emit("agent:end", { + **followup_hook_ctx, + "response": (_followup_response or "")[:500], + }) return _preserve_queued_followup_history_offset(result, followup_result) finally: # Stop progress sender, interrupt monitor, and notification task diff --git a/tests/gateway/test_agent_start_trigger.py b/tests/gateway/test_agent_start_trigger.py new file mode 100644 index 0000000000000..d404b035397c7 --- /dev/null +++ b/tests/gateway/test_agent_start_trigger.py @@ -0,0 +1,105 @@ +"""The MAIN-dispatch ``agent:start`` payload must carry the turn discriminator. + +``agent:start`` is emitted at two sites in ``gateway/run.py``: the MAIN inbound +dispatch (a fresh user message) and the interrupt/drain follow-up path. To let +hooks tell the two apart, each payload carries ``trigger`` (a string, kept open +for future turn kinds like ``"goal"``/``"schedule"``) and ``interrupt_depth`` +(an int). + +The drain emit is exercised end-to-end in ``test_drain_emits_agent_start.py``, +which drives the real ``_run_agent`` drain path and asserts +``trigger="interrupt"`` with the live ``_interrupt_depth + 1`` value. The MAIN +emit, by contrast, sits ~630 lines deep inside ``_handle_message_with_agent``, +behind session-store, DB and env I/O that make the method impractical to drive +in isolation. Rather than mock that whole world, this test statically inspects +the actual dict literal the production code hands to ``hooks.emit("agent:start", +...)`` and pins ``trigger="message"``/``interrupt_depth=0`` — a fresh inbound +turn is never an interrupt. It asserts against the real construction, not a +copy of it. +""" + +import ast +import inspect + +import gateway.run + + +def _module_tree(): + return ast.parse(inspect.getsource(gateway.run)) + + +def _find_function(tree, name): + for node in ast.walk(tree): + if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)) and node.name == name: + return node + return None + + +def _is_emit_agent_start(node): + return ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "emit" + and node.args + and isinstance(node.args[0], ast.Constant) + and node.args[0].value == "agent:start" + ) + + +def _agent_start_payload_dict(func): + """Return the ``ast.Dict`` literal emitted as the agent:start payload. + + Handles both an inline dict argument and a payload passed by name (the main + path builds ``hook_ctx = {...}`` then emits ``hook_ctx``); for the latter we + resolve the last in-function assignment of that name to a dict literal. + """ + payload = None + for node in ast.walk(func): + if isinstance(node, ast.Call) and _is_emit_agent_start(node) and len(node.args) >= 2: + payload = node.args[1] + break + if isinstance(payload, ast.Dict): + return payload + if isinstance(payload, ast.Name): + resolved = None + for node in ast.walk(func): + if isinstance(node, ast.Assign) and isinstance(node.value, ast.Dict): + if any(isinstance(t, ast.Name) and t.id == payload.id for t in node.targets): + resolved = node.value + return resolved + return None + + +def _string_keys(dict_node): + return {k.value for k in dict_node.keys if isinstance(k, ast.Constant)} + + +def _const_items(dict_node): + items = {} + for key, value in zip(dict_node.keys, dict_node.values): + if isinstance(key, ast.Constant) and isinstance(value, ast.Constant): + items[key.value] = value.value + return items + + +def test_main_dispatch_agent_start_payload_is_message_trigger_depth0(): + func = _find_function(_module_tree(), "_handle_message_with_agent") + assert func is not None, "could not locate _handle_message_with_agent" + + payload = _agent_start_payload_dict(func) + assert payload is not None, "could not locate the main-dispatch agent:start payload dict" + + keys = _string_keys(payload) + assert { + "platform", + "user_id", + "chat_id", + "session_id", + "message", + "trigger", + "interrupt_depth", + } <= keys, f"main agent:start payload is missing discriminator keys; has {sorted(map(str, keys))}" + + consts = _const_items(payload) + assert consts.get("trigger") == "message" + assert consts.get("interrupt_depth") == 0 diff --git a/tests/gateway/test_drain_agent_end_symmetry.py b/tests/gateway/test_drain_agent_end_symmetry.py new file mode 100644 index 0000000000000..b6ed712580686 --- /dev/null +++ b/tests/gateway/test_drain_agent_end_symmetry.py @@ -0,0 +1,175 @@ +"""agent:start / agent:end symmetry on the interrupt/drain follow-up turn. + +Companion to ``test_drain_emits_agent_start.py``. That suite proved the drain +path emits ``agent:start`` for the follow-up turn; this one is about the +matching ``agent:end``. + +Ground truth (verified against commit 451386527, see FINDINGS.md): ``agent:end`` +is emitted at exactly ONE site in ``gateway/run.py`` — the MAIN dispatch in +``_handle_message_with_agent`` — while ``agent:start`` is emitted at TWO sites +(main dispatch + the drain path in ``_run_agent``). So a drained turn fires +``agent:start`` once more than ``agent:end``: the follow-up turn's start has no +matching end. These tests drive the REAL ``_run_agent`` drain path (same seam +as the start suite) and assert the desired SYMMETRIC contract: every +``agent:start`` on the drain path is paired by an ``agent:end`` carrying the +same ``trigger``/``interrupt_depth`` discriminator, plus the follow-up turn's +response — and that NO end is emitted on the paths where no start is either. +""" + +import pytest + +from gateway.platforms.base import MessageEvent, MessageType + +# Reuse the drain harness verbatim (DRY) — same fixtures the agent:start suite +# uses to drive the real _run_agent / interrupt-drain path. +from tests.gateway.test_drain_emits_agent_start import ( + SESSION_ID, + _drive_drain, + _source, +) + + +def _starts(hooks): + return [ctx for (etype, ctx) in hooks.calls if etype == "agent:start"] + + +def _ends(hooks): + return [ctx for (etype, ctx) in hooks.calls if etype == "agent:end"] + + +@pytest.mark.asyncio +async def test_drain_followup_start_and_end_are_balanced(monkeypatch, tmp_path): + """One drained follow-up → equal agent:start and agent:end counts.""" + followup = MessageEvent( + text="follow up text", + message_type=MessageType.TEXT, + source=_source(user_id="userB"), + message_id="m2", + ) + hooks = await _drive_drain( + monkeypatch, tmp_path, followup, prepared_text="follow up text" + ) + + starts, ends = _starts(hooks), _ends(hooks) + assert len(starts) == 1, f"expected one drain agent:start, got {len(starts)}" + assert len(ends) == len(starts), ( + "drain follow-up must emit a matching agent:end for its agent:start " + f"(start={len(starts)}, end={len(ends)})" + ) + + +@pytest.mark.asyncio +async def test_drain_end_mirrors_start_payload(monkeypatch, tmp_path): + """The drain agent:end mirrors the start payload + carries the response.""" + followup = MessageEvent( + text="follow up text", + message_type=MessageType.TEXT, + source=_source(user_id="userB"), + message_id="m2", + ) + hooks = await _drive_drain( + monkeypatch, tmp_path, followup, prepared_text="follow up text" + ) + + ends = _ends(hooks) + assert len(ends) == 1 + end = ends[0] + # Same discriminator + identity fields as the start emit it pairs with. + assert end["trigger"] == "interrupt" + assert end["interrupt_depth"] == 1 + assert end["platform"] == "telegram" + assert end["user_id"] == "userB" + assert end["chat_id"] == "9001" + assert end["session_id"] == SESSION_ID + assert end["message"] == "follow up text" + # Mirrors the main-path end (9428): a response field, capped at 500 chars. + assert "response" in end + assert isinstance(end["response"], str) + assert len(end["response"]) <= 500 + + +@pytest.mark.asyncio +async def test_nested_drain_end_increments_depth(monkeypatch, tmp_path): + """An interrupt-of-an-interrupt pairs its start/end at depth 2.""" + followup = MessageEvent( + text="deeper follow up", + message_type=MessageType.TEXT, + source=_source(), + message_id="m8", + ) + hooks = await _drive_drain( + monkeypatch, + tmp_path, + followup, + prepared_text="deeper follow up", + interrupt_depth=1, + ) + + starts, ends = _starts(hooks), _ends(hooks) + assert len(starts) == 1 + assert len(ends) == len(starts) + assert ends[0]["trigger"] == "interrupt" + assert ends[0]["interrupt_depth"] == 2 + + +@pytest.mark.asyncio +async def test_no_end_when_followup_text_is_none(monkeypatch, tmp_path): + """Follow-up dropped before _run_agent (transcription → None): neither a + start NOR an end may fire — symmetric absence.""" + followup = MessageEvent( + text="", + message_type=MessageType.VOICE, + source=_source(), + media_urls=["/tmp/silent.ogg"], + media_types=["audio/ogg"], + message_id="m5", + ) + hooks = await _drive_drain(monkeypatch, tmp_path, followup, prepared_text=None) + + assert _starts(hooks) == [] + assert _ends(hooks) == [] + + +@pytest.mark.asyncio +async def test_no_end_at_max_interrupt_depth(monkeypatch, tmp_path): + """At _MAX_INTERRUPT_DEPTH the drain re-queues instead of recursing and + returns BEFORE the start emit — so neither a start NOR an end may fire. + Guards against a future move of the emit above the depth cap.""" + followup = MessageEvent( + text="too deep", + message_type=MessageType.TEXT, + source=_source(), + message_id="m9", + ) + hooks = await _drive_drain( + monkeypatch, + tmp_path, + followup, + prepared_text="too deep", + interrupt_depth=3, # == GatewayRunner._MAX_INTERRUPT_DEPTH + ) + + assert _starts(hooks) == [] + assert _ends(hooks) == [] + + +@pytest.mark.asyncio +async def test_no_end_when_goal_continuation_inactive(monkeypatch, tmp_path): + """A stale /goal continuation is discarded before _run_agent — neither a + start NOR an end may fire.""" + followup = MessageEvent( + text="[Continuing toward your standing goal]\nGoal: ship the thing", + message_type=MessageType.TEXT, + source=_source(), + message_id="m6", + ) + hooks = await _drive_drain( + monkeypatch, + tmp_path, + followup, + prepared_text="should never be used", + goal_active=False, + ) + + assert _starts(hooks) == [] + assert _ends(hooks) == [] diff --git a/tests/gateway/test_drain_emits_agent_start.py b/tests/gateway/test_drain_emits_agent_start.py new file mode 100644 index 0000000000000..f8033e7ea4159 --- /dev/null +++ b/tests/gateway/test_drain_emits_agent_start.py @@ -0,0 +1,359 @@ +"""Regression: the interrupt/drain follow-up path must emit ``agent:start``. + +In ``busy_input_mode: interrupt`` a message typed while the agent is busy +interrupts the run, is queued in ``_pending_messages``, and is later promoted +via the drain path inside ``GatewayRunner._run_agent`` (the recursive +follow-up turn). The MAIN dispatch emits ``agent:start`` before running the +agent, but the drain/follow-up path historically did NOT — so every hook +listening on ``agent:start`` (SessionStart-style integrations, activity +loggers, visualizers) silently missed interrupt/drain follow-up turns. + +These tests drive ``_run_agent`` directly with a frozen pending event (the +same seam used by ``test_run_progress_interrupt.py``) and assert: + + * a plain-text follow-up emits ``agent:start`` once, with a payload shaped + like the main-path one (platform, user_id, chat_id, session_id, message); + * a voice follow-up emits the FINAL transcribed text — not the raw audio + placeholder — because the payload is built from ``next_message`` (the + output of ``_prepare_inbound_message_text``), not the queued placeholder; + * the message is truncated to 500 chars, mirroring the main path; + * NO ``agent:start`` is emitted when the follow-up is discarded before it + reaches ``_run_agent`` (transcription yields ``None``; a stale /goal + continuation is dropped). +""" + +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, + SendResult, +) +from gateway.session import SessionSource + +SESSION_ID = "sess-drain" +SESSION_KEY = "agent:main:telegram:dm:9001" + + +class _CaptureAdapter(BasePlatformAdapter): + def __init__(self, platform=Platform.TELEGRAM): + super().__init__(PlatformConfig(enabled=True, token="***"), platform) + self.sent = [] + + 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: + self.sent.append({"chat_id": chat_id, "content": content}) + return SendResult(success=True, message_id="x") + + async def edit_message(self, chat_id, message_id, content) -> SendResult: + return SendResult(success=True, message_id=message_id) + + 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): + return {"id": chat_id} + + +class _RecordingHooks: + """Records every emit so tests can assert agent:start (or its absence).""" + + def __init__(self): + self.calls = [] + self.loaded_hooks = False + + async def emit(self, event_type, context=None): + self.calls.append((event_type, context)) + + def agent_start_payloads(self): + return [ctx for (etype, ctx) in self.calls if etype == "agent:start"] + + +class _DrainAgent: + """Fake AIAgent whose run returns an interrupted result with a queued + follow-up still pending — exactly what the interrupt drain path consumes.""" + + def __init__(self, **kwargs): + self.tool_progress_callback = kwargs.get("tool_progress_callback") + self.tools = [] + self._interrupted = False + + @property + def is_interrupted(self) -> bool: + return self._interrupted + + def run_conversation(self, message, conversation_history=None, task_id=None, **kwargs): + return { + "final_response": "partial answer", + "messages": [{"role": "user", "content": "original turn"}], + "api_calls": 1, + "interrupted": True, + } + + +def _make_runner(adapter, hooks): + 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 = hooks + runner.config = SimpleNamespace( + thread_sessions_per_user=False, + group_sessions_per_user=False, + stt_enabled=False, + ) + return runner + + +def _source(user_id="userA"): + return SessionSource( + platform=Platform.TELEGRAM, + chat_id="9001", + chat_type="dm", + user_id=user_id, + ) + + +async def _drive_drain( + monkeypatch, + tmp_path, + pending_event, + *, + prepared_text, + goal_active=None, + interrupt_depth=0, +): + """Run _run_agent with one frozen pending event queued for the drain. + + ``prepared_text`` is what the (stubbed) preprocessing pipeline yields for + the pending event — i.e. the FINAL text the follow-up turn runs on (a + transcription for a voice note). ``None`` means the follow-up is dropped. + """ + monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "all") + + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *a, **k: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + + fake_run_agent = types.ModuleType("run_agent") + fake_run_agent.AIAgent = _DrainAgent + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + + adapter = _CaptureAdapter() + hooks = _RecordingHooks() + runner = _make_runner(adapter, hooks) + + 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": "fake"} + ) + + # Stub the preprocessing pipeline so the drain's transcription step is + # deterministic and isolated. This stands in for the real STT/vision + # pipeline; the contract under test is that the emit uses ITS result. + async def _fake_prepare(*, event, source, history): + return prepared_text + + monkeypatch.setattr(runner, "_prepare_inbound_message_text", _fake_prepare) + + if goal_active is not None: + monkeypatch.setattr( + runner, "_goal_still_active_for_session", lambda sid: goal_active + ) + + # Queue the follow-up exactly as an interrupt would have. + adapter._pending_messages[SESSION_KEY] = pending_event + + await runner._run_agent( + message="original turn", + context_prompt="", + history=[], + source=_source(), + session_id=SESSION_ID, + session_key=SESSION_KEY, + _interrupt_depth=interrupt_depth, + ) + return hooks + + +@pytest.mark.asyncio +async def test_text_followup_emits_agent_start(monkeypatch, tmp_path): + """A plain-text drained follow-up emits one agent:start with a main-path + shaped payload built from the FOLLOW-UP's source.""" + followup = MessageEvent( + text="follow up text", + message_type=MessageType.TEXT, + source=_source(user_id="userB"), + message_id="m2", + ) + hooks = await _drive_drain( + monkeypatch, tmp_path, followup, prepared_text="follow up text" + ) + + payloads = hooks.agent_start_payloads() + assert len(payloads) == 1, ( + "drain follow-up must emit agent:start exactly once " + f"(emitted {len(payloads)})" + ) + assert payloads[0] == { + "platform": "telegram", + "user_id": "userB", + "chat_id": "9001", + "session_id": SESSION_ID, + "message": "follow up text", + "trigger": "interrupt", + "interrupt_depth": 1, + } + + +@pytest.mark.asyncio +async def test_voice_followup_emits_transcribed_text_not_audio(monkeypatch, tmp_path): + """A voice drained follow-up emits the transcribed text, never the raw + audio path / media placeholder.""" + followup = MessageEvent( + text="", + message_type=MessageType.VOICE, + source=_source(), + media_urls=["/tmp/voice-xyz.ogg"], + media_types=["audio/ogg"], + message_id="m3", + ) + transcript = "remind me to water the plants at six" + hooks = await _drive_drain( + monkeypatch, tmp_path, followup, prepared_text=transcript + ) + + payloads = hooks.agent_start_payloads() + assert len(payloads) == 1 + assert payloads[0]["message"] == transcript + assert "/tmp/voice-xyz.ogg" not in payloads[0]["message"] + assert "User sent audio" not in payloads[0]["message"] + + +@pytest.mark.asyncio +async def test_agent_start_message_truncated_to_500(monkeypatch, tmp_path): + """The payload message is capped at 500 chars, like the main path.""" + followup = MessageEvent( + text="x" * 600, + message_type=MessageType.TEXT, + source=_source(), + message_id="m4", + ) + hooks = await _drive_drain( + monkeypatch, tmp_path, followup, prepared_text="x" * 600 + ) + + payloads = hooks.agent_start_payloads() + assert len(payloads) == 1 + assert len(payloads[0]["message"]) == 500 + + +@pytest.mark.asyncio +async def test_no_emit_when_followup_text_is_none(monkeypatch, tmp_path): + """If preprocessing yields None the follow-up never reaches _run_agent — + no agent:start may fire.""" + followup = MessageEvent( + text="", + message_type=MessageType.VOICE, + source=_source(), + media_urls=["/tmp/silent.ogg"], + media_types=["audio/ogg"], + message_id="m5", + ) + hooks = await _drive_drain( + monkeypatch, tmp_path, followup, prepared_text=None + ) + + assert hooks.agent_start_payloads() == [] + + +@pytest.mark.asyncio +async def test_no_emit_when_goal_continuation_inactive(monkeypatch, tmp_path): + """A stale /goal continuation is discarded before _run_agent — no + agent:start may fire.""" + followup = MessageEvent( + text="[Continuing toward your standing goal]\nGoal: ship the thing", + message_type=MessageType.TEXT, + source=_source(), + message_id="m6", + ) + hooks = await _drive_drain( + monkeypatch, + tmp_path, + followup, + prepared_text="should never be used", + goal_active=False, + ) + + assert hooks.agent_start_payloads() == [] + + +@pytest.mark.asyncio +async def test_followup_carries_interrupt_trigger_and_depth(monkeypatch, tmp_path): + """A first-level drained follow-up is tagged trigger="interrupt", depth 1.""" + followup = MessageEvent( + text="follow up text", + message_type=MessageType.TEXT, + source=_source(), + message_id="m7", + ) + hooks = await _drive_drain( + monkeypatch, tmp_path, followup, prepared_text="follow up text" + ) + + payloads = hooks.agent_start_payloads() + assert len(payloads) == 1 + assert payloads[0]["trigger"] == "interrupt" + assert payloads[0]["interrupt_depth"] == 1 + + +@pytest.mark.asyncio +async def test_nested_interrupt_increments_depth(monkeypatch, tmp_path): + """An interrupt-of-an-interrupt reports depth _interrupt_depth + 1. + + Driving ``_run_agent`` with ``_interrupt_depth=1`` (a turn that is itself a + follow-up) makes its drained follow-up the second level — the emit must use + ``_interrupt_depth + 1`` == 2, not a hard-coded 1. + """ + followup = MessageEvent( + text="deeper follow up", + message_type=MessageType.TEXT, + source=_source(), + message_id="m8", + ) + hooks = await _drive_drain( + monkeypatch, + tmp_path, + followup, + prepared_text="deeper follow up", + interrupt_depth=1, + ) + + payloads = hooks.agent_start_payloads() + assert len(payloads) == 1 + assert payloads[0]["trigger"] == "interrupt" + assert payloads[0]["interrupt_depth"] == 2