File size: 5,917 Bytes
9aa5185 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | """Tests for /queue message consumption after normal agent completion.
Verifies that messages queued via /queue (which store in
adapter._pending_messages WITHOUT triggering an interrupt) are consumed
after the agent finishes its current task — not silently dropped.
"""
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from gateway.platforms.base import (
BasePlatformAdapter,
MessageEvent,
MessageType,
PlatformConfig,
Platform,
)
# ---------------------------------------------------------------------------
# Minimal adapter for testing pending message storage
# ---------------------------------------------------------------------------
class _StubAdapter(BasePlatformAdapter):
def __init__(self):
super().__init__(PlatformConfig(enabled=True, token="test"), Platform.TELEGRAM)
async def connect(self) -> bool:
return True
async def disconnect(self) -> None:
self._mark_disconnected()
async def send(self, chat_id, content, reply_to=None, metadata=None):
from gateway.platforms.base import SendResult
return SendResult(success=True, message_id="msg-1")
async def get_chat_info(self, chat_id):
return {"id": chat_id, "type": "dm"}
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestQueueMessageStorage:
"""Verify /queue stores messages correctly in adapter._pending_messages."""
def test_queue_stores_message_in_pending(self):
adapter = _StubAdapter()
session_key = "telegram:user:123"
event = MessageEvent(
text="do this next",
message_type=MessageType.TEXT,
source=MagicMock(chat_id="123", platform=Platform.TELEGRAM),
message_id="q1",
)
adapter._pending_messages[session_key] = event
assert session_key in adapter._pending_messages
assert adapter._pending_messages[session_key].text == "do this next"
def test_get_pending_message_consumes_and_clears(self):
adapter = _StubAdapter()
session_key = "telegram:user:123"
event = MessageEvent(
text="queued prompt",
message_type=MessageType.TEXT,
source=MagicMock(chat_id="123", platform=Platform.TELEGRAM),
message_id="q2",
)
adapter._pending_messages[session_key] = event
retrieved = adapter.get_pending_message(session_key)
assert retrieved is not None
assert retrieved.text == "queued prompt"
# Should be consumed (cleared)
assert adapter.get_pending_message(session_key) is None
def test_queue_does_not_set_interrupt_event(self):
"""The whole point of /queue — no interrupt signal."""
adapter = _StubAdapter()
session_key = "telegram:user:123"
# Simulate an active session (agent running)
adapter._active_sessions[session_key] = asyncio.Event()
# Store a queued message (what /queue does)
event = MessageEvent(
text="queued",
message_type=MessageType.TEXT,
source=MagicMock(),
message_id="q3",
)
adapter._pending_messages[session_key] = event
# The interrupt event should NOT be set
assert not adapter._active_sessions[session_key].is_set()
assert not adapter.has_pending_interrupt(session_key)
def test_regular_message_sets_interrupt_event(self):
"""Contrast: regular messages DO trigger interrupt."""
adapter = _StubAdapter()
session_key = "telegram:user:123"
adapter._active_sessions[session_key] = asyncio.Event()
# Simulate regular message arrival (what handle_message does)
event = MessageEvent(
text="new message",
message_type=MessageType.TEXT,
source=MagicMock(),
message_id="m1",
)
adapter._pending_messages[session_key] = event
adapter._active_sessions[session_key].set() # this is what handle_message does
assert adapter.has_pending_interrupt(session_key)
class TestQueueConsumptionAfterCompletion:
"""Verify that pending messages are consumed after normal completion."""
def test_pending_message_available_after_normal_completion(self):
"""After agent finishes without interrupt, pending message should
still be retrievable from adapter._pending_messages."""
adapter = _StubAdapter()
session_key = "telegram:user:123"
# Simulate: agent starts, /queue stores a message, agent finishes
adapter._active_sessions[session_key] = asyncio.Event()
event = MessageEvent(
text="process this after",
message_type=MessageType.TEXT,
source=MagicMock(),
message_id="q4",
)
adapter._pending_messages[session_key] = event
# Agent finishes (no interrupt)
del adapter._active_sessions[session_key]
# The queued message should still be retrievable
retrieved = adapter.get_pending_message(session_key)
assert retrieved is not None
assert retrieved.text == "process this after"
def test_multiple_queues_last_one_wins(self):
"""If user /queue's multiple times, last message overwrites."""
adapter = _StubAdapter()
session_key = "telegram:user:123"
for text in ["first", "second", "third"]:
event = MessageEvent(
text=text,
message_type=MessageType.TEXT,
source=MagicMock(),
message_id=f"q-{text}",
)
adapter._pending_messages[session_key] = event
retrieved = adapter.get_pending_message(session_key)
assert retrieved.text == "third"
|