bug: Connection.__init__ ordering bug: _receive_timeout set after recv task is spawned (breaks under asyncio.eager_task_factory)

[jjallaire]

Summary

agent-client-protocol==0.10.0 introduced receive_timeout on Connection, but Connection.__init__ assigns self._receive_timeout after spawning the recv task. On event loops that have asyncio.eager_task_factory installed, _receive_loop runs synchronously inside create_task and tries to read self._receive_timeout before __init__ has set it → AttributeError, connection torn down.

• Affected: 0.10.0
• Last known good: 0.9.0 (feature didn't exist)
• Environments observed: any consumer running on a loop with eager tasks, e.g. anything embedding textual — textual/app.py:2251-2252 calls loop.set_task_factory(asyncio.eager_task_factory). In our case that's inspect-ai's TUI.

Traceback

File ".../acp/connection.py", line 156, in _receive_loop
 line = await asyncio.wait_for(self._reader.readline(), timeout=self._receive_timeout)
AttributeError: 'Connection' object has no attribute '_receive_timeout'

RuntimeWarning: coroutine 'StreamReader.readline' was never awaited

(The readline never awaited warning is a downstream symptom: Python evaluates wait_for's args first, so self._reader.readline() creates an orphaned coroutine before the attribute lookup fails.)

Root cause

acp/connection.py (0.10.0), abbreviated:

def __init__(self, ..., receive_timeout: float | None = None) -> None:
 ...
 if listening:
 self._recv_task = self._tasks.create( # L90: spawn recv task
 self._receive_loop(),
 name="acp.Connection.receive",
 on_error=self._on_receive_error,
 )
 ...
 self._dispatcher.start()
 self._observers = list(observers or [])
 self._receive_timeout = receive_timeout # L107: too late

TaskSupervisor.create (acp/task/supervisor.py:31-44) calls plain asyncio.create_task(coroutine, name=name). Under eager_task_factory (Python 3.12+), that runs the coroutine synchronously up to its first await. The first await in _receive_loop is:

line = await asyncio.wait_for(self._reader.readline(), timeout=self._receive_timeout)

Both arguments are evaluated before wait_for is called. The self._receive_timeout lookup fires before L107 has assigned it. Without the eager factory the bug is latent — __init__ returns first and _receive_timeout is set before the loop schedules the task.

Proposed fix

Set every attribute the receive loop reads before _tasks.create(self._receive_loop(), ...). One-line move:

def __init__(self, ..., receive_timeout: float | None = None) -> None:
 self._handler = handler
 ...
 self._receive_timeout = receive_timeout # <-- move up
 if listening:
 self._recv_task = self._tasks.create(
 self._receive_loop(),
 ...
 )

More general guideline worth applying across the class: any __init__ that spawns tasks should treat the spawn as a publication point — instance state read by those tasks has to be initialized before the spawn, because eager task factory turns the spawn into "run until first await right now."

Reproduction steps

import asyncio
from acp.client.connection import ClientSideConnection
# ... or however you instantiate a Connection in tests

async def main():
 asyncio.get_running_loop().set_task_factory(asyncio.eager_task_factory)
 # construct any Connection; AttributeError raises during init
 ...

asyncio.run(main())

Expected result

No crash

Actual result

Crash with:

File ".../acp/connection.py", line 156, in _receive_loop
 line = await asyncio.wait_for(self._reader.readline(), timeout=self._receive_timeout)
AttributeError: 'Connection' object has no attribute '_receive_timeout'

RuntimeWarning: coroutine 'StreamReader.readline' was never awaited

Versions / environment

SDK version 0.10.0; ACP protocol 1; Python 3.12.10; macOS 26.3.1

Open to submitting a fix?

• I’m willing to open a PR for this bug.