diff --git a/tests/gateway/test_goal_continuation_drain.py b/tests/gateway/test_goal_continuation_drain.py new file mode 100644 index 00000000000..58c111dde64 --- /dev/null +++ b/tests/gateway/test_goal_continuation_drain.py @@ -0,0 +1,198 @@ +"""Regression: /goal continuations enqueued via the adapter FIFO are drained +automatically — no extra user message required (#47699). + +Issue #47699 reported that a Slack ``/goal`` continuation could be enqueued by +``_post_turn_goal_continuation`` → ``_enqueue_fifo`` (adapter pending slot) but +never consumed until the next real inbound message woke the session. + +On the current tree the continuation is enqueued while the adapter's +``_process_message_background`` frame is still live (the runner hook runs +inside ``self._message_handler(event)``), so the adapter's in-band pending +drain — and, for late arrivals, the finally-block late-arrival drain — spawns +the follow-up turn without user input. These tests pin that contract from the +adapter dispatch layer down, so a future refactor that moves the goal hook +after the drain (or changes FIFO key derivation) fails loudly instead of +silently stalling goal loops on messaging gateways. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType +from gateway.session import SessionSource, build_session_key + + +class _DrainProbeAdapter(BasePlatformAdapter): + """Minimal concrete adapter that records deliveries.""" + + def __init__(self) -> None: + super().__init__(PlatformConfig(enabled=True, token="x"), Platform.SLACK) + self.sent: list[str] = [] + + async def start(self): # pragma: no cover - unused + pass + + async def stop(self): # pragma: no cover - unused + pass + + async def connect(self): # pragma: no cover - unused + pass + + async def disconnect(self): # pragma: no cover - unused + pass + + async def get_chat_info(self, chat_id): # pragma: no cover - unused + return {} + + async def send(self, chat_id, content, reply_to=None, metadata=None): + self.sent.append(content) + + class _R: + success = True + message_id = "m1" + + return _R() + + async def send_typing(self, chat_id, metadata=None): + pass + + +def _slack_thread_source() -> SessionSource: + return SessionSource( + platform=Platform.SLACK, + user_id="U1", + chat_id="C1", + user_name="tester", + chat_type="channel", + thread_id="1718600000.000100", + ) + + +CONTINUATION_TEXT = "[Continuing toward your standing goal]\nGoal: ship it" + + +@pytest.fixture() +def hermes_home(tmp_path, monkeypatch): + from pathlib import Path + + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setattr(Path, "home", lambda: tmp_path) + monkeypatch.setenv("HERMES_HOME", str(home)) + + from hermes_cli import goals + + goals._DB_CACHE.clear() + yield home + goals._DB_CACHE.clear() + + +@pytest.mark.asyncio +async def test_fifo_enqueued_continuation_is_drained_without_new_user_message(): + """A continuation placed in the adapter pending slot during the handler + frame (exactly what _post_turn_goal_continuation does via _enqueue_fifo) + must start a second turn automatically.""" + adapter = _DrainProbeAdapter() + src = _slack_thread_source() + key = build_session_key(src) + handled: list[str] = [] + + async def handler(event): + handled.append(event.text) + if len(handled) == 1: + # Mirror the runner's goal hook: enqueue the synthetic + # continuation into the adapter FIFO while this frame is live. + cont = MessageEvent( + text=CONTINUATION_TEXT, + message_type=MessageType.TEXT, + source=src, + ) + adapter._pending_messages[key] = cont + return f"reply-{len(handled)}" + + adapter.set_message_handler(handler) + event = MessageEvent(text="do the thing", message_type=MessageType.TEXT, source=src) + + await adapter._process_message_background(event, key) + # The in-band drain hands off to a fresh task (#17758); let it run. + for _ in range(40): + if len(handled) >= 2: + break + await asyncio.sleep(0.05) + + assert handled == ["do the thing", CONTINUATION_TEXT], ( + "goal continuation was enqueued but never drained — " + "a user nudge would be required (#47699)" + ) + assert adapter.sent == ["reply-1", "reply-2"] + # The drain task must have released the session guard when the chain ended. + for _ in range(40): + if key not in adapter._active_sessions: + break + await asyncio.sleep(0.05) + assert key not in adapter._active_sessions + + +@pytest.mark.asyncio +async def test_runner_goal_hook_enqueues_into_the_key_the_adapter_drains(hermes_home): + """_post_turn_goal_continuation resolves the FIFO key via + _session_key_for_source; the adapter drain uses build_session_key on the + event source. These must agree or the continuation is orphaned under a + key nobody drains (the silent-stall shape from #47699).""" + from unittest.mock import MagicMock, patch + from datetime import datetime + import uuid + + from gateway.run import GatewayRunner + from gateway.session import SessionEntry + from hermes_cli.goals import GoalManager + + src = _slack_thread_source() + adapter_key = build_session_key(src) + + runner = object.__new__(GatewayRunner) + from gateway.config import GatewayConfig + + runner.config = GatewayConfig( + platforms={Platform.SLACK: PlatformConfig(enabled=True, token="x")}, + ) + runner._queued_events = {} + session_entry = SessionEntry( + session_key=adapter_key, + session_id=f"goal-sess-{uuid.uuid4().hex[:8]}", + created_at=datetime.now(), + updated_at=datetime.now(), + platform=Platform.SLACK, + chat_type="channel", + ) + runner.session_store = MagicMock() + runner.session_store.get_or_create_session.return_value = session_entry + runner.session_store._generate_session_key.return_value = adapter_key + + adapter = _DrainProbeAdapter() + runner.adapters = {Platform.SLACK: adapter} + + GoalManager(session_entry.session_id).set("ship it") + with patch( + "hermes_cli.goals.judge_goal", + return_value=("continue", "still needs work", False, None, False), + ): + await runner._post_turn_goal_continuation( + session_entry=session_entry, + source=src, + final_response="partial progress", + ) + await asyncio.sleep(0.05) + + assert adapter_key in adapter._pending_messages, ( + "continuation enqueued under a different key than the adapter " + f"drains: pending keys={list(adapter._pending_messages)} " + f"expected={adapter_key}" + ) + assert adapter._pending_messages[adapter_key].text.startswith( + "[Continuing toward your standing goal]" + )