dinos
db1abce8d8
* chore: add pytest-asyncio in auto mode * test: migrate channel and stream tests to native async Convert run_async() wrapper tests to plain 'async def test_*' under pytest-asyncio auto mode. collect_events() in stream_v3_fakes becomes a coroutine awaited at every call site. * test: migrate command and model/middleware tests to native async Convert run_async() wrappers (import, alias, and fixture forms) to plain 'async def test_*'. Multi-call tests merge onto one loop as sequential awaits; none asserted on loop identity. * test: migrate TUI, notifier, gateway, and session tests to native async TUI/notifier/gateway files convert run_async wrappers to plain async tests. test_sessions.py's unittest.TestCase classes move to unittest.IsolatedAsyncioTestCase (pytest-asyncio does not await async methods on plain TestCase; converting blindly would have made ~70 tests silently vacuous). Its setUpClass keeps a one-shot asyncio.run() since IsolatedAsyncioTestCase has no async class-level hook. TestLoadingWidget in test_tui_widgets.py drops its TestCase base for the same reason. * test: replace direct asyncio.run() calls with native async tests Convert tests that called asyncio.run() (directly or via a local _run helper) to plain 'async def test_*'; delete the local helpers. * test: drop undeclared anyio markers and delete run_async helper The @pytest.mark.anyio tests relied on anyio being a transitive dep of httpx; auto-mode pytest-asyncio collects them natively. run_async() and its fixture are unreferenced after the migration, so remove them — pytest-asyncio's per-test loop teardown covers the pending-task cancellation the helper existed for (verified: full suite runs with no 'Event loop is closed' errors or destroyed-task warnings). * test: add autouse fixture for watcher cleanup * refactor: remove redundant hasattr calls * refactor: extract shared HITL/ask_user interaction grammar Extract prompt/question formatting, the reply grammar (approval letters, ask_user choice letters + the 'Other' sub-flow, stop-commands), the ApprovalPolicy (config auto-approve rule + session registry + session-key derivation), per-flow timeout constants, and the bilingual feedback strings into channels/interaction.py. Both drivers now point at the shared functions: this reverses cli/channel.py's imports of consumer privates and closes the /stop drift at the parsing layer (serve-mode ask_user now checks stop-commands before parsing an answer, matching the CLI path). * refactor: add interaction engine + registry; port InboundConsumer Introduce InteractionIO (transport adapter Protocol), PendingReplyRegistry (one asyncio-based reply router per process), and the engine coroutines resolve_ask_user / resolve_approval in channels/interaction.py. Port InboundConsumer onto them: a _ConsumerIO adapter over bus.publish_outbound + the registry, one ApprovalPolicy replacing the config/session auto-approve checks, and a single reply-interception point (registry.try_resolve) replacing the parallel ask_user/HITL pending dicts. _resolve_ask_user and the approval section of _stream_with_hitl are now thin engine calls. Behavior unification (serve mode): an unrecognized HITL reply now declines with the shared 'Unrecognized reply' notice instead of rejecting-and-refeeding as a fresh turn, and /stop mid-approval cancels cleanly — both via the shared parser. * refactor: port CLI channel bridge onto the interaction engine Replace the ~250-line parallel bodies of channel_ask_user_prompt / channel_hitl_prompt with thin bridges that run resolve_ask_user / resolve_approval on the bus loop via run_coroutine_threadsafe(...).result() (outer = engine per-flow timeout + slack, so the engine's own timeout fires first). The 15s send timeout moves into the _BridgeIO adapter. Delete the _pending_hitl / _hitl_lock / _hitl_auto_approve module globals and the _register_hitl_wait / _try_set_hitl_reply / _pop_hitl_reply helpers, absorbed by one bus-loop PendingReplyRegistry + one ApprovalPolicy. The bus consumer feeds the registry via try_resolve ahead of normal enqueue. * refactor: restore serve-mode refeed for unrecognized HITL replies Gate-review fix: the engine no longer decides transport policy for unparseable approval replies. resolve_approval now returns an ApprovalOutcome carrying unrecognized_reply (raw text) when parsing fails, sending no feedback itself; recognized reject keeps the sharedi rejection message. Consumer driver (serve mode) restores the pre-engine semantics: an unrecognized reply rejects the pending action, confirms with the rejection message, and the text is re-dispatched as a NEW agent turn — _stream_with_hitl returns the captured text and _handle_message starts the refeed turn only after the current one has released the chat lock (old fall-through ordering). CLI bridge keeps its old no-refeed path byte-for-byte: 'Unrecognized reply. Action rejected.' and decline. Tests: serve refeed pinned end-to-end (prompt → unrecognized text → rejection feedback → text reaches the stream path as a new turn), CLI no-refeed pinned (notice sent, nothing enqueued), engine test updated to assert the outcome struct with no engine-side feedback. * refactor: polish the interaction engine surface - English feedback strings (Approved / Rejected / auto-approving) - drop the consumer's backwards-compatible re-exports and both modules' private timeout aliases; callers use the canonical interaction names - replace byte-for-byte prompt goldens with structural format tests and assert feedback via the shared constants instead of string literals - strip audit/design shorthand (R1/R2/G3, stage numbers) from comments * fix: propagate pending reply task cancellation * fix: preserve reply context when refeeding HITL replies * fix: honor HITL session grants without bus loop * fix: bound bridge waits by send latency * fix: handle empty ask_user replies explicitly * chore: remove stale interaction helpers * fix: harden interaction engine reply edge cases Review follow-ups on the interaction engine: - normalize ask_user choices before .get(): the tool args come from model JSON and only presence is validated, so plain-string choices must render and parse instead of crashing the turn - treat only None as an approval timeout, so an empty/media-only reply flows through the unrecognized path and serve mode refeeds it with its preserved context - intercept prompt replies before _get_thread_id so a consumed reply cannot create an orphan graph thread or touch the sender-session LRU - close engine coroutines the bridge failed to schedule (no bus loop / scheduling error) to avoid never-awaited warnings - clear pending-response and channel-request state in the bridge test fixture --------- Co-authored-by: Xi Zhang <106144707+X-iZhang@users.noreply.github.com>
630 lines
20 KiB
Python
630 lines
20 KiB
Python
"""Tests for bus-mode queue bridge (_bus_inbound_consumer).
|
|
|
|
The consumer no longer calls the agent directly. Instead it enqueues a
|
|
``ChannelMessage`` on a thread-safe queue and waits for the main CLI
|
|
thread to set a response via ``_set_channel_response()``.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from EvoScientist.channels.bus.events import InboundMessage
|
|
from EvoScientist.channels.bus.message_bus import MessageBus
|
|
from EvoScientist.channels.channel_manager import ChannelManager
|
|
from tests.fakes import QueueFakeChannel as FakeChannel
|
|
|
|
|
|
def _drain_queue(q):
|
|
"""Drain a queue.Queue before a test to avoid cross-test leaks."""
|
|
while not q.empty():
|
|
try:
|
|
q.get_nowait()
|
|
except Exception:
|
|
break
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_channel_state():
|
|
"""Reset shared channel bridge state before and after each test."""
|
|
from EvoScientist.cli import channel as channel_mod
|
|
from EvoScientist.cli.channel import _message_queue
|
|
from EvoScientist.stream import display as display_mod
|
|
|
|
def _reset() -> None:
|
|
_drain_queue(_message_queue)
|
|
with channel_mod._response_lock:
|
|
channel_mod._pending_responses.clear()
|
|
with channel_mod._channel_request_lock:
|
|
channel_mod._channel_requests.clear()
|
|
channel_mod._session_requests.clear()
|
|
channel_mod._cancelled_channel_messages.clear()
|
|
channel_mod._reply_registry.clear()
|
|
channel_mod._approval_policy.clear_sessions()
|
|
with display_mod._stream_cancel_lock:
|
|
display_mod._stream_cancel_event.clear()
|
|
display_mod._stream_cancel_events.clear()
|
|
display_mod._stream_cancel_events[
|
|
display_mod._DEFAULT_STREAM_CANCEL_SCOPE
|
|
] = display_mod._stream_cancel_event
|
|
|
|
_reset()
|
|
yield
|
|
_reset()
|
|
|
|
|
|
class TestBusInboundConsumer:
|
|
"""Test the _bus_inbound_consumer queue bridge."""
|
|
|
|
async def test_processes_inbound_and_publishes_outbound(self):
|
|
"""InboundMessage -> queue -> response -> OutboundMessage flow."""
|
|
from EvoScientist.cli.channel import (
|
|
_bus_inbound_consumer,
|
|
_message_queue,
|
|
_set_channel_response,
|
|
)
|
|
|
|
_drain_queue(_message_queue)
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
consumer = asyncio.create_task(_bus_inbound_consumer(bus, manager))
|
|
|
|
await bus.publish_inbound(
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="hello agent",
|
|
)
|
|
)
|
|
|
|
# Wait for consumer to enqueue the message
|
|
for _ in range(20):
|
|
if not _message_queue.empty():
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
msg = _message_queue.get_nowait()
|
|
assert msg.content == "hello agent"
|
|
assert msg.sender == "user1"
|
|
assert msg.channel_type == "fake"
|
|
|
|
# Simulate main-thread response
|
|
_set_channel_response(msg.msg_id, "Reply to: hello agent")
|
|
|
|
outbound = await asyncio.wait_for(
|
|
bus.consume_outbound(),
|
|
timeout=2.0,
|
|
)
|
|
assert outbound.channel == "fake"
|
|
assert outbound.chat_id == "chat1"
|
|
assert "Reply to: hello agent" in outbound.content
|
|
|
|
consumer.cancel()
|
|
try:
|
|
await consumer
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
async def test_no_response_fallback(self):
|
|
"""Empty response is replaced with 'No response' fallback."""
|
|
from EvoScientist.cli.channel import (
|
|
_bus_inbound_consumer,
|
|
_message_queue,
|
|
_set_channel_response,
|
|
)
|
|
|
|
_drain_queue(_message_queue)
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
consumer = asyncio.create_task(_bus_inbound_consumer(bus, manager))
|
|
|
|
await bus.publish_inbound(
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="test",
|
|
)
|
|
)
|
|
|
|
for _ in range(20):
|
|
if not _message_queue.empty():
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
msg = _message_queue.get_nowait()
|
|
# Set empty response — falsy, so consumer falls back to "No response"
|
|
_set_channel_response(msg.msg_id, "")
|
|
|
|
outbound = await asyncio.wait_for(
|
|
bus.consume_outbound(),
|
|
timeout=2.0,
|
|
)
|
|
assert outbound.content == "No response"
|
|
|
|
consumer.cancel()
|
|
try:
|
|
await consumer
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
async def test_late_response_after_timeout_still_publishes(self, monkeypatch):
|
|
"""A response that arrives after the bridge timeout is still forwarded."""
|
|
from EvoScientist.cli import channel as channel_mod
|
|
from EvoScientist.cli.channel import (
|
|
_bus_inbound_consumer,
|
|
_message_queue,
|
|
_set_channel_response,
|
|
)
|
|
|
|
monkeypatch.setattr(channel_mod, "_RESPONSE_TIMEOUT", 0.05)
|
|
monkeypatch.setattr(channel_mod, "_LATE_RESPONSE_TIMEOUT", 1.0)
|
|
|
|
_drain_queue(_message_queue)
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
consumer = asyncio.create_task(_bus_inbound_consumer(bus, manager))
|
|
|
|
await bus.publish_inbound(
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="slow request",
|
|
message_id="msg-123",
|
|
)
|
|
)
|
|
|
|
for _ in range(20):
|
|
if not _message_queue.empty():
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
msg = _message_queue.get_nowait()
|
|
|
|
notice = await asyncio.wait_for(
|
|
bus.consume_outbound(),
|
|
timeout=1.0,
|
|
)
|
|
assert "Still working on it" in notice.content
|
|
assert notice.reply_to == "msg-123"
|
|
|
|
_set_channel_response(msg.msg_id, "final answer")
|
|
|
|
outbound = await asyncio.wait_for(
|
|
bus.consume_outbound(),
|
|
timeout=1.0,
|
|
)
|
|
assert outbound.content == "final answer"
|
|
assert outbound.reply_to == "msg-123"
|
|
|
|
consumer.cancel()
|
|
try:
|
|
await consumer
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
async def test_late_timeout_keeps_active_request_cancellable(self, monkeypatch):
|
|
"""Late timeout must not discard an active request's cancel scope."""
|
|
from EvoScientist.cli import channel as channel_mod
|
|
from EvoScientist.cli.channel import (
|
|
_channel_message_cancel_scope,
|
|
_channel_request_state,
|
|
_claim_channel_request,
|
|
_handle_bus_message,
|
|
_message_queue,
|
|
)
|
|
from EvoScientist.stream import display as display_mod
|
|
|
|
monkeypatch.setattr(channel_mod, "_RESPONSE_TIMEOUT", 0.05)
|
|
monkeypatch.setattr(channel_mod, "_LATE_RESPONSE_TIMEOUT", 0.05)
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
task = asyncio.create_task(
|
|
_handle_bus_message(
|
|
bus,
|
|
manager,
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="still running",
|
|
message_id="msg-active",
|
|
),
|
|
)
|
|
)
|
|
|
|
queued = None
|
|
for _ in range(20):
|
|
with _message_queue.mutex:
|
|
queued = _message_queue.queue[0] if _message_queue.queue else None
|
|
if queued is not None:
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert queued is not None
|
|
assert _claim_channel_request(queued) is True
|
|
|
|
notice = await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
|
|
assert "Still working on it" in notice.content
|
|
|
|
await task
|
|
|
|
assert _channel_request_state(queued.msg_id) == "active"
|
|
cancel_scope = _channel_message_cancel_scope(queued)
|
|
assert not display_mod.is_stream_cancel_requested(cancel_scope)
|
|
|
|
await _handle_bus_message(
|
|
bus,
|
|
manager,
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="/stop",
|
|
message_id="msg-stop-active",
|
|
),
|
|
)
|
|
|
|
ack = await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
|
|
assert ack.content == "Stopped."
|
|
assert ack.reply_to == "msg-stop-active"
|
|
assert display_mod.is_stream_cancel_requested(cancel_scope)
|
|
|
|
async def test_cancelled_wait_cleans_pending_response(self):
|
|
"""Cancelling a pending bus message should not leak its response slot."""
|
|
from EvoScientist.cli import channel as channel_mod
|
|
from EvoScientist.cli.channel import _handle_bus_message, _message_queue
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
task = asyncio.create_task(
|
|
_handle_bus_message(
|
|
bus,
|
|
manager,
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="cancel me",
|
|
),
|
|
)
|
|
)
|
|
|
|
for _ in range(20):
|
|
if not _message_queue.empty():
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
queued = _message_queue.get_nowait()
|
|
with channel_mod._response_lock:
|
|
assert queued.msg_id in channel_mod._pending_responses
|
|
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
|
|
with channel_mod._response_lock:
|
|
assert queued.msg_id not in channel_mod._pending_responses
|
|
|
|
async def test_consumer_shutdown_cleans_pending_response(self):
|
|
"""Stopping the consumer should cancel late waits and clear state."""
|
|
from EvoScientist.cli import channel as channel_mod
|
|
from EvoScientist.cli.channel import _bus_inbound_consumer, _message_queue
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
consumer = asyncio.create_task(_bus_inbound_consumer(bus, manager))
|
|
|
|
await bus.publish_inbound(
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="slow shutdown",
|
|
)
|
|
)
|
|
|
|
for _ in range(20):
|
|
if not _message_queue.empty():
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
queued = _message_queue.get_nowait()
|
|
with channel_mod._response_lock:
|
|
assert queued.msg_id in channel_mod._pending_responses
|
|
|
|
consumer.cancel()
|
|
await consumer
|
|
|
|
with channel_mod._response_lock:
|
|
assert queued.msg_id not in channel_mod._pending_responses
|
|
|
|
async def test_stop_during_hitl_wait_releases_wait_and_acks(self):
|
|
"""`/stop` should wake a pending interaction wait and publish an ack.
|
|
|
|
The bus consumer delivers ``/stop`` into the reply registry (so the
|
|
blocking engine unwinds) AND acks with "Stopped." — the registry
|
|
interception sits ahead of normal enqueue, so the message never
|
|
becomes a fresh agent turn.
|
|
"""
|
|
from EvoScientist.cli import channel as channel_mod
|
|
from EvoScientist.cli.channel import _bus_inbound_consumer, _message_queue
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
# Simulate a HITL/ask_user prompt waiting for this chat's reply.
|
|
reply_fut = asyncio.ensure_future(
|
|
channel_mod._reply_registry.wait("fake:chat1", timeout=5.0)
|
|
)
|
|
await asyncio.sleep(0.01) # let the wait register
|
|
|
|
consumer = asyncio.create_task(_bus_inbound_consumer(bus, manager))
|
|
|
|
await bus.publish_inbound(
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="/stop",
|
|
message_id="m-stop-1",
|
|
)
|
|
)
|
|
|
|
# The pending wait receives "/stop" (engine will treat it as cancel).
|
|
released = await asyncio.wait_for(reply_fut, timeout=2.0)
|
|
assert released == "/stop"
|
|
|
|
outbound = await asyncio.wait_for(bus.consume_outbound(), timeout=2.0)
|
|
assert outbound.content == "Stopped."
|
|
assert outbound.reply_to == "m-stop-1"
|
|
assert _message_queue.empty()
|
|
|
|
consumer.cancel()
|
|
try:
|
|
await consumer
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
async def test_stop_cancels_queued_request_before_main_thread_processes_it(self):
|
|
"""`/stop` should cancel a queued request instead of only acking."""
|
|
from EvoScientist.cli import channel as channel_mod
|
|
from EvoScientist.cli.channel import (
|
|
_claim_or_complete_channel_request,
|
|
_handle_bus_message,
|
|
_message_queue,
|
|
)
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
task = asyncio.create_task(
|
|
_handle_bus_message(
|
|
bus,
|
|
manager,
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="please work",
|
|
message_id="m-work-1",
|
|
),
|
|
)
|
|
)
|
|
|
|
queued = None
|
|
for _ in range(20):
|
|
with _message_queue.mutex:
|
|
queued = _message_queue.queue[0] if _message_queue.queue else None
|
|
if queued is not None:
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
assert queued is not None
|
|
with channel_mod._response_lock:
|
|
assert queued.msg_id in channel_mod._pending_responses
|
|
|
|
await _handle_bus_message(
|
|
bus,
|
|
manager,
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="/stop",
|
|
message_id="m-stop-2",
|
|
),
|
|
)
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
|
|
skipped = _message_queue.get_nowait()
|
|
assert skipped.msg_id == queued.msg_id
|
|
assert _claim_or_complete_channel_request(skipped) is False
|
|
|
|
with channel_mod._response_lock:
|
|
assert queued.msg_id not in channel_mod._pending_responses
|
|
with channel_mod._channel_request_lock:
|
|
assert queued.msg_id not in channel_mod._channel_requests
|
|
assert queued.msg_id not in channel_mod._cancelled_channel_messages
|
|
assert "fake:chat1" not in channel_mod._session_requests
|
|
|
|
outbound = await asyncio.wait_for(bus.consume_outbound(), timeout=2.0)
|
|
assert outbound.content == "Stopped."
|
|
assert outbound.reply_to == "m-stop-2"
|
|
with pytest.raises(asyncio.TimeoutError):
|
|
await asyncio.wait_for(bus.consume_outbound(), timeout=0.2)
|
|
|
|
async def test_stop_leaves_resolved_response_available_for_delivery(self):
|
|
"""`/stop` must not steal a response whose waiter already resolved."""
|
|
from EvoScientist.cli import channel as channel_mod
|
|
from EvoScientist.cli.channel import (
|
|
ChannelMessage,
|
|
_cancel_channel_session,
|
|
_claim_channel_request,
|
|
_complete_channel_request,
|
|
_enqueue_channel_message,
|
|
_pop_channel_response,
|
|
_set_channel_response,
|
|
)
|
|
|
|
msg = ChannelMessage(
|
|
msg_id="msg-resolved",
|
|
content="already answered",
|
|
sender="user1",
|
|
channel_type="fake",
|
|
metadata={},
|
|
channel_ref=None,
|
|
bus_ref=None,
|
|
chat_id="chat1",
|
|
message_id="m-resolved",
|
|
)
|
|
|
|
waiter = _enqueue_channel_message(msg)
|
|
assert _claim_channel_request(msg) is True
|
|
|
|
_set_channel_response(msg.msg_id, "final answer")
|
|
assert await asyncio.wait_for(asyncio.shield(waiter), timeout=1.0) == (
|
|
"final answer"
|
|
)
|
|
|
|
cancelled_count, active_count = _cancel_channel_session("fake", "chat1")
|
|
assert cancelled_count == 0
|
|
assert active_count == 0
|
|
|
|
with channel_mod._response_lock:
|
|
assert msg.msg_id in channel_mod._pending_responses
|
|
with channel_mod._channel_request_lock:
|
|
assert msg.msg_id not in channel_mod._cancelled_channel_messages
|
|
|
|
assert _pop_channel_response(msg.msg_id) == "final answer"
|
|
_complete_channel_request(msg.msg_id)
|
|
|
|
async def test_message_counting(self):
|
|
"""Messages are counted via record_message."""
|
|
from EvoScientist.cli.channel import (
|
|
_bus_inbound_consumer,
|
|
_message_queue,
|
|
_set_channel_response,
|
|
)
|
|
|
|
_drain_queue(_message_queue)
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
consumer = asyncio.create_task(_bus_inbound_consumer(bus, manager))
|
|
|
|
await bus.publish_inbound(
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="u1",
|
|
chat_id="c1",
|
|
content="test",
|
|
)
|
|
)
|
|
|
|
for _ in range(20):
|
|
if not _message_queue.empty():
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
msg = _message_queue.get_nowait()
|
|
_set_channel_response(msg.msg_id, "ok")
|
|
|
|
await asyncio.wait_for(bus.consume_outbound(), timeout=2.0)
|
|
|
|
assert manager._message_counts["fake"]["received"] == 1
|
|
assert manager._message_counts["fake"]["sent"] == 1
|
|
|
|
consumer.cancel()
|
|
try:
|
|
await consumer
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
async def test_channel_message_carries_metadata(self):
|
|
"""ChannelMessage carries metadata, chat_id, and message_id."""
|
|
from EvoScientist.cli.channel import (
|
|
_bus_inbound_consumer,
|
|
_message_queue,
|
|
_set_channel_response,
|
|
)
|
|
|
|
_drain_queue(_message_queue)
|
|
|
|
bus = MessageBus()
|
|
manager = ChannelManager(bus)
|
|
ch = FakeChannel()
|
|
manager.register(ch)
|
|
|
|
consumer = asyncio.create_task(_bus_inbound_consumer(bus, manager))
|
|
|
|
await bus.publish_inbound(
|
|
InboundMessage(
|
|
channel="fake",
|
|
sender_id="user1",
|
|
chat_id="chat1",
|
|
content="with metadata",
|
|
metadata={"key": "value"},
|
|
message_id="msg-123",
|
|
)
|
|
)
|
|
|
|
for _ in range(20):
|
|
if not _message_queue.empty():
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
|
|
msg = _message_queue.get_nowait()
|
|
assert msg.content == "with metadata"
|
|
assert msg.metadata == {"key": "value"}
|
|
assert msg.chat_id == "chat1"
|
|
assert msg.message_id == "msg-123"
|
|
assert msg.channel_ref is ch
|
|
|
|
_set_channel_response(msg.msg_id, "done")
|
|
|
|
outbound = await asyncio.wait_for(
|
|
bus.consume_outbound(),
|
|
timeout=2.0,
|
|
)
|
|
assert outbound.reply_to == "msg-123"
|
|
|
|
consumer.cancel()
|
|
try:
|
|
await consumer
|
|
except asyncio.CancelledError:
|
|
pass
|