Skip to content

Commit 740c93b

Browse files
committed
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.
1 parent 0fb33b5 commit 740c93b

2 files changed

Lines changed: 77 additions & 76 deletions

File tree

‎src/agentex/protocol/acp.py‎

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -129,20 +129,21 @@ class InterruptTaskParams(BaseModel):
129129

130130
# Methods whose handler must finish before the RPC responds.
131131
#
132-
# TASK_CREATE and EVENT_SEND are here because dispatching them in the background
133-
# loses work silently. The handlers run against Temporal: TASK_CREATE starts the
134-
# workflow, EVENT_SEND signals it. Backgrounding both means a caller that does
135-
# task/create followed by event/send can have its signal reach Temporal before
136-
# the workflow exists, and the signal is dropped with `workflow not found`. The
137-
# caller cannot tell, because the background path already answered
138-
# {"status": "processing"} and the handler's exception is only logged.
132+
# TASK_CREATE is here because its whole contract is to hand back an id for a
133+
# task that now exists. Its handler starts the Temporal workflow, so answering
134+
# before the handler runs returns an id the server cannot yet route to: a caller
135+
# that follows task/create with event/send can have its signal arrive first and
136+
# be dropped with `workflow not found`. Nothing batches task creations, so there
137+
# is no benefit to trade against, and start_workflow is a short RPC.
139138
#
140-
# Both handlers are short Temporal RPCs, and the caller applies its own timeout,
141-
# so awaiting them costs little and makes failures visible and retryable.
139+
# EVENT_SEND deliberately stays asynchronous. Callers send events in quick
140+
# succession and the workflow drains them as a batch; awaiting each send
141+
# serialises them and no batch ever holds more than one event. Once TASK_CREATE
142+
# is synchronous the workflow is addressable before any event is sent, which is
143+
# what the race needed.
142144
RPC_SYNC_METHODS = [
143145
RPCMethod.MESSAGE_SEND,
144146
RPCMethod.TASK_CREATE,
145-
RPCMethod.EVENT_SEND,
146147
]
147148

148149
PARAMS_MODEL_BY_METHOD: dict[RPCMethod, type[BaseModel]] = {

‎tests/test_acp_sync_dispatch.py‎

Lines changed: 66 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,21 @@
1-
"""Unit tests for synchronous dispatch of ``task/create`` and ``event/send``.
1+
"""Unit tests for the dispatch mode of ``task/create`` and ``event/send``.
22
3-
Background dispatch of these two methods loses work silently. The ACP server
4-
used to answer ``{"status": "processing"}`` before running the handler, so a
5-
caller doing ``task/create`` then ``event/send`` could have its signal reach
6-
Temporal before the workflow existed. The signal was dropped with
7-
``workflow not found``, and the caller saw success either way because the
8-
handler's exception was raised into a background task that only logged it.
3+
``task/create`` hands back an id for a task that now exists, and its handler is
4+
what starts the Temporal workflow. Dispatching it in the background answered
5+
before the handler ran, so the id named a workflow the server could not yet
6+
route to. A caller that followed ``task/create`` with ``event/send`` could have
7+
its signal arrive first and be dropped with ``workflow not found``, and it saw
8+
success either way because the handler's exception was raised into a background
9+
task that only logged it.
910
10-
These tests pin the two properties that fix depends on:
11+
``event/send`` keeps its background dispatch on purpose: callers send events in
12+
quick succession and the workflow drains them as a batch, which awaiting each
13+
send would serialise. These tests pin both halves so neither is changed by
14+
accident:
1115
12-
1. The response is not sent until the handler has finished, so a caller that
13-
sequences two calls gets the ordering it asked for.
14-
2. A handler that raises produces a JSON-RPC error, so the failure is visible
15-
and retryable at the client.
16+
1. ``task/create`` finishes its handler before responding, and surfaces a
17+
handler failure as a JSON-RPC error.
18+
2. ``event/send`` still acknowledges immediately without waiting.
1619
1720
``SyncACP`` is used as the transport because it constructs with no Temporal
1821
connection and no network, and the dispatch under test lives in the shared
@@ -63,6 +66,14 @@ def _event() -> Event:
6366
)
6467

6568

69+
def _create_params() -> CreateTaskParams:
70+
return CreateTaskParams(agent=_agent(), task=_task(), params=None)
71+
72+
73+
def _event_params() -> SendEventParams:
74+
return SendEventParams(agent=_agent(), task=_task(), event=_event())
75+
76+
6677
class _FakeRequest:
6778
"""The slice of ``starlette.requests.Request`` that ``_handle_jsonrpc`` uses."""
6879

@@ -74,13 +85,6 @@ async def json(self) -> dict[str, Any]:
7485
return self._payload
7586

7687

77-
async def _dispatch(acp: SyncACP, method: RPCMethod, params: Any) -> JSONRPCResponse:
78-
"""Call the JSON-RPC entry point and narrow its untyped return."""
79-
response = await acp._handle_jsonrpc(_FakeRequest(_rpc(method, params))) # pyright: ignore[reportArgumentType]
80-
assert isinstance(response, JSONRPCResponse)
81-
return response
82-
83-
8488
def _rpc(method: RPCMethod, params: Any) -> dict[str, Any]:
8589
return {
8690
"jsonrpc": "2.0",
@@ -90,27 +94,36 @@ def _rpc(method: RPCMethod, params: Any) -> dict[str, Any]:
9094
}
9195

9296

97+
async def _dispatch(acp: SyncACP, method: RPCMethod, params: Any) -> JSONRPCResponse:
98+
"""Call the JSON-RPC entry point and narrow its untyped return."""
99+
response = await acp._handle_jsonrpc(_FakeRequest(_rpc(method, params))) # pyright: ignore[reportArgumentType]
100+
assert isinstance(response, JSONRPCResponse)
101+
return response
102+
103+
93104
# ---------------------------------------------------------------------------
94-
# Protocol
105+
# Which methods are synchronous
95106
# ---------------------------------------------------------------------------
96107

97108

98109
class TestSyncMethodSet:
99-
@pytest.mark.parametrize(
100-
"method",
101-
[RPCMethod.MESSAGE_SEND, RPCMethod.TASK_CREATE, RPCMethod.EVENT_SEND],
102-
)
110+
@pytest.mark.parametrize("method", [RPCMethod.MESSAGE_SEND, RPCMethod.TASK_CREATE])
103111
def test_method_is_synchronous(self, method: RPCMethod) -> None:
104112
assert method in RPC_SYNC_METHODS
105113

114+
def test_event_send_stays_asynchronous(self) -> None:
115+
# Batching depends on it: callers send events in quick succession and the
116+
# workflow drains them together. Awaiting each send serialises them.
117+
assert RPCMethod.EVENT_SEND not in RPC_SYNC_METHODS
118+
106119

107120
# ---------------------------------------------------------------------------
108-
# Dispatch ordering
121+
# task/create: handler completes before the response
109122
# ---------------------------------------------------------------------------
110123

111124

112-
class TestHandlerCompletesBeforeResponse:
113-
async def test_task_create_awaits_handler(self) -> None:
125+
class TestTaskCreateAwaitsHandler:
126+
async def test_handler_runs_before_response(self) -> None:
114127
acp = SyncACP()
115128
order: list[str] = []
116129

@@ -121,64 +134,51 @@ async def handler(params: CreateTaskParams) -> None:
121134
await asyncio.sleep(0)
122135
order.append("handler")
123136

124-
response = await _dispatch(
125-
acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task(), params=None)
126-
)
127-
order.append("response")
128-
129-
assert order == ["handler", "response"]
130-
assert response.error is None
131-
assert response.result != {"status": "processing"}
132-
133-
async def test_event_send_awaits_handler(self) -> None:
134-
acp = SyncACP()
135-
order: list[str] = []
136-
137-
@acp.on_task_event_send
138-
async def handler(params: SendEventParams) -> None:
139-
await asyncio.sleep(0)
140-
order.append("handler")
141-
142-
response = await _dispatch(
143-
acp, RPCMethod.EVENT_SEND, SendEventParams(agent=_agent(), task=_task(), event=_event())
144-
)
137+
response = await _dispatch(acp, RPCMethod.TASK_CREATE, _create_params())
145138
order.append("response")
146139

147140
assert order == ["handler", "response"]
148141
assert response.error is None
149142
assert response.result != {"status": "processing"}
150143

151-
152-
# ---------------------------------------------------------------------------
153-
# Failure visibility
154-
# ---------------------------------------------------------------------------
155-
156-
157-
class TestHandlerFailureReachesCaller:
158-
async def test_task_create_failure_returns_error(self) -> None:
144+
async def test_handler_failure_returns_error(self) -> None:
159145
acp = SyncACP()
160146

161147
@acp.on_task_create
162148
async def handler(params: CreateTaskParams) -> None:
163149
raise RuntimeError("workflow not found for ID: test-task-123")
164150

165-
response = await _dispatch(
166-
acp, RPCMethod.TASK_CREATE, CreateTaskParams(agent=_agent(), task=_task(), params=None)
167-
)
151+
response = await _dispatch(acp, RPCMethod.TASK_CREATE, _create_params())
168152

169153
assert response.error is not None
170154
assert "workflow not found" in response.error.message
171155

172-
async def test_event_send_failure_returns_error(self) -> None:
156+
157+
# ---------------------------------------------------------------------------
158+
# event/send: acknowledges without waiting
159+
# ---------------------------------------------------------------------------
160+
161+
162+
class TestEventSendIsBackgrounded:
163+
async def test_response_does_not_wait_for_handler(self) -> None:
173164
acp = SyncACP()
165+
started = asyncio.Event()
166+
order: list[str] = []
174167

175168
@acp.on_task_event_send
176169
async def handler(params: SendEventParams) -> None:
177-
raise RuntimeError("workflow not found for ID: test-task-123")
170+
await asyncio.sleep(0)
171+
order.append("handler")
172+
started.set()
173+
174+
response = await _dispatch(acp, RPCMethod.EVENT_SEND, _event_params())
175+
order.append("response")
178176

179-
response = await _dispatch(
180-
acp, RPCMethod.EVENT_SEND, SendEventParams(agent=_agent(), task=_task(), event=_event())
181-
)
177+
# The acknowledgment comes back before the handler has run, which is what
178+
# lets a caller enqueue several events without waiting on each one.
179+
assert order == ["response"]
180+
assert response.result == {"status": "processing"}
182181

183-
assert response.error is not None
184-
assert "workflow not found" in response.error.message
182+
# The handler still runs, just afterwards.
183+
await asyncio.wait_for(started.wait(), timeout=5)
184+
assert order == ["response", "handler"]

0 commit comments

Comments
 (0)