Gate check on localstack-integration; make docker timeouts asyncio-safe

- Add localstack-integration to check.needs so a live-test failure blocks
  merges, per maintainer decision. Cover the contract in test_ci_workflow.py.
- Bound docker run/stop with subprocess.run(timeout=...) on a worker thread
  instead of cancelling an async subprocess. Cancelling anyio.run_process
  mid-flight leaves a dangling child on the asyncio backend, which surfaced as
  unraisable-exception warnings that failed unrelated asyncio tests. The
  blocking call kills and reaps the child on timeout.

.github/workflows/main.yml

@@ -216,7 +216,7 @@ jobs:
 
   check:
     if: always()
-    needs: [lint, test, test-floor, test-latest, coverage]
+    needs: [lint, test, test-floor, test-latest, coverage, localstack-integration]
     runs-on: ubuntu-latest
     steps:
       - uses: re-actors/alls-green@05ac9388f0aebcb5727afa17fcccfecd6f8ec5fe # v1.2.2

pydantic_ai_harness/localstack/_container.py

@@ -3,10 +3,12 @@
 from __future__ import annotations
 
 import os
+import subprocess
 from collections.abc import Mapping
 from pathlib import Path
 
 import anyio
+import anyio.to_thread
 import httpx
 from typing_extensions import Self
 
@@ -85,18 +87,10 @@ class LocalStackContainer:
         return self._container_id
 
     async def __aenter__(self) -> Self:
-        """Start the container and wait for it to become ready.
-
-        `startup_timeout` bounds the whole sequence, so an unresponsive Docker
-        daemon that hangs `docker run` fails here rather than blocking forever.
-        """
+        """Start the container and wait for it to become ready."""
+        self._container_id = await self._start()
         try:
-            with anyio.fail_after(self._startup_timeout):
-                self._container_id = await self._start()
-                await self._wait_until_ready()
-        except TimeoutError as e:
-            await self._stop()
-            raise LocalStackError(f'LocalStack did not become ready within {self._startup_timeout}s.') from e
+            await self._wait_until_ready()
         except BaseException:
             await self._stop()
             raise
@@ -151,30 +145,53 @@ class LocalStackContainer:
         environment.update(container_environment)
         return environment
 
+    async def _run_docker(self, argv: list[str], environment: Mapping[str, str]) -> subprocess.CompletedProcess[bytes]:
+        """Run a docker command in a worker thread, bounded by `startup_timeout`.
+
+        A blocking `subprocess.run` with a timeout is used instead of an async
+        subprocess because it kills and reaps the child when the timeout fires.
+        Cancelling an async subprocess mid-run can leave a dangling process on
+        the asyncio backend, so an unresponsive daemon would otherwise leak a
+        process (and emit unraisable-exception warnings) instead of failing cleanly.
+        """
+        docker_env = dict(environment)
+
+        def run() -> subprocess.CompletedProcess[bytes]:
+            return subprocess.run(argv, env=docker_env, capture_output=True, timeout=self._startup_timeout)
+
+        return await anyio.to_thread.run_sync(run)
+
     async def _start(self) -> str:
         """Launch the container detached and return its id."""
         container_environment = self._effective_environment()
         try:
-            result = await anyio.run_process(
+            result = await self._run_docker(
                 self._run_argv(container_environment),
-                env=self._docker_environment(container_environment),
-                check=False,
+                self._docker_environment(container_environment),
             )
         except FileNotFoundError as e:
             raise LocalStackError(
                 f'Docker CLI {self._docker_path!r} not found. Install Docker to manage LocalStack.'
             ) from e
+        except subprocess.TimeoutExpired as e:
+            raise LocalStackError(
+                f'Docker did not start the LocalStack container within {self._startup_timeout}s.'
+            ) from e
         if result.returncode != 0:
             stderr = result.stderr.decode('utf-8', errors='replace').strip()
             raise LocalStackError(f'Failed to start LocalStack container: {stderr}')
         return result.stdout.decode('utf-8', errors='replace').strip()
 
     async def _wait_until_ready(self) -> None:
-        """Poll the health endpoint until LocalStack responds; `__aenter__` bounds the wait."""
+        """Poll the health endpoint until LocalStack responds or `startup_timeout` elapses."""
         url = self._readiness_url
-        async with httpx.AsyncClient() as client:
-            while not await self._is_ready(client, url):
-                await anyio.sleep(self._poll_interval)
+        try:
+            with anyio.fail_after(self._startup_timeout):
+                async with httpx.AsyncClient() as client:
+                    while not await self._is_ready(client, url):
+                        await anyio.sleep(self._poll_interval)
+        except TimeoutError as e:
+            raise LocalStackError(f'LocalStack did not become ready within {self._startup_timeout}s.') from e
 
     async def _is_ready(self, client: httpx.AsyncClient, url: str) -> bool:
         """Return True once the health endpoint answers with HTTP 200."""
@@ -192,7 +209,6 @@ class LocalStackContainer:
         self._container_id = None
         with anyio.CancelScope(shield=True):
             try:
-                with anyio.fail_after(self._startup_timeout):
-                    await anyio.run_process([self._docker_path, 'stop', container_id], check=False)
-            except (FileNotFoundError, TimeoutError):
+                await self._run_docker([self._docker_path, 'stop', container_id], self._docker_environment({}))
+            except (FileNotFoundError, subprocess.TimeoutExpired):
                 pass

tests/localstack/test_ci_workflow.py

@@ -33,6 +33,15 @@ def test_localstack_ci_authenticates_with_an_auth_token() -> None:
     assert not any('LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT' in line for line in lines)
 
 
+def test_localstack_ci_gates_the_aggregate_check() -> None:
+    lines = _workflow_lines()
+
+    # The aggregate `check` job must depend on localstack-integration so a live
+    # test failure blocks merges rather than passing silently.
+    needs = next(line for line in lines if line.strip().startswith('needs: [lint, test'))
+    assert 'localstack-integration' in needs
+
+
 def test_localstack_ci_scopes_the_auth_token_to_the_test_step() -> None:
     lines = _workflow_lines()
 

tests/localstack/test_container.py

@@ -71,7 +71,8 @@ def _failing_docker_stub(tmp_path: Path) -> str:
 def _hanging_docker_stub(tmp_path: Path) -> str:
     """A fake `docker` CLI that never returns, standing in for an unresponsive daemon."""
     stub = tmp_path / 'docker-hang'
-    stub.write_text('#!/bin/sh\nsleep 30\n')
+    # `exec` so the shell becomes `sleep`, letting the timeout's kill reap a single process.
+    stub.write_text('#!/bin/sh\nexec sleep 30\n')
     stub.chmod(stub.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
     return str(stub)
 
@@ -243,10 +244,10 @@ class TestLifecycle:
     async def test_readiness_timeout_stops_container(self, tmp_path: Path) -> None:
         docker, log = _docker_stub(tmp_path)
         port = unused_tcp_port()
-        # startup_timeout must clear `docker run` (the stub subprocess) so the deadline
-        # is spent on readiness polling, which never succeeds against the unused port.
-        with pytest.raises(LocalStackError, match='did not become ready within 0.3s'):
-            async with LocalStackContainer(docker_path=docker, host_port=port, startup_timeout=0.3, poll_interval=0.01):
+        # startup_timeout also bounds `docker run` (the stub subprocess); keep it well
+        # above subprocess-spawn time so the readiness poll, not the launch, times out.
+        with pytest.raises(LocalStackError, match='did not become ready within 1.0s'):
+            async with LocalStackContainer(docker_path=docker, host_port=port, startup_timeout=1.0, poll_interval=0.01):
                 pass  # pragma: no cover
         assert 'stop container-abc123' in log.read_text()
 
@@ -261,7 +262,7 @@ class TestLifecycle:
                 pass  # pragma: no cover
 
     async def test_start_timeout_when_docker_hangs(self, tmp_path: Path) -> None:
-        with pytest.raises(LocalStackError, match='did not become ready within 0.05s'):
+        with pytest.raises(LocalStackError, match='Docker did not start the LocalStack container within 0.05s'):
             async with LocalStackContainer(docker_path=_hanging_docker_stub(tmp_path), startup_timeout=0.05):
                 pass  # pragma: no cover
 

tests/localstack/test_localstack.py

@@ -337,12 +337,12 @@ class TestContainerManagement:
 
     async def test_managed_defaults_to_edge_port_when_endpoint_has_no_port(self, tmp_path: Path) -> None:
         docker, log = _docker_stub(tmp_path)
-        with pytest.raises(LocalStackError, match='did not become ready within 0.3s'):
+        with pytest.raises(LocalStackError, match='did not become ready within 1.0s'):
             async with _toolset(
                 endpoint_url='http://localhost',
                 manage_container=True,
                 docker_path=docker,
-                startup_timeout=0.3,
+                startup_timeout=1.0,
             ):
                 pass  # pragma: no cover
         log_text = log.read_text()