diff --git a/.github/workflows/test-suite.yml b/.github/workflows/test-suite.yml
index c3ad08f14..94e493bde 100644
--- a/.github/workflows/test-suite.yml
+++ b/.github/workflows/test-suite.yml
@@ -3,9 +3,9 @@ name: Test Suite
 
 on:
   push:
-    branches: ["master"]
+    branches: ["master", "openpipe/linear-connection-pool-1.0.9"]
   pull_request:
-    branches: ["master"]
+    branches: ["master", "openpipe/linear-connection-pool-1.0.9"]
 
 jobs:
   tests:
diff --git a/httpcore/_async/connection.py b/httpcore/_async/connection.py
index b42581dff..1611e868b 100644
--- a/httpcore/_async/connection.py
+++ b/httpcore/_async/connection.py
@@ -184,6 +184,12 @@ def is_available(self) -> bool:
             )
         return self._connection.is_available()
 
+    def _is_multiplexable(self) -> bool:
+        if self._connection is None:
+            # Preserve speculative HTTP/2 sharing until the protocol is known.
+            return self.is_available()
+        return self._connection._is_multiplexable()
+
     def has_expired(self) -> bool:
         if self._connection is None:
             return self._connect_failed
diff --git a/httpcore/_async/connection_pool.py b/httpcore/_async/connection_pool.py
index afeb2799c..681d35513 100644
--- a/httpcore/_async/connection_pool.py
+++ b/httpcore/_async/connection_pool.py
@@ -4,6 +4,7 @@
 import sys
 import types
 import typing
+from collections import OrderedDict
 
 from .._backends.auto import AutoBackend
 from .._backends.base import SOCKET_OPTION, AsyncNetworkBackend
@@ -119,6 +120,7 @@ def __init__(
         # and the set of connections that are servicing those requests.
         self._connections: list[AsyncConnectionInterface] = []
         self._requests: list[AsyncPoolRequest] = []
+        self._request_connections: dict[AsyncConnectionInterface, int] = {}
 
         # We only mutate the state of the connection pool within an 'optional_thread_lock'
         # context. This holds a threading lock unless we're running in async mode,
@@ -241,7 +243,9 @@ async def handle_async_request(self, request: Request) -> Response:
                     # handle a request, but then become unavailable.
                     #
                     # In this case we clear the connection and try again.
-                    pool_request.clear_connection()
+                    with self._optional_thread_lock:
+                        self._release_request_connection(pool_request)
+                        pool_request.clear_connection()
                 else:
                     break  # pragma: nocover
 
@@ -249,6 +253,7 @@ async def handle_async_request(self, request: Request) -> Response:
             with self._optional_thread_lock:
                 # For any exception or cancellation we remove the request from
                 # the queue, and then re-assign requests to connections.
+                self._release_request_connection(pool_request)
                 self._requests.remove(pool_request)
                 closing = self._assign_requests_to_connections()
 
@@ -267,6 +272,24 @@ async def handle_async_request(self, request: Request) -> Response:
             extensions=response.extensions,
         )
 
+    def _reserve_connection(
+        self, pool_request: AsyncPoolRequest, connection: AsyncConnectionInterface
+    ) -> None:
+        pool_request.assign_to_connection(connection)
+        self._request_connections[connection] = (
+            self._request_connections.get(connection, 0) + 1
+        )
+
+    def _release_request_connection(self, pool_request: AsyncPoolRequest) -> None:
+        connection = pool_request.connection
+        if connection is not None:
+            pool_request.connection = None
+            count = self._request_connections[connection] - 1
+            if count:
+                self._request_connections[connection] = count
+            else:
+                del self._request_connections[connection]
+
     def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
         """
         Manage the state of the connection pool, assigning incoming
@@ -279,13 +302,14 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
         """
         closing_connections: list[AsyncConnectionInterface] = []
         retained_connections: list[AsyncConnectionInterface] = []
+        assigned = self._request_connections
 
         # First we handle cleaning up any connections that are closed
         # or have expired their keep-alive, in a single pass.
         for connection in self._connections:
             if connection.is_closed():
                 continue
-            elif connection.has_expired():
+            elif connection not in assigned and connection.has_expired():
                 closing_connections.append(connection)
             else:
                 retained_connections.append(connection)
@@ -293,13 +317,21 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
         # Then we close any surplus idle connections, to enforce the
         # max_keepalive_connections setting.
         idle_surplus = (
-            sum(connection.is_idle() for connection in retained_connections)
+            sum(
+                connection.is_idle()
+                for connection in retained_connections
+                if connection not in assigned
+            )
             - self._max_keepalive_connections
         )
         if idle_surplus > 0:
             kept: list[AsyncConnectionInterface] = []
             for connection in retained_connections:
-                if idle_surplus > 0 and connection.is_idle():
+                if (
+                    idle_surplus > 0
+                    and connection not in assigned
+                    and connection.is_idle()
+                ):
                     closing_connections.append(connection)
                     idle_surplus -= 1
                 else:
@@ -310,15 +342,18 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
 
         # Snapshot the reusable connections once instead of rebuilding the
         # list for every queued request.
-        available_connections = [
-            connection
+        available_connections = OrderedDict(
+            (id(connection), connection)
             for connection in self._connections
             if connection.is_available()
-        ]
+            and (connection not in assigned or connection._is_multiplexable())
+        )
         new_connection_budget = self._max_connections - len(self._connections)
 
         # Assign queued requests to connections.
         for pool_request in self._requests:
+            if not available_connections and new_connection_budget <= 0:
+                break
             if not pool_request.is_queued():
                 continue
             origin = pool_request.request.url.origin
@@ -329,29 +364,31 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
             # 2. We can create a new connection to handle the request.
             # 3. We can close an idle connection and then create a new connection
             #    to handle the request.
-            for connection in available_connections:
+            for connection in available_connections.values():
                 if connection.can_handle_request(origin):
-                    pool_request.assign_to_connection(connection)
+                    if not connection._is_multiplexable():
+                        del available_connections[id(connection)]
+                    self._reserve_connection(pool_request, connection)
                     break
             else:
                 if new_connection_budget > 0:
                     connection = self.create_connection(origin)
                     self._connections.append(connection)
-                    if connection.is_available():
-                        available_connections.append(connection)
-                    pool_request.assign_to_connection(connection)
+                    if connection.is_available() and connection._is_multiplexable():
+                        available_connections[id(connection)] = connection
+                    self._reserve_connection(pool_request, connection)
                     new_connection_budget -= 1
                     continue
-                for idx, connection in enumerate(available_connections):
-                    if connection.is_idle():
-                        del available_connections[idx]
+                for connection in available_connections.values():
+                    if connection not in assigned and connection.is_idle():
+                        del available_connections[id(connection)]
                         self._connections.remove(connection)
                         closing_connections.append(connection)
                         connection = self.create_connection(origin)
                         self._connections.append(connection)
-                        if connection.is_available():
-                            available_connections.append(connection)
-                        pool_request.assign_to_connection(connection)
+                        if connection.is_available() and connection._is_multiplexable():
+                            available_connections[id(connection)] = connection
+                        self._reserve_connection(pool_request, connection)
                         break
 
         return closing_connections
@@ -425,13 +462,18 @@ async def __aiter__(self) -> typing.AsyncIterator[bytes]:
             raise exc from None
 
     async def aclose(self) -> None:
-        if not self._closed:
+        with self._pool._optional_thread_lock:
+            if self._closed:
+                return
             self._closed = True
+
+        try:
             with AsyncShieldCancellation():
                 if hasattr(self._stream, "aclose"):
                     await self._stream.aclose()
-
+        finally:
             with self._pool._optional_thread_lock:
+                self._pool._release_request_connection(self._pool_request)
                 self._pool._requests.remove(self._pool_request)
                 closing = self._pool._assign_requests_to_connections()
 
diff --git a/httpcore/_async/http2.py b/httpcore/_async/http2.py
index dbd0beeb4..e7e432a87 100644
--- a/httpcore/_async/http2.py
+++ b/httpcore/_async/http2.py
@@ -519,6 +519,9 @@ def is_available(self) -> bool:
             )
         )
 
+    def _is_multiplexable(self) -> bool:
+        return True
+
     def has_expired(self) -> bool:
         now = time.monotonic()
         return self._expire_at is not None and now > self._expire_at
diff --git a/httpcore/_async/http_proxy.py b/httpcore/_async/http_proxy.py
index cc9d92066..b4be6f498 100644
--- a/httpcore/_async/http_proxy.py
+++ b/httpcore/_async/http_proxy.py
@@ -217,6 +217,9 @@ def info(self) -> str:
     def is_available(self) -> bool:
         return self._connection.is_available()
 
+    def _is_multiplexable(self) -> bool:
+        return self._connection._is_multiplexable()
+
     def has_expired(self) -> bool:
         return self._connection.has_expired()
 
@@ -354,6 +357,9 @@ def info(self) -> str:
     def is_available(self) -> bool:
         return self._connection.is_available()
 
+    def _is_multiplexable(self) -> bool:
+        return self._connection._is_multiplexable()
+
     def has_expired(self) -> bool:
         return self._connection.has_expired()
 
diff --git a/httpcore/_async/interfaces.py b/httpcore/_async/interfaces.py
index 361583bed..98e988eea 100644
--- a/httpcore/_async/interfaces.py
+++ b/httpcore/_async/interfaces.py
@@ -112,6 +112,10 @@ def is_available(self) -> bool:
         """
         raise NotImplementedError()  # pragma: nocover
 
+    def _is_multiplexable(self) -> bool:
+        """Whether multiple pool requests may hold an assignment at once."""
+        return False
+
     def has_expired(self) -> bool:
         """
         Return `True` if the connection is in a state where it should be closed.
diff --git a/httpcore/_async/socks_proxy.py b/httpcore/_async/socks_proxy.py
index b363f55a0..6afab55d4 100644
--- a/httpcore/_async/socks_proxy.py
+++ b/httpcore/_async/socks_proxy.py
@@ -317,6 +317,11 @@ def is_available(self) -> bool:
             )
         return self._connection.is_available()
 
+    def _is_multiplexable(self) -> bool:
+        if self._connection is None:
+            return self.is_available()
+        return self._connection._is_multiplexable()
+
     def has_expired(self) -> bool:
         if self._connection is None:  # pragma: nocover
             return self._connect_failed
diff --git a/httpcore/_sync/connection.py b/httpcore/_sync/connection.py
index 363f8be81..a9f86194c 100644
--- a/httpcore/_sync/connection.py
+++ b/httpcore/_sync/connection.py
@@ -184,6 +184,12 @@ def is_available(self) -> bool:
             )
         return self._connection.is_available()
 
+    def _is_multiplexable(self) -> bool:
+        if self._connection is None:
+            # Preserve speculative HTTP/2 sharing until the protocol is known.
+            return self.is_available()
+        return self._connection._is_multiplexable()
+
     def has_expired(self) -> bool:
         if self._connection is None:
             return self._connect_failed
diff --git a/httpcore/_sync/connection_pool.py b/httpcore/_sync/connection_pool.py
index f6d3b7a8f..8d92699a0 100644
--- a/httpcore/_sync/connection_pool.py
+++ b/httpcore/_sync/connection_pool.py
@@ -4,6 +4,7 @@
 import sys
 import types
 import typing
+from collections import OrderedDict
 
 from .._backends.sync import SyncBackend
 from .._backends.base import SOCKET_OPTION, NetworkBackend
@@ -119,6 +120,7 @@ def __init__(
         # and the set of connections that are servicing those requests.
         self._connections: list[ConnectionInterface] = []
         self._requests: list[PoolRequest] = []
+        self._request_connections: dict[ConnectionInterface, int] = {}
 
         # We only mutate the state of the connection pool within an 'optional_thread_lock'
         # context. This holds a threading lock unless we're running in async mode,
@@ -241,7 +243,9 @@ def handle_request(self, request: Request) -> Response:
                     # handle a request, but then become unavailable.
                     #
                     # In this case we clear the connection and try again.
-                    pool_request.clear_connection()
+                    with self._optional_thread_lock:
+                        self._release_request_connection(pool_request)
+                        pool_request.clear_connection()
                 else:
                     break  # pragma: nocover
 
@@ -249,6 +253,7 @@ def handle_request(self, request: Request) -> Response:
             with self._optional_thread_lock:
                 # For any exception or cancellation we remove the request from
                 # the queue, and then re-assign requests to connections.
+                self._release_request_connection(pool_request)
                 self._requests.remove(pool_request)
                 closing = self._assign_requests_to_connections()
 
@@ -267,6 +272,24 @@ def handle_request(self, request: Request) -> Response:
             extensions=response.extensions,
         )
 
+    def _reserve_connection(
+        self, pool_request: PoolRequest, connection: ConnectionInterface
+    ) -> None:
+        pool_request.assign_to_connection(connection)
+        self._request_connections[connection] = (
+            self._request_connections.get(connection, 0) + 1
+        )
+
+    def _release_request_connection(self, pool_request: PoolRequest) -> None:
+        connection = pool_request.connection
+        if connection is not None:
+            pool_request.connection = None
+            count = self._request_connections[connection] - 1
+            if count:
+                self._request_connections[connection] = count
+            else:
+                del self._request_connections[connection]
+
     def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
         """
         Manage the state of the connection pool, assigning incoming
@@ -279,13 +302,14 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
         """
         closing_connections: list[ConnectionInterface] = []
         retained_connections: list[ConnectionInterface] = []
+        assigned = self._request_connections
 
         # First we handle cleaning up any connections that are closed
         # or have expired their keep-alive, in a single pass.
         for connection in self._connections:
             if connection.is_closed():
                 continue
-            elif connection.has_expired():
+            elif connection not in assigned and connection.has_expired():
                 closing_connections.append(connection)
             else:
                 retained_connections.append(connection)
@@ -293,13 +317,21 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
         # Then we close any surplus idle connections, to enforce the
         # max_keepalive_connections setting.
         idle_surplus = (
-            sum(connection.is_idle() for connection in retained_connections)
+            sum(
+                connection.is_idle()
+                for connection in retained_connections
+                if connection not in assigned
+            )
             - self._max_keepalive_connections
         )
         if idle_surplus > 0:
             kept: list[ConnectionInterface] = []
             for connection in retained_connections:
-                if idle_surplus > 0 and connection.is_idle():
+                if (
+                    idle_surplus > 0
+                    and connection not in assigned
+                    and connection.is_idle()
+                ):
                     closing_connections.append(connection)
                     idle_surplus -= 1
                 else:
@@ -310,15 +342,18 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
 
         # Snapshot the reusable connections once instead of rebuilding the
         # list for every queued request.
-        available_connections = [
-            connection
+        available_connections = OrderedDict(
+            (id(connection), connection)
             for connection in self._connections
             if connection.is_available()
-        ]
+            and (connection not in assigned or connection._is_multiplexable())
+        )
         new_connection_budget = self._max_connections - len(self._connections)
 
         # Assign queued requests to connections.
         for pool_request in self._requests:
+            if not available_connections and new_connection_budget <= 0:
+                break
             if not pool_request.is_queued():
                 continue
             origin = pool_request.request.url.origin
@@ -329,29 +364,31 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
             # 2. We can create a new connection to handle the request.
             # 3. We can close an idle connection and then create a new connection
             #    to handle the request.
-            for connection in available_connections:
+            for connection in available_connections.values():
                 if connection.can_handle_request(origin):
-                    pool_request.assign_to_connection(connection)
+                    if not connection._is_multiplexable():
+                        del available_connections[id(connection)]
+                    self._reserve_connection(pool_request, connection)
                     break
             else:
                 if new_connection_budget > 0:
                     connection = self.create_connection(origin)
                     self._connections.append(connection)
-                    if connection.is_available():
-                        available_connections.append(connection)
-                    pool_request.assign_to_connection(connection)
+                    if connection.is_available() and connection._is_multiplexable():
+                        available_connections[id(connection)] = connection
+                    self._reserve_connection(pool_request, connection)
                     new_connection_budget -= 1
                     continue
-                for idx, connection in enumerate(available_connections):
-                    if connection.is_idle():
-                        del available_connections[idx]
+                for connection in available_connections.values():
+                    if connection not in assigned and connection.is_idle():
+                        del available_connections[id(connection)]
                         self._connections.remove(connection)
                         closing_connections.append(connection)
                         connection = self.create_connection(origin)
                         self._connections.append(connection)
-                        if connection.is_available():
-                            available_connections.append(connection)
-                        pool_request.assign_to_connection(connection)
+                        if connection.is_available() and connection._is_multiplexable():
+                            available_connections[id(connection)] = connection
+                        self._reserve_connection(pool_request, connection)
                         break
 
         return closing_connections
@@ -425,13 +462,18 @@ def __iter__(self) -> typing.Iterator[bytes]:
             raise exc from None
 
     def close(self) -> None:
-        if not self._closed:
+        with self._pool._optional_thread_lock:
+            if self._closed:
+                return
             self._closed = True
+
+        try:
             with ShieldCancellation():
                 if hasattr(self._stream, "close"):
                     self._stream.close()
-
+        finally:
             with self._pool._optional_thread_lock:
+                self._pool._release_request_connection(self._pool_request)
                 self._pool._requests.remove(self._pool_request)
                 closing = self._pool._assign_requests_to_connections()
 
diff --git a/httpcore/_sync/http2.py b/httpcore/_sync/http2.py
index ddcc18900..bf679a98d 100644
--- a/httpcore/_sync/http2.py
+++ b/httpcore/_sync/http2.py
@@ -519,6 +519,9 @@ def is_available(self) -> bool:
             )
         )
 
+    def _is_multiplexable(self) -> bool:
+        return True
+
     def has_expired(self) -> bool:
         now = time.monotonic()
         return self._expire_at is not None and now > self._expire_at
diff --git a/httpcore/_sync/http_proxy.py b/httpcore/_sync/http_proxy.py
index ecca88f7d..323bdaa75 100644
--- a/httpcore/_sync/http_proxy.py
+++ b/httpcore/_sync/http_proxy.py
@@ -217,6 +217,9 @@ def info(self) -> str:
     def is_available(self) -> bool:
         return self._connection.is_available()
 
+    def _is_multiplexable(self) -> bool:
+        return self._connection._is_multiplexable()
+
     def has_expired(self) -> bool:
         return self._connection.has_expired()
 
@@ -354,6 +357,9 @@ def info(self) -> str:
     def is_available(self) -> bool:
         return self._connection.is_available()
 
+    def _is_multiplexable(self) -> bool:
+        return self._connection._is_multiplexable()
+
     def has_expired(self) -> bool:
         return self._connection.has_expired()
 
diff --git a/httpcore/_sync/interfaces.py b/httpcore/_sync/interfaces.py
index e673d4cc1..93697d64f 100644
--- a/httpcore/_sync/interfaces.py
+++ b/httpcore/_sync/interfaces.py
@@ -112,6 +112,10 @@ def is_available(self) -> bool:
         """
         raise NotImplementedError()  # pragma: nocover
 
+    def _is_multiplexable(self) -> bool:
+        """Whether multiple pool requests may hold an assignment at once."""
+        return False
+
     def has_expired(self) -> bool:
         """
         Return `True` if the connection is in a state where it should be closed.
diff --git a/httpcore/_sync/socks_proxy.py b/httpcore/_sync/socks_proxy.py
index 0ca96ddfb..cac49453a 100644
--- a/httpcore/_sync/socks_proxy.py
+++ b/httpcore/_sync/socks_proxy.py
@@ -317,6 +317,11 @@ def is_available(self) -> bool:
             )
         return self._connection.is_available()
 
+    def _is_multiplexable(self) -> bool:
+        if self._connection is None:
+            return self.is_available()
+        return self._connection._is_multiplexable()
+
     def has_expired(self) -> bool:
         if self._connection is None:  # pragma: nocover
             return self._connect_failed
diff --git a/pyproject.toml b/pyproject.toml
index 1bdd99eb9..2494bfed9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -56,13 +56,18 @@ Source = "https://github.com/encode/httpcore"
 path = "httpcore/__init__.py"
 
 [tool.hatch.build.targets.sdist]
+core-metadata-version = "2.4"
 include = [
     "/httpcore",
     "/CHANGELOG.md",
     "/README.md",
+    "/scripts/unasync.py",
     "/tests"
 ]
 
+[tool.hatch.build.targets.wheel]
+core-metadata-version = "2.4"
+
 [tool.hatch.metadata.hooks.fancy-pypi-readme]
 content-type = "text/markdown"
 
diff --git a/scripts/unasync.py b/scripts/unasync.py
index 5a5627d71..9a6549676 100644
--- a/scripts/unasync.py
+++ b/scripts/unasync.py
@@ -5,6 +5,7 @@
 from pprint import pprint
 
 SUBS = [
+    (r'httpcore\._async', 'httpcore._sync'),
     ('from .._backends.auto import AutoBackend', 'from .._backends.sync import SyncBackend'),
     ('import trio as concurrency', 'from tests import concurrency'),
     ('AsyncIterator', 'Iterator'),
diff --git a/tests/_async/test_connection_pool.py b/tests/_async/test_connection_pool.py
index 5b979da0e..87c7867da 100644
--- a/tests/_async/test_connection_pool.py
+++ b/tests/_async/test_connection_pool.py
@@ -40,7 +40,7 @@ async def test_connection_pool_does_not_multiplex_new_http11_connections():
 
 
 @pytest.mark.anyio
-async def test_connection_pool_reuses_replacement_within_assignment_pass():
+async def test_connection_pool_reuses_replacement_within_assignment_pass(monkeypatch):
     pool = httpcore.AsyncConnectionPool(max_connections=1, http2=True)
     pool._requests = [
         AsyncPoolRequest(httpcore.Request("GET", "https://new.example.com/"))
@@ -54,7 +54,7 @@ async def test_connection_pool_reuses_replacement_within_assignment_pass():
     idle.can_handle_request.return_value = False
     pool._connections = [idle]
     replacement = pool.create_connection(pool._requests[0].request.url.origin)
-    pool.create_connection = Mock(return_value=replacement)
+    monkeypatch.setattr(pool, "create_connection", Mock(return_value=replacement))
 
     closing = pool._assign_requests_to_connections()
 
diff --git a/tests/_async/test_connection_reservations.py b/tests/_async/test_connection_reservations.py
new file mode 100644
index 000000000..ab87fb0a3
--- /dev/null
+++ b/tests/_async/test_connection_reservations.py
@@ -0,0 +1,380 @@
+from __future__ import annotations
+
+from unittest.mock import Mock
+
+import hpack
+import hyperframe.frame
+import pytest
+
+import httpcore
+from httpcore._async.connection_pool import AsyncPoolRequest
+from httpcore._async.http_proxy import (
+    AsyncForwardHTTPConnection,
+    AsyncTunnelHTTPConnection,
+)
+from httpcore._async.socks_proxy import AsyncSocks5Connection
+
+ORIGIN = httpcore.Origin(b"https", b"example.com", 443)
+RESPONSE = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"
+
+
+async def idle_connection() -> httpcore.AsyncHTTP11Connection:
+    connection = httpcore.AsyncHTTP11Connection(
+        ORIGIN, httpcore.AsyncMockStream([RESPONSE] * 4)
+    )
+    response = await connection.request("GET", "https://example.com/")
+    assert response.content == b"OK"
+    return connection
+
+
+def pool_request(url: str = "https://example.com/") -> AsyncPoolRequest:
+    return AsyncPoolRequest(httpcore.Request("GET", url))
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("max_connections", [2, 100_000])
+async def test_reserves_idle_http11_within_pass(max_connections):
+    async with httpcore.AsyncConnectionPool(max_connections=max_connections) as pool:
+        connections: list[httpcore.AsyncConnectionInterface] = [
+            await idle_connection(),
+            await idle_connection(),
+        ]
+        pool._connections = connections.copy()
+        pool._requests = [pool_request(), pool_request()]
+
+        assert pool._assign_requests_to_connections() == []
+        assert [request.connection for request in pool._requests] == connections
+        assert pool.connections == connections
+
+
+@pytest.mark.anyio
+async def test_reserves_idle_http11_across_passes_and_releases():
+    async with httpcore.AsyncConnectionPool(max_connections=1) as pool:
+        connection = await idle_connection()
+        pool._connections = [connection]
+        owner, waiting = pool_request(), pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        pool._requests.append(waiting)
+        pool._assign_requests_to_connections()
+        assert owner.connection is connection
+        assert waiting.connection is None
+
+        # Requeueing the owner releases the reservation but preserves FIFO order.
+        pool._release_request_connection(owner)
+        owner.clear_connection()
+        pool._assign_requests_to_connections()
+        assert owner.connection is connection
+        assert waiting.connection is None
+
+        # Error/cancellation/response closure all remove the owner from this list.
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        pool._assign_requests_to_connections()
+        assert waiting.connection is connection
+
+
+@pytest.mark.anyio
+async def test_reserved_idle_connection_is_not_evicted():
+    async with httpcore.AsyncConnectionPool(max_connections=1) as pool:
+        connection = await idle_connection()
+        pool._connections = [connection]
+        owner = pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        waiting = pool_request("https://other.example/")
+        pool._requests.append(waiting)
+        pool._max_keepalive_connections = 0
+
+        assert pool._assign_requests_to_connections() == []
+        assert owner.connection is connection
+        assert waiting.connection is None
+        assert pool.connections == [connection]
+
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        assert pool._assign_requests_to_connections() == [connection]
+        await connection.aclose()
+        assert waiting.connection is pool.connections[0]
+        assert waiting.connection is not connection
+
+
+@pytest.mark.anyio
+async def test_mixed_pool_evicts_only_unreserved_idle_surplus():
+    async with httpcore.AsyncConnectionPool(max_connections=3) as pool:
+        reserved, surplus, retained = [await idle_connection() for _ in range(3)]
+        pool._connections = [reserved, surplus, retained]
+        owner = pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        pool._max_keepalive_connections = 1
+
+        assert pool._assign_requests_to_connections() == [surplus]
+        await surplus.aclose()
+        assert pool.connections == [reserved, retained]
+        assert owner.connection is reserved
+        assert pool._request_connections == {reserved: 1}
+
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        assert pool._assign_requests_to_connections() == [reserved]
+        await reserved.aclose()
+        assert pool.connections == [retained]
+        response = await pool.request("GET", "https://example.com/")
+        assert response.status == 200 and response.content == b"OK"
+        assert pool._requests == [] and pool._request_connections == {}
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "error_type", [httpcore.ReadError, httpcore.ConnectionNotAvailable]
+)
+async def test_reservation_released_after_handler_error(error_type):
+    class RejectOnce(httpcore.AsyncHTTP11Connection):
+        reject = False
+
+        async def handle_async_request(self, request):
+            if self.reject:
+                self.reject = False
+                raise error_type()
+            return await super().handle_async_request(request)
+
+    connection = RejectOnce(ORIGIN, httpcore.AsyncMockStream([RESPONSE] * 4))
+    await connection.request("GET", "https://example.com/")
+    connection.reject = True
+    async with httpcore.AsyncConnectionPool(max_connections=1) as pool:
+        pool._connections = [connection]
+        if error_type is httpcore.ReadError:
+            with pytest.raises(httpcore.ReadError):
+                await pool.request("GET", "https://example.com/")
+        else:
+            response = await pool.request("GET", "https://example.com/")
+            assert response.content == b"OK"
+        assert pool._requests == []
+        assert pool._request_connections == {}
+        response = await pool.request("GET", "https://example.com/")
+        assert response.status == 200 and response.content == b"OK"
+        assert pool._requests == []
+        assert pool._request_connections == {}
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("http2", [False, True])
+async def test_speculative_multiplexing_and_http11_fallback(http2):
+    async with httpcore.AsyncConnectionPool(max_connections=2, http2=http2) as pool:
+        pool._requests = [pool_request(), pool_request()]
+        pool._assign_requests_to_connections()
+        assert len(pool.connections) == (1 if http2 else 2)
+        connection = pool.connections[0]
+        assert isinstance(connection, httpcore.AsyncHTTPConnection)
+        assert connection._is_multiplexable() is http2
+
+        # After a speculative H2 connection negotiates H1, reserve it singly.
+        connection._connection = await idle_connection()
+        assert not connection._is_multiplexable()
+        extra = pool_request()
+        pool._requests.append(extra)
+        pool._assign_requests_to_connections()
+        assert extra.connection is not connection
+
+
+@pytest.mark.anyio
+async def test_existing_http2_connection_remains_shared():
+    async with httpcore.AsyncConnectionPool(max_connections=1, http2=True) as pool:
+        connection = httpcore.AsyncHTTP2Connection(ORIGIN, httpcore.AsyncMockStream([]))
+        pool._connections = [connection]
+        owner, waiting = pool_request(), pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        pool._requests.append(waiting)
+        pool._assign_requests_to_connections()
+        assert owner.connection is waiting.connection is connection
+        assert pool._request_connections == {connection: 2}
+        pool._max_keepalive_connections = 0
+        assert pool._assign_requests_to_connections() == []
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        assert pool._request_connections == {connection: 1}
+        # Releasing the same owner again must not consume the other reservation.
+        pool._release_request_connection(owner)
+        assert pool._request_connections == {connection: 1}
+        assert pool._assign_requests_to_connections() == []
+        pool._release_request_connection(waiting)
+        pool._requests.remove(waiting)
+        assert pool._request_connections == {}
+        assert pool._assign_requests_to_connections() == [connection]
+        await connection.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "wrapper_type", [AsyncForwardHTTPConnection, AsyncTunnelHTTPConnection]
+)
+async def test_proxy_reservation_capability_delegates(wrapper_type):
+    wrapper = wrapper_type(
+        proxy_origin=httpcore.Origin(b"http", b"proxy.example", 8080),
+        remote_origin=ORIGIN,
+    )
+    assert not wrapper._is_multiplexable()
+    wrapper._connection = await idle_connection()
+    assert not wrapper._is_multiplexable()
+    await wrapper.aclose()
+    wrapper._connection = httpcore.AsyncHTTP2Connection(
+        ORIGIN, httpcore.AsyncMockStream([])
+    )
+    assert wrapper._is_multiplexable()
+    await wrapper.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("http2", [False, True])
+async def test_socks_reservation_capability_delegates(http2):
+    wrapper = AsyncSocks5Connection(
+        proxy_origin=httpcore.Origin(b"http", b"proxy.example", 1080),
+        remote_origin=ORIGIN,
+        http2=http2,
+    )
+    assert wrapper._is_multiplexable() is http2
+    wrapper._connection = await idle_connection()
+    assert not wrapper._is_multiplexable()
+    await wrapper.aclose()
+    wrapper._connection = httpcore.AsyncHTTP2Connection(
+        ORIGIN, httpcore.AsyncMockStream([])
+    )
+    assert wrapper._is_multiplexable()
+    await wrapper.aclose()
+
+
+@pytest.mark.anyio
+async def test_reserved_connection_skips_expiry_until_release(monkeypatch):
+    async with httpcore.AsyncConnectionPool(max_connections=1) as pool:
+        connection = await idle_connection()
+        pool._connections = [connection]
+        owner = pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        has_expired = Mock(return_value=True)
+        monkeypatch.setattr(connection, "has_expired", has_expired)
+        assert pool._assign_requests_to_connections() == []
+        has_expired.assert_not_called()
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        assert pool._assign_requests_to_connections() == [connection]
+        has_expired.assert_called_once()
+        await connection.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("transition", ["availability", "assignment"])
+async def test_reserved_connection_survives_state_transition(transition, monkeypatch):
+    async with httpcore.AsyncConnectionPool(max_connections=1) as pool:
+        connection = await idle_connection()
+        pool._connections = [connection]
+        owner, waiting = pool_request(), pool_request()
+        pool._requests = [owner, waiting]
+
+        def activate() -> bool:
+            monkeypatch.setattr(connection, "is_idle", lambda: False)
+            return True
+
+        if transition == "availability":
+            pool._reserve_connection(owner, connection)
+            monkeypatch.setattr(connection, "is_available", activate)
+        else:
+            original_assign = owner.assign_to_connection
+
+            def assign(connection: httpcore.AsyncConnectionInterface | None) -> None:
+                original_assign(connection)
+                activate()
+
+            monkeypatch.setattr(owner, "assign_to_connection", assign)
+        pool._assign_requests_to_connections()
+        assert owner.connection is connection
+        assert waiting.connection is None
+        assert pool._request_connections == {connection: 1}
+
+
+@pytest.mark.anyio
+async def test_reservation_released_when_response_close_trace_raises():
+    error = RuntimeError("close callback failed")
+
+    async def trace(name: str, info: dict[str, object]) -> None:
+        if name == "http11.response_closed.complete":
+            raise error
+
+    backend = httpcore.AsyncMockBackend([RESPONSE] * 2)
+    async with httpcore.AsyncConnectionPool(
+        max_connections=1, network_backend=backend
+    ) as pool:
+        response = await pool.handle_async_request(
+            httpcore.Request(
+                "GET",
+                "https://example.com/",
+                headers={"Host": "example.com"},
+                extensions={"trace": trace},
+            )
+        )
+        assert response.status == 200 and await response.aread() == b"OK"
+        with pytest.raises(RuntimeError) as caught:
+            await response.aclose()
+        assert caught.value is error
+        assert pool._requests == [] and pool._request_connections == {}
+        assert pool.connections[0].is_idle()
+        await response.aclose()
+        response = await pool.request(
+            "GET", "https://example.com/", extensions={"timeout": {"pool": 0}}
+        )
+        assert response.status == 200 and response.content == b"OK"
+        assert pool._requests == [] and pool._request_connections == {}
+
+
+@pytest.mark.anyio
+async def test_duplicate_http2_response_close_preserves_other_reservation():
+    encoder = hpack.Encoder()
+    buffer = [
+        hyperframe.frame.SettingsFrame(
+            settings={hyperframe.frame.SettingsFrame.MAX_CONCURRENT_STREAMS: 2}
+        ).serialize()
+    ]
+    for stream_id in (1, 3):
+        buffer.extend(
+            [
+                hyperframe.frame.HeadersFrame(
+                    stream_id=stream_id,
+                    data=encoder.encode([(b":status", b"200")]),
+                    flags=["END_HEADERS"],
+                ).serialize(),
+                hyperframe.frame.DataFrame(
+                    stream_id=stream_id, data=b"OK", flags=["END_STREAM"]
+                ).serialize(),
+            ]
+        )
+    backend = httpcore.AsyncMockBackend(buffer, http2=True)
+    async with httpcore.AsyncConnectionPool(
+        max_connections=1,
+        max_keepalive_connections=0,
+        http2=True,
+        network_backend=backend,
+    ) as pool:
+        first = await pool.handle_async_request(
+            httpcore.Request(
+                "GET", "https://example.com/", headers={"Host": "example.com"}
+            )
+        )
+        assert first.status == 200 and await first.aread() == b"OK"
+        second = await pool.handle_async_request(
+            httpcore.Request(
+                "GET", "https://example.com/", headers={"Host": "example.com"}
+            )
+        )
+        (connection,) = pool.connections
+        assert pool._request_connections == {connection: 2}
+        await first.aclose()
+        await first.aclose()
+        assert pool._request_connections == {connection: 1}
+        assert pool.connections == [connection]
+        assert not connection.is_closed()
+        assert second.status == 200 and await second.aread() == b"OK"
+        await second.aclose()
+        assert pool._requests == [] and pool._request_connections == {}
+        assert connection.is_closed() and pool.connections == []
diff --git a/tests/_sync/test_connection_pool.py b/tests/_sync/test_connection_pool.py
index 06bdd26e9..6cb5c8705 100644
--- a/tests/_sync/test_connection_pool.py
+++ b/tests/_sync/test_connection_pool.py
@@ -25,6 +25,7 @@ def test_connection_pool_reuses_new_connection_within_assignment_pass():
     assert {request.connection for request in pool._requests} == set(pool.connections)
 
 
+
 def test_connection_pool_does_not_multiplex_new_http11_connections():
     pool = httpcore.ConnectionPool(max_connections=10)
     pool._requests = [
@@ -38,7 +39,8 @@ def test_connection_pool_does_not_multiplex_new_http11_connections():
     assert len({request.connection for request in pool._requests}) == 10
 
 
-def test_connection_pool_reuses_replacement_within_assignment_pass():
+
+def test_connection_pool_reuses_replacement_within_assignment_pass(monkeypatch):
     pool = httpcore.ConnectionPool(max_connections=1, http2=True)
     pool._requests = [
         PoolRequest(httpcore.Request("GET", "https://new.example.com/"))
@@ -52,7 +54,7 @@ def test_connection_pool_reuses_replacement_within_assignment_pass():
     idle.can_handle_request.return_value = False
     pool._connections = [idle]
     replacement = pool.create_connection(pool._requests[0].request.url.origin)
-    pool.create_connection = Mock(return_value=replacement)
+    monkeypatch.setattr(pool, "create_connection", Mock(return_value=replacement))
 
     closing = pool._assign_requests_to_connections()
 
@@ -61,6 +63,7 @@ def test_connection_pool_reuses_replacement_within_assignment_pass():
     assert {request.connection for request in pool._requests} == {replacement}
 
 
+
 def test_connection_pool_with_keepalive():
     """
     By default HTTP/1.1 requests should be returned to the connection pool.
@@ -586,6 +589,7 @@ def test_connection_pool_closes_idle_connection_for_different_origin():
         assert "https://b.com:443" in repr(pool.connections[0])
 
 
+
 def test_connection_pool_concurrency():
     """
     HTTP/1.1 requests made in concurrency must not ever exceed the maximum number
diff --git a/tests/_sync/test_connection_reservations.py b/tests/_sync/test_connection_reservations.py
new file mode 100644
index 000000000..92732bdd0
--- /dev/null
+++ b/tests/_sync/test_connection_reservations.py
@@ -0,0 +1,380 @@
+from __future__ import annotations
+
+from unittest.mock import Mock
+
+import hpack
+import hyperframe.frame
+import pytest
+
+import httpcore
+from httpcore._sync.connection_pool import PoolRequest
+from httpcore._sync.http_proxy import (
+    ForwardHTTPConnection,
+    TunnelHTTPConnection,
+)
+from httpcore._sync.socks_proxy import Socks5Connection
+
+ORIGIN = httpcore.Origin(b"https", b"example.com", 443)
+RESPONSE = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"
+
+
+def idle_connection() -> httpcore.HTTP11Connection:
+    connection = httpcore.HTTP11Connection(
+        ORIGIN, httpcore.MockStream([RESPONSE] * 4)
+    )
+    response = connection.request("GET", "https://example.com/")
+    assert response.content == b"OK"
+    return connection
+
+
+def pool_request(url: str = "https://example.com/") -> PoolRequest:
+    return PoolRequest(httpcore.Request("GET", url))
+
+
+
+@pytest.mark.parametrize("max_connections", [2, 100_000])
+def test_reserves_idle_http11_within_pass(max_connections):
+    with httpcore.ConnectionPool(max_connections=max_connections) as pool:
+        connections: list[httpcore.ConnectionInterface] = [
+            idle_connection(),
+            idle_connection(),
+        ]
+        pool._connections = connections.copy()
+        pool._requests = [pool_request(), pool_request()]
+
+        assert pool._assign_requests_to_connections() == []
+        assert [request.connection for request in pool._requests] == connections
+        assert pool.connections == connections
+
+
+
+def test_reserves_idle_http11_across_passes_and_releases():
+    with httpcore.ConnectionPool(max_connections=1) as pool:
+        connection = idle_connection()
+        pool._connections = [connection]
+        owner, waiting = pool_request(), pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        pool._requests.append(waiting)
+        pool._assign_requests_to_connections()
+        assert owner.connection is connection
+        assert waiting.connection is None
+
+        # Requeueing the owner releases the reservation but preserves FIFO order.
+        pool._release_request_connection(owner)
+        owner.clear_connection()
+        pool._assign_requests_to_connections()
+        assert owner.connection is connection
+        assert waiting.connection is None
+
+        # Error/cancellation/response closure all remove the owner from this list.
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        pool._assign_requests_to_connections()
+        assert waiting.connection is connection
+
+
+
+def test_reserved_idle_connection_is_not_evicted():
+    with httpcore.ConnectionPool(max_connections=1) as pool:
+        connection = idle_connection()
+        pool._connections = [connection]
+        owner = pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        waiting = pool_request("https://other.example/")
+        pool._requests.append(waiting)
+        pool._max_keepalive_connections = 0
+
+        assert pool._assign_requests_to_connections() == []
+        assert owner.connection is connection
+        assert waiting.connection is None
+        assert pool.connections == [connection]
+
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        assert pool._assign_requests_to_connections() == [connection]
+        connection.close()
+        assert waiting.connection is pool.connections[0]
+        assert waiting.connection is not connection
+
+
+
+def test_mixed_pool_evicts_only_unreserved_idle_surplus():
+    with httpcore.ConnectionPool(max_connections=3) as pool:
+        reserved, surplus, retained = [idle_connection() for _ in range(3)]
+        pool._connections = [reserved, surplus, retained]
+        owner = pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        pool._max_keepalive_connections = 1
+
+        assert pool._assign_requests_to_connections() == [surplus]
+        surplus.close()
+        assert pool.connections == [reserved, retained]
+        assert owner.connection is reserved
+        assert pool._request_connections == {reserved: 1}
+
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        assert pool._assign_requests_to_connections() == [reserved]
+        reserved.close()
+        assert pool.connections == [retained]
+        response = pool.request("GET", "https://example.com/")
+        assert response.status == 200 and response.content == b"OK"
+        assert pool._requests == [] and pool._request_connections == {}
+
+
+
+@pytest.mark.parametrize(
+    "error_type", [httpcore.ReadError, httpcore.ConnectionNotAvailable]
+)
+def test_reservation_released_after_handler_error(error_type):
+    class RejectOnce(httpcore.HTTP11Connection):
+        reject = False
+
+        def handle_request(self, request):
+            if self.reject:
+                self.reject = False
+                raise error_type()
+            return super().handle_request(request)
+
+    connection = RejectOnce(ORIGIN, httpcore.MockStream([RESPONSE] * 4))
+    connection.request("GET", "https://example.com/")
+    connection.reject = True
+    with httpcore.ConnectionPool(max_connections=1) as pool:
+        pool._connections = [connection]
+        if error_type is httpcore.ReadError:
+            with pytest.raises(httpcore.ReadError):
+                pool.request("GET", "https://example.com/")
+        else:
+            response = pool.request("GET", "https://example.com/")
+            assert response.content == b"OK"
+        assert pool._requests == []
+        assert pool._request_connections == {}
+        response = pool.request("GET", "https://example.com/")
+        assert response.status == 200 and response.content == b"OK"
+        assert pool._requests == []
+        assert pool._request_connections == {}
+
+
+
+@pytest.mark.parametrize("http2", [False, True])
+def test_speculative_multiplexing_and_http11_fallback(http2):
+    with httpcore.ConnectionPool(max_connections=2, http2=http2) as pool:
+        pool._requests = [pool_request(), pool_request()]
+        pool._assign_requests_to_connections()
+        assert len(pool.connections) == (1 if http2 else 2)
+        connection = pool.connections[0]
+        assert isinstance(connection, httpcore.HTTPConnection)
+        assert connection._is_multiplexable() is http2
+
+        # After a speculative H2 connection negotiates H1, reserve it singly.
+        connection._connection = idle_connection()
+        assert not connection._is_multiplexable()
+        extra = pool_request()
+        pool._requests.append(extra)
+        pool._assign_requests_to_connections()
+        assert extra.connection is not connection
+
+
+
+def test_existing_http2_connection_remains_shared():
+    with httpcore.ConnectionPool(max_connections=1, http2=True) as pool:
+        connection = httpcore.HTTP2Connection(ORIGIN, httpcore.MockStream([]))
+        pool._connections = [connection]
+        owner, waiting = pool_request(), pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        pool._requests.append(waiting)
+        pool._assign_requests_to_connections()
+        assert owner.connection is waiting.connection is connection
+        assert pool._request_connections == {connection: 2}
+        pool._max_keepalive_connections = 0
+        assert pool._assign_requests_to_connections() == []
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        assert pool._request_connections == {connection: 1}
+        # Releasing the same owner again must not consume the other reservation.
+        pool._release_request_connection(owner)
+        assert pool._request_connections == {connection: 1}
+        assert pool._assign_requests_to_connections() == []
+        pool._release_request_connection(waiting)
+        pool._requests.remove(waiting)
+        assert pool._request_connections == {}
+        assert pool._assign_requests_to_connections() == [connection]
+        connection.close()
+
+
+
+@pytest.mark.parametrize(
+    "wrapper_type", [ForwardHTTPConnection, TunnelHTTPConnection]
+)
+def test_proxy_reservation_capability_delegates(wrapper_type):
+    wrapper = wrapper_type(
+        proxy_origin=httpcore.Origin(b"http", b"proxy.example", 8080),
+        remote_origin=ORIGIN,
+    )
+    assert not wrapper._is_multiplexable()
+    wrapper._connection = idle_connection()
+    assert not wrapper._is_multiplexable()
+    wrapper.close()
+    wrapper._connection = httpcore.HTTP2Connection(
+        ORIGIN, httpcore.MockStream([])
+    )
+    assert wrapper._is_multiplexable()
+    wrapper.close()
+
+
+
+@pytest.mark.parametrize("http2", [False, True])
+def test_socks_reservation_capability_delegates(http2):
+    wrapper = Socks5Connection(
+        proxy_origin=httpcore.Origin(b"http", b"proxy.example", 1080),
+        remote_origin=ORIGIN,
+        http2=http2,
+    )
+    assert wrapper._is_multiplexable() is http2
+    wrapper._connection = idle_connection()
+    assert not wrapper._is_multiplexable()
+    wrapper.close()
+    wrapper._connection = httpcore.HTTP2Connection(
+        ORIGIN, httpcore.MockStream([])
+    )
+    assert wrapper._is_multiplexable()
+    wrapper.close()
+
+
+
+def test_reserved_connection_skips_expiry_until_release(monkeypatch):
+    with httpcore.ConnectionPool(max_connections=1) as pool:
+        connection = idle_connection()
+        pool._connections = [connection]
+        owner = pool_request()
+        pool._requests = [owner]
+        pool._assign_requests_to_connections()
+        has_expired = Mock(return_value=True)
+        monkeypatch.setattr(connection, "has_expired", has_expired)
+        assert pool._assign_requests_to_connections() == []
+        has_expired.assert_not_called()
+        pool._release_request_connection(owner)
+        pool._requests.remove(owner)
+        assert pool._assign_requests_to_connections() == [connection]
+        has_expired.assert_called_once()
+        connection.close()
+
+
+
+@pytest.mark.parametrize("transition", ["availability", "assignment"])
+def test_reserved_connection_survives_state_transition(transition, monkeypatch):
+    with httpcore.ConnectionPool(max_connections=1) as pool:
+        connection = idle_connection()
+        pool._connections = [connection]
+        owner, waiting = pool_request(), pool_request()
+        pool._requests = [owner, waiting]
+
+        def activate() -> bool:
+            monkeypatch.setattr(connection, "is_idle", lambda: False)
+            return True
+
+        if transition == "availability":
+            pool._reserve_connection(owner, connection)
+            monkeypatch.setattr(connection, "is_available", activate)
+        else:
+            original_assign = owner.assign_to_connection
+
+            def assign(connection: httpcore.ConnectionInterface | None) -> None:
+                original_assign(connection)
+                activate()
+
+            monkeypatch.setattr(owner, "assign_to_connection", assign)
+        pool._assign_requests_to_connections()
+        assert owner.connection is connection
+        assert waiting.connection is None
+        assert pool._request_connections == {connection: 1}
+
+
+
+def test_reservation_released_when_response_close_trace_raises():
+    error = RuntimeError("close callback failed")
+
+    def trace(name: str, info: dict[str, object]) -> None:
+        if name == "http11.response_closed.complete":
+            raise error
+
+    backend = httpcore.MockBackend([RESPONSE] * 2)
+    with httpcore.ConnectionPool(
+        max_connections=1, network_backend=backend
+    ) as pool:
+        response = pool.handle_request(
+            httpcore.Request(
+                "GET",
+                "https://example.com/",
+                headers={"Host": "example.com"},
+                extensions={"trace": trace},
+            )
+        )
+        assert response.status == 200 and response.read() == b"OK"
+        with pytest.raises(RuntimeError) as caught:
+            response.close()
+        assert caught.value is error
+        assert pool._requests == [] and pool._request_connections == {}
+        assert pool.connections[0].is_idle()
+        response.close()
+        response = pool.request(
+            "GET", "https://example.com/", extensions={"timeout": {"pool": 0}}
+        )
+        assert response.status == 200 and response.content == b"OK"
+        assert pool._requests == [] and pool._request_connections == {}
+
+
+
+def test_duplicate_http2_response_close_preserves_other_reservation():
+    encoder = hpack.Encoder()
+    buffer = [
+        hyperframe.frame.SettingsFrame(
+            settings={hyperframe.frame.SettingsFrame.MAX_CONCURRENT_STREAMS: 2}
+        ).serialize()
+    ]
+    for stream_id in (1, 3):
+        buffer.extend(
+            [
+                hyperframe.frame.HeadersFrame(
+                    stream_id=stream_id,
+                    data=encoder.encode([(b":status", b"200")]),
+                    flags=["END_HEADERS"],
+                ).serialize(),
+                hyperframe.frame.DataFrame(
+                    stream_id=stream_id, data=b"OK", flags=["END_STREAM"]
+                ).serialize(),
+            ]
+        )
+    backend = httpcore.MockBackend(buffer, http2=True)
+    with httpcore.ConnectionPool(
+        max_connections=1,
+        max_keepalive_connections=0,
+        http2=True,
+        network_backend=backend,
+    ) as pool:
+        first = pool.handle_request(
+            httpcore.Request(
+                "GET", "https://example.com/", headers={"Host": "example.com"}
+            )
+        )
+        assert first.status == 200 and first.read() == b"OK"
+        second = pool.handle_request(
+            httpcore.Request(
+                "GET", "https://example.com/", headers={"Host": "example.com"}
+            )
+        )
+        (connection,) = pool.connections
+        assert pool._request_connections == {connection: 2}
+        first.close()
+        first.close()
+        assert pool._request_connections == {connection: 1}
+        assert pool.connections == [connection]
+        assert not connection.is_closed()
+        assert second.status == 200 and second.read() == b"OK"
+        second.close()
+        assert pool._requests == [] and pool._request_connections == {}
+        assert connection.is_closed() and pool.connections == []
diff --git a/tests/test_connection_pool_reuse.py b/tests/test_connection_pool_reuse.py
new file mode 100644
index 000000000..cff5685b2
--- /dev/null
+++ b/tests/test_connection_pool_reuse.py
@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import asyncio
+
+import httpcore
+
+
+def test_warm_http11_burst_does_not_retry_reserved_connections(monkeypatch):
+    attempts = 0
+    original = httpcore.AsyncHTTP11Connection.handle_async_request
+
+    async def counted_request(
+        self: httpcore.AsyncHTTP11Connection, request: httpcore.Request
+    ) -> httpcore.Response:
+        nonlocal attempts
+        attempts += 1
+        return await original(self, request)
+
+    class CountingPool(httpcore.AsyncConnectionPool):
+        assignment_passes = 0
+
+        def _assign_requests_to_connections(
+            self,
+        ) -> list[httpcore.AsyncConnectionInterface]:
+            self.assignment_passes += 1
+            return super()._assign_requests_to_connections()
+
+    monkeypatch.setattr(
+        httpcore.AsyncHTTP11Connection, "handle_async_request", counted_request
+    )
+
+    async def run() -> None:
+        nonlocal attempts
+        count = 16
+        backend = httpcore.AsyncMockBackend(
+            [b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK"] * (2 * count)
+        )
+        pool = CountingPool(
+            max_connections=100_000,
+            max_keepalive_connections=100_000,
+            network_backend=backend,
+        )
+        async with pool:
+
+            async def fetch() -> None:
+                response = await pool.request("GET", "http://example.com/")
+                assert response.status == 200
+                assert response.content == b"OK"
+
+            await asyncio.gather(*(fetch() for _ in range(count)))
+            assert len(pool.connections) == count
+            assert all(connection.is_idle() for connection in pool.connections)
+
+            attempts = pool.assignment_passes = 0
+            await asyncio.gather(*(fetch() for _ in range(count)))
+
+            assert attempts == count
+            assert pool.assignment_passes == 2 * count
+            assert len(pool.connections) == count
+            assert pool._requests == []
+            assert pool._request_connections == {}
+        assert pool.connections == []
+
+    asyncio.run(run())
diff --git a/tests/test_unasync.py b/tests/test_unasync.py
new file mode 100644
index 000000000..4b2fafe30
--- /dev/null
+++ b/tests/test_unasync.py
@@ -0,0 +1,22 @@
+import pathlib
+import runpy
+
+import pytest
+
+
+@pytest.mark.parametrize(
+    "source, expected",
+    [
+        (
+            "from httpcore._async.connection_pool import AsyncPoolRequest\n",
+            "from httpcore._sync.connection_pool import PoolRequest\n",
+        ),
+        (
+            "from httpcore._async.http_proxy import AsyncTunnelHTTPConnection\n",
+            "from httpcore._sync.http_proxy import TunnelHTTPConnection\n",
+        ),
+    ],
+)
+def test_unasync_internal_import(source, expected):
+    script = pathlib.Path(__file__).parents[1] / "scripts" / "unasync.py"
+    assert runpy.run_path(str(script))["unasync_line"](source) == expected