From da17925f31ad262b7bb1f1634f8f659bc0208967 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Mon, 14 Sep 2026 12:44:39 -0400 Subject: [PATCH 1/5] fix(acp): await task/create and event/send instead of backgrounding them Only MESSAGE_SEND was in RPC_SYNC_METHODS, so every other JSON-RPC method took the background branch in base_acp_server._handle_jsonrpc: the handler was dispatched with asyncio.create_task and the server immediately answered {"status": "processing"}. That loses work silently for the two methods that talk to Temporal. A caller doing task/create followed by event/send gets its task id back before TemporalACP.handle_task_create has run start_workflow, so the signal can arrive before the workflow exists and is dropped with "workflow not found". The caller sees success either way, because the response was sent before the handler ran and _process_request only logs the exception. Observed in an eval harness as workflows that start, never receive their payload, and sit on their wait timer until the client's timeout fires. Both handlers are short Temporal RPCs (start_workflow, send_signal) and the caller applies its own timeout, so awaiting them is cheap. The sync path also returns a real JSON-RPC error when the handler raises, which makes the failure visible and retryable. --- src/agentex/protocol/acp.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/agentex/protocol/acp.py b/src/agentex/protocol/acp.py index 7e310cd89..038b8e147 100644 --- a/src/agentex/protocol/acp.py +++ b/src/agentex/protocol/acp.py @@ -127,8 +127,22 @@ class InterruptTaskParams(BaseModel): ) +# Methods whose handler must finish before the RPC responds. +# +# TASK_CREATE and EVENT_SEND are here because dispatching them in the background +# loses work silently. The handlers run against Temporal: TASK_CREATE starts the +# workflow, EVENT_SEND signals it. Backgrounding both means a caller that does +# task/create followed by event/send can have its signal reach Temporal before +# the workflow exists, and the signal is dropped with `workflow not found`. The +# caller cannot tell, because the background path already answered +# {"status": "processing"} and the handler's exception is only logged. +# +# Both handlers are short Temporal RPCs, and the caller applies its own timeout, +# so awaiting them costs little and makes failures visible and retryable. RPC_SYNC_METHODS = [ RPCMethod.MESSAGE_SEND, + RPCMethod.TASK_CREATE, + RPCMethod.EVENT_SEND, ] PARAMS_MODEL_BY_METHOD: dict[RPCMethod, type[BaseModel]] = { From c5820663d2fbacd3adbf8d83c6a2007eff387ec0 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Mon, 14 Sep 2026 21:50:13 -0400 Subject: [PATCH 2/5] test(acp): pin synchronous dispatch for task/create and event/send Two properties the fix depends on, exercised through SyncACP so the test needs no Temporal connection and no network: - the handler finishes before the response is sent, so a caller that sequences task/create then event/send gets the ordering it asked for - a handler that raises produces a JSON-RPC error rather than a success, so the failure is visible and retryable at the client Each async handler yields with `await asyncio.sleep(0)` first, so a regression back to background dispatch loses the ordering race and fails the test. --- tests/test_acp_sync_dispatch.py | 186 ++++++++++++++++++++++++++++++++ 1 file changed, 186 insertions(+) create mode 100644 tests/test_acp_sync_dispatch.py diff --git a/tests/test_acp_sync_dispatch.py b/tests/test_acp_sync_dispatch.py new file mode 100644 index 000000000..7306249a7 --- /dev/null +++ b/tests/test_acp_sync_dispatch.py @@ -0,0 +1,186 @@ +"""Unit tests for synchronous dispatch of ``task/create`` and ``event/send``. + +Background dispatch of these two methods loses work silently. The ACP server +used to answer ``{"status": "processing"}`` before running the handler, so a +caller doing ``task/create`` then ``event/send`` could have its signal reach +Temporal before the workflow existed. The signal was dropped with +``workflow not found``, and the caller saw success either way because the +handler's exception was raised into a background task that only logged it. + +These tests pin the two properties that fix depends on: + +1. The response is not sent until the handler has finished, so a caller that + sequences two calls gets the ordering it asked for. +2. A handler that raises produces a JSON-RPC error, so the failure is visible + and retryable at the client. + +``SyncACP`` is used as the transport because it constructs with no Temporal +connection and no network, and the dispatch under test lives in the shared +``BaseACPServer``. +""" + +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest + +from agentex.types.task import Task +from agentex.types.agent import Agent +from agentex.types.event import Event +from agentex.protocol.acp import ( + RPC_SYNC_METHODS, + RPCMethod, + SendEventParams, + CreateTaskParams, +) +from agentex.lib.sdk.fastacp.impl.sync_acp import SyncACP + + +def _agent() -> Agent: + return Agent( + id="test-agent-456", + name="test-agent", + description="test-agent", + acp_type="async", + created_at="2023-01-01T00:00:00Z", + updated_at="2023-01-01T00:00:00Z", + ) + + +def _task() -> Task: + return Task(id="test-task-123", status="RUNNING") + + +def _event() -> Event: + return Event( + id="test-event-789", + agent_id="test-agent-456", + sequence_id=1, + task_id="test-task-123", + ) + + +class _FakeRequest: + """The slice of ``starlette.requests.Request`` that ``_handle_jsonrpc`` uses.""" + + def __init__(self, payload: dict[str, Any]) -> None: + self._payload = payload + self.headers: dict[str, str] = {} + + async def json(self) -> dict[str, Any]: + return self._payload + + +def _rpc(method: RPCMethod, params: Any) -> dict[str, Any]: + return { + "jsonrpc": "2.0", + "method": method.value, + "params": params.model_dump(mode="json"), + "id": f"{method.value}-test", + } + + +# --------------------------------------------------------------------------- +# Protocol +# --------------------------------------------------------------------------- + + +class TestSyncMethodSet: + @pytest.mark.parametrize( + "method", + [RPCMethod.MESSAGE_SEND, RPCMethod.TASK_CREATE, RPCMethod.EVENT_SEND], + ) + def test_method_is_synchronous(self, method: RPCMethod) -> None: + assert method in RPC_SYNC_METHODS + + +# --------------------------------------------------------------------------- +# Dispatch ordering +# --------------------------------------------------------------------------- + + +class TestHandlerCompletesBeforeResponse: + async def test_task_create_awaits_handler(self) -> None: + acp = SyncACP() + order: list[str] = [] + + @acp.on_task_create + async def handler(params: CreateTaskParams) -> None: + # Yield control so a backgrounded handler would lose the race and + # let the response be sent first. + await asyncio.sleep(0) + order.append("handler") + + response = await acp._handle_jsonrpc( + _FakeRequest(_rpc(RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task()))) + ) + order.append("response") + + assert order == ["handler", "response"] + assert response.error is None + assert response.result != {"status": "processing"} + + async def test_event_send_awaits_handler(self) -> None: + acp = SyncACP() + order: list[str] = [] + + @acp.on_task_event_send + async def handler(params: SendEventParams) -> None: + await asyncio.sleep(0) + order.append("handler") + + response = await acp._handle_jsonrpc( + _FakeRequest( + _rpc( + RPCMethod.EVENT_SEND, + SendEventParams(agent=_agent(), task=_task(), event=_event()), + ) + ) + ) + order.append("response") + + assert order == ["handler", "response"] + assert response.error is None + assert response.result != {"status": "processing"} + + +# --------------------------------------------------------------------------- +# Failure visibility +# --------------------------------------------------------------------------- + + +class TestHandlerFailureReachesCaller: + async def test_task_create_failure_returns_error(self) -> None: + acp = SyncACP() + + @acp.on_task_create + async def handler(params: CreateTaskParams) -> None: + raise RuntimeError("workflow not found for ID: test-task-123") + + response = await acp._handle_jsonrpc( + _FakeRequest(_rpc(RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task()))) + ) + + assert response.error is not None + assert "workflow not found" in response.error["message"] + + async def test_event_send_failure_returns_error(self) -> None: + acp = SyncACP() + + @acp.on_task_event_send + async def handler(params: SendEventParams) -> None: + raise RuntimeError("workflow not found for ID: test-task-123") + + response = await acp._handle_jsonrpc( + _FakeRequest( + _rpc( + RPCMethod.EVENT_SEND, + SendEventParams(agent=_agent(), task=_task(), event=_event()), + ) + ) + ) + + assert response.error is not None + assert "workflow not found" in response.error["message"] From 18556d72fb7a003c6e8f193f91c9b8bb50d90e49 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Mon, 14 Sep 2026 21:55:58 -0400 Subject: [PATCH 3/5] test(acp): read the JSON-RPC error as a model, not a dict JSONRPCResponse.error is typed JSONRPCError | None, so pydantic revalidates the dict the server passes in and the attribute is a model. Indexing it raised TypeError: 'JSONRPCError' object is not subscriptable. _handle_jsonrpc has no return annotation, so pyright could not see result or error either. Route the calls through a helper that asserts the type, which narrows it for the type checker and removes the twelve reportAttributeAccessIssue errors. --- tests/test_acp_sync_dispatch.py | 38 ++++++++++++++------------------- 1 file changed, 16 insertions(+), 22 deletions(-) diff --git a/tests/test_acp_sync_dispatch.py b/tests/test_acp_sync_dispatch.py index 7306249a7..c57ff2986 100644 --- a/tests/test_acp_sync_dispatch.py +++ b/tests/test_acp_sync_dispatch.py @@ -35,6 +35,7 @@ SendEventParams, CreateTaskParams, ) +from agentex.protocol.json_rpc import JSONRPCResponse from agentex.lib.sdk.fastacp.impl.sync_acp import SyncACP @@ -73,6 +74,13 @@ async def json(self) -> dict[str, Any]: return self._payload +async def _dispatch(acp: SyncACP, method: RPCMethod, params: Any) -> JSONRPCResponse: + """Call the JSON-RPC entry point and narrow its untyped return.""" + response = await acp._handle_jsonrpc(_FakeRequest(_rpc(method, params))) # pyright: ignore[reportArgumentType] + assert isinstance(response, JSONRPCResponse) + return response + + def _rpc(method: RPCMethod, params: Any) -> dict[str, Any]: return { "jsonrpc": "2.0", @@ -113,9 +121,7 @@ async def handler(params: CreateTaskParams) -> None: await asyncio.sleep(0) order.append("handler") - response = await acp._handle_jsonrpc( - _FakeRequest(_rpc(RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task()))) - ) + response = await _dispatch(acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task())) order.append("response") assert order == ["handler", "response"] @@ -131,13 +137,8 @@ async def handler(params: SendEventParams) -> None: await asyncio.sleep(0) order.append("handler") - response = await acp._handle_jsonrpc( - _FakeRequest( - _rpc( - RPCMethod.EVENT_SEND, - SendEventParams(agent=_agent(), task=_task(), event=_event()), - ) - ) + response = await _dispatch( + acp, RPCMethod.EVENT_SEND, SendEventParams(agent=_agent(), task=_task(), event=_event()) ) order.append("response") @@ -159,12 +160,10 @@ async def test_task_create_failure_returns_error(self) -> None: async def handler(params: CreateTaskParams) -> None: raise RuntimeError("workflow not found for ID: test-task-123") - response = await acp._handle_jsonrpc( - _FakeRequest(_rpc(RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task()))) - ) + response = await _dispatch(acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task())) assert response.error is not None - assert "workflow not found" in response.error["message"] + assert "workflow not found" in response.error.message async def test_event_send_failure_returns_error(self) -> None: acp = SyncACP() @@ -173,14 +172,9 @@ async def test_event_send_failure_returns_error(self) -> None: async def handler(params: SendEventParams) -> None: raise RuntimeError("workflow not found for ID: test-task-123") - response = await acp._handle_jsonrpc( - _FakeRequest( - _rpc( - RPCMethod.EVENT_SEND, - SendEventParams(agent=_agent(), task=_task(), event=_event()), - ) - ) + response = await _dispatch( + acp, RPCMethod.EVENT_SEND, SendEventParams(agent=_agent(), task=_task(), event=_event()) ) assert response.error is not None - assert "workflow not found" in response.error["message"] + assert "workflow not found" in response.error.message From 0fb33b545706f423174756e108948ff31206f430 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Mon, 14 Sep 2026 21:59:44 -0400 Subject: [PATCH 4/5] test(acp): pass params explicitly to CreateTaskParams CreateTaskParams declares `params: dict[str, Any] | None = Field(None, ...)` with a positional default, which pyright does not read as a default, so it treats the field as required. Pydantic disagrees and reports it optional. Pass it explicitly rather than change the model, which is outside this PR's scope. --- tests/test_acp_sync_dispatch.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/test_acp_sync_dispatch.py b/tests/test_acp_sync_dispatch.py index c57ff2986..bc1f88406 100644 --- a/tests/test_acp_sync_dispatch.py +++ b/tests/test_acp_sync_dispatch.py @@ -121,7 +121,9 @@ async def handler(params: CreateTaskParams) -> None: await asyncio.sleep(0) order.append("handler") - response = await _dispatch(acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task())) + response = await _dispatch( + acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task(), params=None) + ) order.append("response") assert order == ["handler", "response"] @@ -160,7 +162,9 @@ async def test_task_create_failure_returns_error(self) -> None: async def handler(params: CreateTaskParams) -> None: raise RuntimeError("workflow not found for ID: test-task-123") - response = await _dispatch(acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task())) + response = await _dispatch( + acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task(), params=None) + ) assert response.error is not None assert "workflow not found" in response.error.message From 740c93b2832e7952c8490fa971226d286cebb298 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Tue, 15 Sep 2026 16:30:10 -0400 Subject: [PATCH 5/5] fix(acp): keep event/send asynchronous, only task/create becomes sync Making event/send synchronous broke batching. The 080_batch_events tutorial sends events in quick succession and asserts the workflow drains more than one per batch; awaiting each send serialises them, so no batch ever held more than one event and two of its tests failed. Background dispatch there is deliberate design, not an oversight. task/create is the call that actually needed fixing. Its contract is to hand back an id for a task that now exists, and its handler is what starts the workflow, so answering before the handler ran returned an id the server could not yet route to. With it synchronous the workflow is addressable before any event is sent, which is all the race needed. Tests now pin both directions: task/create awaits its handler and surfaces a handler failure as a JSON-RPC error, and event/send still acknowledges without waiting. Not fixed here: a send to a genuinely dead target still fails silently, because the background path logs the handler's exception and there is no status method for a caller to ask. That needs a design which reports failures without serialising sends, and is left for separate work. --- src/agentex/protocol/acp.py | 21 ++--- tests/test_acp_sync_dispatch.py | 132 ++++++++++++++++---------------- 2 files changed, 77 insertions(+), 76 deletions(-) diff --git a/src/agentex/protocol/acp.py b/src/agentex/protocol/acp.py index 038b8e147..917deec48 100644 --- a/src/agentex/protocol/acp.py +++ b/src/agentex/protocol/acp.py @@ -129,20 +129,21 @@ class InterruptTaskParams(BaseModel): # Methods whose handler must finish before the RPC responds. # -# TASK_CREATE and EVENT_SEND are here because dispatching them in the background -# loses work silently. The handlers run against Temporal: TASK_CREATE starts the -# workflow, EVENT_SEND signals it. Backgrounding both means a caller that does -# task/create followed by event/send can have its signal reach Temporal before -# the workflow exists, and the signal is dropped with `workflow not found`. The -# caller cannot tell, because the background path already answered -# {"status": "processing"} and the handler's exception is only logged. +# TASK_CREATE is here because its whole contract is to hand back an id for a +# task that now exists. Its handler starts the Temporal workflow, so answering +# before the handler runs returns an id the server cannot yet route to: a caller +# that follows task/create with event/send can have its signal arrive first and +# be dropped with `workflow not found`. Nothing batches task creations, so there +# is no benefit to trade against, and start_workflow is a short RPC. # -# Both handlers are short Temporal RPCs, and the caller applies its own timeout, -# so awaiting them costs little and makes failures visible and retryable. +# EVENT_SEND deliberately stays asynchronous. Callers send events in quick +# succession and the workflow drains them as a batch; awaiting each send +# serialises them and no batch ever holds more than one event. Once TASK_CREATE +# is synchronous the workflow is addressable before any event is sent, which is +# what the race needed. RPC_SYNC_METHODS = [ RPCMethod.MESSAGE_SEND, RPCMethod.TASK_CREATE, - RPCMethod.EVENT_SEND, ] PARAMS_MODEL_BY_METHOD: dict[RPCMethod, type[BaseModel]] = { diff --git a/tests/test_acp_sync_dispatch.py b/tests/test_acp_sync_dispatch.py index bc1f88406..4ed044b35 100644 --- a/tests/test_acp_sync_dispatch.py +++ b/tests/test_acp_sync_dispatch.py @@ -1,18 +1,21 @@ -"""Unit tests for synchronous dispatch of ``task/create`` and ``event/send``. +"""Unit tests for the dispatch mode of ``task/create`` and ``event/send``. -Background dispatch of these two methods loses work silently. The ACP server -used to answer ``{"status": "processing"}`` before running the handler, so a -caller doing ``task/create`` then ``event/send`` could have its signal reach -Temporal before the workflow existed. The signal was dropped with -``workflow not found``, and the caller saw success either way because the -handler's exception was raised into a background task that only logged it. +``task/create`` hands back an id for a task that now exists, and its handler is +what starts the Temporal workflow. Dispatching it in the background answered +before the handler ran, so the id named a workflow the server could not yet +route to. A caller that followed ``task/create`` with ``event/send`` could have +its signal arrive first and be dropped with ``workflow not found``, and it saw +success either way because the handler's exception was raised into a background +task that only logged it. -These tests pin the two properties that fix depends on: +``event/send`` keeps its background dispatch on purpose: callers send events in +quick succession and the workflow drains them as a batch, which awaiting each +send would serialise. These tests pin both halves so neither is changed by +accident: -1. The response is not sent until the handler has finished, so a caller that - sequences two calls gets the ordering it asked for. -2. A handler that raises produces a JSON-RPC error, so the failure is visible - and retryable at the client. +1. ``task/create`` finishes its handler before responding, and surfaces a + handler failure as a JSON-RPC error. +2. ``event/send`` still acknowledges immediately without waiting. ``SyncACP`` is used as the transport because it constructs with no Temporal connection and no network, and the dispatch under test lives in the shared @@ -63,6 +66,14 @@ def _event() -> Event: ) +def _create_params() -> CreateTaskParams: + return CreateTaskParams(agent=_agent(), task=_task(), params=None) + + +def _event_params() -> SendEventParams: + return SendEventParams(agent=_agent(), task=_task(), event=_event()) + + class _FakeRequest: """The slice of ``starlette.requests.Request`` that ``_handle_jsonrpc`` uses.""" @@ -74,13 +85,6 @@ async def json(self) -> dict[str, Any]: return self._payload -async def _dispatch(acp: SyncACP, method: RPCMethod, params: Any) -> JSONRPCResponse: - """Call the JSON-RPC entry point and narrow its untyped return.""" - response = await acp._handle_jsonrpc(_FakeRequest(_rpc(method, params))) # pyright: ignore[reportArgumentType] - assert isinstance(response, JSONRPCResponse) - return response - - def _rpc(method: RPCMethod, params: Any) -> dict[str, Any]: return { "jsonrpc": "2.0", @@ -90,27 +94,36 @@ def _rpc(method: RPCMethod, params: Any) -> dict[str, Any]: } +async def _dispatch(acp: SyncACP, method: RPCMethod, params: Any) -> JSONRPCResponse: + """Call the JSON-RPC entry point and narrow its untyped return.""" + response = await acp._handle_jsonrpc(_FakeRequest(_rpc(method, params))) # pyright: ignore[reportArgumentType] + assert isinstance(response, JSONRPCResponse) + return response + + # --------------------------------------------------------------------------- -# Protocol +# Which methods are synchronous # --------------------------------------------------------------------------- class TestSyncMethodSet: - @pytest.mark.parametrize( - "method", - [RPCMethod.MESSAGE_SEND, RPCMethod.TASK_CREATE, RPCMethod.EVENT_SEND], - ) + @pytest.mark.parametrize("method", [RPCMethod.MESSAGE_SEND, RPCMethod.TASK_CREATE]) def test_method_is_synchronous(self, method: RPCMethod) -> None: assert method in RPC_SYNC_METHODS + def test_event_send_stays_asynchronous(self) -> None: + # Batching depends on it: callers send events in quick succession and the + # workflow drains them together. Awaiting each send serialises them. + assert RPCMethod.EVENT_SEND not in RPC_SYNC_METHODS + # --------------------------------------------------------------------------- -# Dispatch ordering +# task/create: handler completes before the response # --------------------------------------------------------------------------- -class TestHandlerCompletesBeforeResponse: - async def test_task_create_awaits_handler(self) -> None: +class TestTaskCreateAwaitsHandler: + async def test_handler_runs_before_response(self) -> None: acp = SyncACP() order: list[str] = [] @@ -121,64 +134,51 @@ async def handler(params: CreateTaskParams) -> None: await asyncio.sleep(0) order.append("handler") - response = await _dispatch( - acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task(), params=None) - ) - order.append("response") - - assert order == ["handler", "response"] - assert response.error is None - assert response.result != {"status": "processing"} - - async def test_event_send_awaits_handler(self) -> None: - acp = SyncACP() - order: list[str] = [] - - @acp.on_task_event_send - async def handler(params: SendEventParams) -> None: - await asyncio.sleep(0) - order.append("handler") - - response = await _dispatch( - acp, RPCMethod.EVENT_SEND, SendEventParams(agent=_agent(), task=_task(), event=_event()) - ) + response = await _dispatch(acp, RPCMethod.TASK_CREATE, _create_params()) order.append("response") assert order == ["handler", "response"] assert response.error is None assert response.result != {"status": "processing"} - -# --------------------------------------------------------------------------- -# Failure visibility -# --------------------------------------------------------------------------- - - -class TestHandlerFailureReachesCaller: - async def test_task_create_failure_returns_error(self) -> None: + async def test_handler_failure_returns_error(self) -> None: acp = SyncACP() @acp.on_task_create async def handler(params: CreateTaskParams) -> None: raise RuntimeError("workflow not found for ID: test-task-123") - response = await _dispatch( - acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task(), params=None) - ) + response = await _dispatch(acp, RPCMethod.TASK_CREATE, _create_params()) assert response.error is not None assert "workflow not found" in response.error.message - async def test_event_send_failure_returns_error(self) -> None: + +# --------------------------------------------------------------------------- +# event/send: acknowledges without waiting +# --------------------------------------------------------------------------- + + +class TestEventSendIsBackgrounded: + async def test_response_does_not_wait_for_handler(self) -> None: acp = SyncACP() + started = asyncio.Event() + order: list[str] = [] @acp.on_task_event_send async def handler(params: SendEventParams) -> None: - raise RuntimeError("workflow not found for ID: test-task-123") + await asyncio.sleep(0) + order.append("handler") + started.set() + + response = await _dispatch(acp, RPCMethod.EVENT_SEND, _event_params()) + order.append("response") - response = await _dispatch( - acp, RPCMethod.EVENT_SEND, SendEventParams(agent=_agent(), task=_task(), event=_event()) - ) + # The acknowledgment comes back before the handler has run, which is what + # lets a caller enqueue several events without waiting on each one. + assert order == ["response"] + assert response.result == {"status": "processing"} - assert response.error is not None - assert "workflow not found" in response.error.message + # The handler still runs, just afterwards. + await asyncio.wait_for(started.wait(), timeout=5) + assert order == ["response", "handler"]