fix(gateway): emit agent:end on interrupt/drain follow-up turns

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

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

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) == []