test: fill A2A adapter unit test gaps

src/band/integrations/a2a/adapter.py

@@ -37,6 +37,13 @@
 
 logger = logging.getLogger(__name__)
 
+# httpx's read timeout resets on every chunk received, so this bounds the gap
+# between SSE events, not the turn as a whole. Generous enough for the
+# multi-second silences of a live LLM call or tool loop; still finite, so a
+# peer that accepts the connection and then hangs eventually fails the turn
+# instead of blocking the room forever.
+_SSE_READ_TIMEOUT_S = 300.0
+
 
 class A2AAdapter(SimpleAdapter[A2ASessionState]):
     """Adapter that forwards messages to a remote A2A agent.
@@ -107,7 +114,14 @@ async def on_started(self, agent_name: str, agent_description: str) -> None:
 
         headers = self.auth.to_headers() if self.auth else {}
 
-        self._http_client = httpx.AsyncClient(headers=headers)
+        # httpx's default 5s read timeout fires on the normal, multi-second
+        # gap between SSE events during a real remote turn (a live LLM call,
+        # a tool loop) -- not a hang. Use a generous bound instead of the
+        # default so a genuinely dead peer still fails promptly.
+        self._http_client = httpx.AsyncClient(
+            headers=headers,
+            timeout=httpx.Timeout(10.0, read=_SSE_READ_TIMEOUT_S),
+        )
         factory = ClientFactory(
             ClientConfig(streaming=self.streaming, httpx_client=self._http_client)
         )
@@ -319,12 +333,14 @@ async def on_cleanup(self, room_id: str) -> None:
 
     async def cleanup_all(self) -> None:
         """Close the owned A2A client and its HTTP transport."""
-        if self._client is not None:
-            await self._client.close()
-            self._client = None
-        if self._http_client is not None:
-            await self._http_client.aclose()
-            self._http_client = None
+        client, self._client = self._client, None
+        http_client, self._http_client = self._http_client, None
+        try:
+            if client is not None:
+                await client.close()
+        finally:
+            if http_client is not None:
+                await http_client.aclose()
 
     async def _emit_task_event(
         self, tools: AgentToolsProtocol, task: Task, state: TaskState

src/band/integrations/a2a/gateway/adapter.py

@@ -344,6 +344,7 @@ async def _execute_a2a(
                 request.context_id,
                 request.pending.task.id,
             )
+            await request.pending.fail("A2A request failed")
             raise
         else:
             if completed:

src/band/integrations/a2a/gateway/server.py

@@ -16,26 +16,39 @@
 from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
 from a2a.compat.v0_3.conversions import to_compat_agent_card
 from a2a.utils.constants import PROTOCOL_VERSION_0_3, PROTOCOL_VERSION_CURRENT
+from sse_starlette.sse import AppStatus
 from starlette.applications import Starlette
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 from starlette.routing import BaseRoute, Route
 
+from band.integrations.uvicorn_server import wait_until_started
 from band_rest import Peer
 
 logger = logging.getLogger(__name__)
 
 ExecutorFactory = Callable[[str], AgentExecutor]
 
-# uvicorn's own default (None) waits forever for existing connections to close
-# on stop() -- and a live message:stream SSE response has no other way to end
-# on its own. sse_starlette normally closes it cooperatively on shutdown, but
-# that mechanism is a process-global switch any co-located
-# band.integrations.mcp.local_server permanently disables (see that module's
-# AppStatus.disable_automatic_graceful_drain() call) -- so this bound is the
-# only thing that keeps stop() from hanging once that happens.
+# sse_starlette's shutdown watcher polls whichever uvicorn.Server owns the
+# process's SIGTERM slot and promotes its should_exit to the process-global
+# AppStatus.should_exit -- so a GatewayServer.stop() (which sets should_exit
+# directly, not via a signal) can poison every later GatewayServer's SSE
+# streams in the same process. band.integrations.mcp.local_server disables
+# this for the same reason; calling it here too avoids depending on that
+# import (idempotent, process-wide).
+AppStatus.disable_automatic_graceful_drain()
+
+# The automatic drain above is disabled, so a live message:stream response
+# has no other way to end on stop() -- uvicorn's own default (None) would
+# wait forever for it to close on its own.
 SERVER_STOP_TIMEOUT_S = 5
 
+# How long start() waits for uvicorn to report ready before giving up. Without
+# this wait, start() returns as soon as serve() is merely scheduled -- a caller
+# (e.g. an A2A client dialing in immediately after on_started()) can then race
+# a socket that isn't listening yet.
+SERVER_START_TIMEOUT_S = 5
+
 # The REST endpoints the gateway serves per peer: the messaging binding and
 # the compat card. The upstream factory also returns task read/cancel/list
 # and push-config routes — an unauthenticated window into past conversations.
@@ -241,7 +254,7 @@ async def start(self) -> None:
         import uvicorn
 
         self._app = self._build_app()
-        self._uvicorn = uvicorn.Server(
+        server = uvicorn.Server(
             uvicorn.Config(
                 self._app,
                 host="0.0.0.0",
@@ -250,7 +263,18 @@ async def start(self) -> None:
                 timeout_graceful_shutdown=SERVER_STOP_TIMEOUT_S,
             )
         )
-        self._server_task = asyncio.create_task(self._uvicorn.serve())
+        server_task = asyncio.create_task(server.serve())
+        self._uvicorn = server
+        self._server_task = server_task
+        try:
+            await wait_until_started(
+                server, server_task, timeout_s=SERVER_START_TIMEOUT_S
+            )
+        except BaseException:
+            # A failed/timed-out startup still leaves server_task running and
+            # the socket bound; stop() already unwinds both.
+            await self.stop()
+            raise
         logger.info(
             "Starting A2A Gateway server on port %d with %d peers",
             self.port,

src/band/integrations/mcp/local_server.py

@@ -35,6 +35,7 @@
     build_engine,
     validate_unique_tool_names,
 )
+from band.integrations.uvicorn_server import wait_until_started
 
 logger = logging.getLogger(__name__)
 
@@ -69,12 +70,10 @@
 # something threaded through LocalMCPServer's API.
 #
 # Cost of that global scope: any *other* sse_starlette consumer in the same
-# process -- e.g. the A2A gateway's own message:stream responses
-# (src/band/integrations/a2a/gateway/server.py) -- loses this same
-# cooperative-drain signal too, permanently, the moment this module is
-# imported anywhere in the process. That server's own uvicorn.Config sets
-# timeout_graceful_shutdown precisely so its stop() still bounds how long it
-# waits on a live stream, rather than relying on the now-disabled signal.
+# process loses this same signal too, the moment this module is imported.
+# The A2A gateway (src/band/integrations/a2a/gateway/server.py) hits the
+# identical bug independently and disables this itself; the call here is
+# redundant with that one but harmless (idempotent, process-wide).
 AppStatus.disable_automatic_graceful_drain()
 
 
@@ -238,7 +237,9 @@ async def start(self) -> None:
                 self._uvicorn_server = uvicorn_server
                 self._serve_task = serve_task
 
-                await self._wait_until_started()
+                await wait_until_started(
+                    uvicorn_server, serve_task, timeout_s=SERVER_START_TIMEOUT_S
+                )
             except Exception:
                 await self._stop_locked()
                 raise
@@ -345,15 +346,3 @@ def _reserve_socket(self) -> tuple[socket.socket, int]:
             "Could not find a free localhost MCP port in range "
             f"{self._port_min}-{self._port_max}"
         ) from last_error
-
-    async def _wait_until_started(self) -> None:
-        if self._serve_task is None or self._uvicorn_server is None:
-            raise RuntimeError("Local MCP server task not initialized")
-
-        deadline = asyncio.get_running_loop().time() + SERVER_START_TIMEOUT_S
-        while not self._uvicorn_server.started:
-            if self._serve_task.done():
-                await self._serve_task
-            if asyncio.get_running_loop().time() >= deadline:
-                raise TimeoutError("Timed out waiting for local MCP server startup")
-            await asyncio.sleep(0.05)

src/band/integrations/uvicorn_server.py

@@ -0,0 +1,42 @@
+"""Shared startup wait for integrations that embed their own uvicorn server.
+
+Used by both ``band.integrations.mcp.local_server`` and
+``band.integrations.a2a.gateway.server`` (and the A2A baseline test fixture)
+so the one correctness-sensitive piece -- surfacing a serve task that died
+before the server ever came up -- is fixed in one place.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import uvicorn
+
+POLL_INTERVAL_S = 0.05
+
+
+async def wait_until_started(
+    server: uvicorn.Server,
+    serve_task: asyncio.Task[object],
+    *,
+    timeout_s: float,
+) -> None:
+    """Block until ``server`` reports ready.
+
+    ``serve_task`` only returns once the server stops, so readiness is
+    polled via ``server.started`` instead of awaiting the task directly.
+    But a task that dies before ever setting ``started`` -- a port already
+    in use, a bad TLS config -- would otherwise busy-wait the full
+    ``timeout_s`` and then raise a generic timeout instead of the real
+    failure; checking ``serve_task.done()`` on every pass re-raises that
+    failure immediately.
+    """
+    deadline = asyncio.get_running_loop().time() + timeout_s
+    while not server.started:
+        if serve_task.done():
+            await serve_task
+        if asyncio.get_running_loop().time() >= deadline:
+            raise TimeoutError(
+                f"uvicorn server did not report ready within {timeout_s}s"
+            )
+        await asyncio.sleep(POLL_INTERVAL_S)

tests/e2e/baseline/README.md

@@ -239,7 +239,7 @@ These three are why the toolkit is shaped the way it is — keep them when exten
 | `smoke/matrix/` | runs across the adapter matrix: `test_adapter_matrix.py`, `test_capability_matrix.py` (memory store + recall), `test_context_recall.py` (in-session + rejoin), `test_rehydration_offline.py` / `test_rehydration_partial.py` (cold-boot / partial-reboot `/context` recall), `test_rehydration_cross_framework.py` (a different-framework `peer=` authors, A rehydrates), `test_room_isolation.py`, `test_noisy_room.py`, `test_tool_round_trip.py` (custom-tool subgroup) |
 | `smoke/behavior/` | platform/transport + scenario behavior: `test_delivery_status.py`, `test_processing_barrier.py`, `test_isolation.py`, `test_agent_scenarios.py` |
 | `smoke/inspection/` | `capture.*` observation worked-examples: `test_tool_calls.py`, `test_events.py`, `test_memory.py`, `test_usage.py` (the `Emit.USAGE` fan), plus `test_next_actionable_semantics.py` (a platform `/next` invariant, `@lane`-pinned since it runs no adapter) |
-| `smoke/adapters/` | adapter-specific showcases: `test_agno.py`, `test_copilot_acp.py`, `test_copilot_sdk.py`, `test_crewai.py`, `test_letta.py`, `test_opencode.py`, `test_parlant.py` |
+| `smoke/adapters/` | adapter-specific showcases: `test_a2a.py`, `test_a2a_gateway.py`, `test_a2a_roundtrip.py` (+ their shared `a2aServer.py` fixture -- a scripted, non-Band A2A counterparty), `test_agno.py`, `test_copilot_acp.py`, `test_copilot_sdk.py`, `test_crewai.py`, `test_letta.py`, `test_opencode.py`, `test_parlant.py` |
 
 The `toolkit/` modules are pytest-free and reusable anywhere. The package root
 (`settings`, `requires`, `agents`, `conftest`) is the pytest wiring.