diff --git a/src/httpcore2/httpcore2/_async/connection_pool.py b/src/httpcore2/httpcore2/_async/connection_pool.py
index 2c5e3980..a6682ba6 100644
--- a/src/httpcore2/httpcore2/_async/connection_pool.py
+++ b/src/httpcore2/httpcore2/_async/connection_pool.py
@@ -112,6 +112,11 @@ def __init__(
         self._connections: list[AsyncConnectionInterface] = []
         self._requests: list[AsyncPoolRequest] = []
 
+        # Reference counts of connections held by in-flight requests,
+        # maintained incrementally so assignment passes never rebuild them
+        # by scanning the full request list.
+        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,
         # in which case it is a no-op.
@@ -227,7 +232,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: no cover
 
@@ -235,6 +242,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()
 
@@ -251,6 +259,19 @@ 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:
+            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
@@ -267,7 +288,7 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
         # Connections currently referenced by an in-flight request, including
         # connections that are in the process of being established and idle
         # connections reserved by an assigned-but-not-yet-sent request.
-        request_connections = {r.connection for r in self._requests}
+        request_connections = self._request_connections
 
         # First we handle cleaning up any connections that are closed
         # or have expired their keep-alive, in a single pass. Reserved
@@ -310,16 +331,16 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
         # it per queued request — this is what brings the loop from O(N*M) to
         # O(N+M) in the common case.
         #
-        # An idle connection already assigned to an in-flight request is
-        # reserved: it stays IDLE until the winning task sends on it, so
-        # without this exclusion the next pass would assign it again and the
-        # loser would churn through `ConnectionNotAvailable`. Multiplexing
-        # connections are exempt: they can take further requests while idle.
+        # An established non-multiplexing connection already assigned to an
+        # in-flight request is reserved. Its state may transition from IDLE to
+        # ACTIVE after `is_available()` returns, so use `is_connected()` here
+        # rather than checking its mutable idle state. Multiplexing connections
+        # and not-yet-connected HTTP/2 candidates remain available.
         available_connections = [
             connection
             for connection in self._connections
             if connection.is_available()
-            and not (connection.is_idle() and connection in request_connections and not connection.can_multiplex())
+            and not (connection.is_connected() and connection in request_connections and not connection.can_multiplex())
         ]
         new_connection_budget = self._max_connections - len(self._connections)
 
@@ -342,17 +363,17 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
             #    to handle the request.
             for idx, connection in enumerate(available_connections):
                 if connection.can_handle_request(origin):
-                    pool_request.assign_to_connection(connection)
-                    if connection.is_idle() and not connection.can_multiplex():
-                        # An idle HTTP/1.1 connection can only take this
-                        # single request until it is released.
+                    if connection.is_connected() and not connection.can_multiplex():
+                        # Remove an established HTTP/1.1 connection before
+                        # waking the request, which may transition it to ACTIVE.
                         del available_connections[idx]
+                    self._reserve_connection(pool_request, connection)
                     break
             else:
                 if new_connection_budget > 0:
                     connection = self.create_connection(origin)
                     self._connections.append(connection)
-                    pool_request.assign_to_connection(connection)
+                    self._reserve_connection(pool_request, connection)
                     new_connection_budget -= 1
                     continue
                 for idx, connection in enumerate(available_connections):
@@ -362,7 +383,7 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
                         closing_connections.append(connection)
                         connection = self.create_connection(origin)
                         self._connections.append(connection)
-                        pool_request.assign_to_connection(connection)
+                        self._reserve_connection(pool_request, connection)
                         break
 
         return closing_connections
@@ -434,6 +455,7 @@ async def aclose(self) -> None:
                     await self._stream.aclose()
 
             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/src/httpcore2/httpcore2/_async/http_proxy.py b/src/httpcore2/httpcore2/_async/http_proxy.py
index 18d42ecf..98d4bfbe 100644
--- a/src/httpcore2/httpcore2/_async/http_proxy.py
+++ b/src/httpcore2/httpcore2/_async/http_proxy.py
@@ -216,6 +216,9 @@ def has_expired(self) -> bool:
     def is_idle(self) -> bool:
         return self._connection.is_idle()
 
+    def can_multiplex(self) -> bool:
+        return self._connection.can_multiplex()
+
     def is_closed(self) -> bool:
         return self._connection.is_closed()
 
@@ -345,6 +348,9 @@ def has_expired(self) -> bool:
     def is_idle(self) -> bool:
         return self._connection.is_idle()
 
+    def can_multiplex(self) -> bool:
+        return self._connection.can_multiplex()
+
     def is_closed(self) -> bool:
         return self._connection.is_closed()
 
diff --git a/src/httpcore2/httpcore2/_async/socks_proxy.py b/src/httpcore2/httpcore2/_async/socks_proxy.py
index 2ae84c18..8762da04 100644
--- a/src/httpcore2/httpcore2/_async/socks_proxy.py
+++ b/src/httpcore2/httpcore2/_async/socks_proxy.py
@@ -311,6 +311,9 @@ def is_idle(self) -> bool:
             return self._connect_failed
         return self._connection.is_idle()
 
+    def can_multiplex(self) -> bool:
+        return self._connection is not None and self._connection.can_multiplex()
+
     def is_closed(self) -> bool:
         if self._connection is None:  # pragma: no cover
             return self._connect_failed
diff --git a/src/httpcore2/httpcore2/_sync/connection_pool.py b/src/httpcore2/httpcore2/_sync/connection_pool.py
index a6051bac..62ac6dfa 100644
--- a/src/httpcore2/httpcore2/_sync/connection_pool.py
+++ b/src/httpcore2/httpcore2/_sync/connection_pool.py
@@ -112,6 +112,11 @@ def __init__(
         self._connections: list[ConnectionInterface] = []
         self._requests: list[PoolRequest] = []
 
+        # Reference counts of connections held by in-flight requests,
+        # maintained incrementally so assignment passes never rebuild them
+        # by scanning the full request list.
+        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,
         # in which case it is a no-op.
@@ -227,7 +232,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: no cover
 
@@ -235,6 +242,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()
 
@@ -251,6 +259,19 @@ 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:
+            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
@@ -267,7 +288,7 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
         # Connections currently referenced by an in-flight request, including
         # connections that are in the process of being established and idle
         # connections reserved by an assigned-but-not-yet-sent request.
-        request_connections = {r.connection for r in self._requests}
+        request_connections = self._request_connections
 
         # First we handle cleaning up any connections that are closed
         # or have expired their keep-alive, in a single pass. Reserved
@@ -310,16 +331,16 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
         # it per queued request — this is what brings the loop from O(N*M) to
         # O(N+M) in the common case.
         #
-        # An idle connection already assigned to an in-flight request is
-        # reserved: it stays IDLE until the winning task sends on it, so
-        # without this exclusion the next pass would assign it again and the
-        # loser would churn through `ConnectionNotAvailable`. Multiplexing
-        # connections are exempt: they can take further requests while idle.
+        # An established non-multiplexing connection already assigned to an
+        # in-flight request is reserved. Its state may transition from IDLE to
+        # ACTIVE after `is_available()` returns, so use `is_connected()` here
+        # rather than checking its mutable idle state. Multiplexing connections
+        # and not-yet-connected HTTP/2 candidates remain available.
         available_connections = [
             connection
             for connection in self._connections
             if connection.is_available()
-            and not (connection.is_idle() and connection in request_connections and not connection.can_multiplex())
+            and not (connection.is_connected() and connection in request_connections and not connection.can_multiplex())
         ]
         new_connection_budget = self._max_connections - len(self._connections)
 
@@ -342,17 +363,17 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
             #    to handle the request.
             for idx, connection in enumerate(available_connections):
                 if connection.can_handle_request(origin):
-                    pool_request.assign_to_connection(connection)
-                    if connection.is_idle() and not connection.can_multiplex():
-                        # An idle HTTP/1.1 connection can only take this
-                        # single request until it is released.
+                    if connection.is_connected() and not connection.can_multiplex():
+                        # Remove an established HTTP/1.1 connection before
+                        # waking the request, which may transition it to ACTIVE.
                         del available_connections[idx]
+                    self._reserve_connection(pool_request, connection)
                     break
             else:
                 if new_connection_budget > 0:
                     connection = self.create_connection(origin)
                     self._connections.append(connection)
-                    pool_request.assign_to_connection(connection)
+                    self._reserve_connection(pool_request, connection)
                     new_connection_budget -= 1
                     continue
                 for idx, connection in enumerate(available_connections):
@@ -362,7 +383,7 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
                         closing_connections.append(connection)
                         connection = self.create_connection(origin)
                         self._connections.append(connection)
-                        pool_request.assign_to_connection(connection)
+                        self._reserve_connection(pool_request, connection)
                         break
 
         return closing_connections
@@ -434,6 +455,7 @@ def close(self) -> None:
                     self._stream.close()
 
             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/src/httpcore2/httpcore2/_sync/http_proxy.py b/src/httpcore2/httpcore2/_sync/http_proxy.py
index 42e41421..33dd1f58 100644
--- a/src/httpcore2/httpcore2/_sync/http_proxy.py
+++ b/src/httpcore2/httpcore2/_sync/http_proxy.py
@@ -216,6 +216,9 @@ def has_expired(self) -> bool:
     def is_idle(self) -> bool:
         return self._connection.is_idle()
 
+    def can_multiplex(self) -> bool:
+        return self._connection.can_multiplex()
+
     def is_closed(self) -> bool:
         return self._connection.is_closed()
 
@@ -345,6 +348,9 @@ def has_expired(self) -> bool:
     def is_idle(self) -> bool:
         return self._connection.is_idle()
 
+    def can_multiplex(self) -> bool:
+        return self._connection.can_multiplex()
+
     def is_closed(self) -> bool:
         return self._connection.is_closed()
 
diff --git a/src/httpcore2/httpcore2/_sync/socks_proxy.py b/src/httpcore2/httpcore2/_sync/socks_proxy.py
index 4ec5372f..77d9b902 100644
--- a/src/httpcore2/httpcore2/_sync/socks_proxy.py
+++ b/src/httpcore2/httpcore2/_sync/socks_proxy.py
@@ -311,6 +311,9 @@ def is_idle(self) -> bool:
             return self._connect_failed
         return self._connection.is_idle()
 
+    def can_multiplex(self) -> bool:
+        return self._connection is not None and self._connection.can_multiplex()
+
     def is_closed(self) -> bool:
         if self._connection is None:  # pragma: no cover
             return self._connect_failed
diff --git a/tests/httpcore2/_async/test_connection_pool.py b/tests/httpcore2/_async/test_connection_pool.py
index 11b9a5d4..0273990d 100644
--- a/tests/httpcore2/_async/test_connection_pool.py
+++ b/tests/httpcore2/_async/test_connection_pool.py
@@ -884,3 +884,104 @@ async def fetch(pool: httpcore2.AsyncConnectionPool) -> None:
                 nursery.start_soon(fetch, pool)
 
     assert QueueObservingPool.max_queued_after_pass == 0
+
+
+class RacingConnection(httpcore2.AsyncConnectionInterface):
+    def __init__(self, *, activate_during_availability_check: bool = False) -> None:
+        self.active = False
+        self.closed = False
+        self.activate_during_availability_check = activate_during_availability_check
+
+    async def handle_async_request(self, request: httpcore2.Request) -> httpcore2.Response:
+        raise NotImplementedError
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+    def info(self) -> str:
+        return "racing connection"
+
+    def can_handle_request(self, origin: httpcore2.Origin) -> bool:
+        return True
+
+    def is_connected(self) -> bool:
+        return not self.closed
+
+    def is_available(self) -> bool:
+        available = not self.active and not self.closed
+        if self.activate_during_availability_check:
+            self.active = True
+        return available
+
+    def has_expired(self) -> bool:
+        return False
+
+    def is_idle(self) -> bool:
+        return not self.active and not self.closed
+
+    def is_closed(self) -> bool:
+        return self.closed
+
+
+class RacingPoolRequest:
+    def __init__(
+        self,
+        *,
+        connection: RacingConnection | None = None,
+        activate_on_assignment: bool = False,
+    ) -> None:
+        self.request = httpcore2.Request("GET", "https://example.com/")
+        self.connection = connection
+        self.activate_on_assignment = activate_on_assignment
+
+    def assign_to_connection(self, connection: RacingConnection) -> None:
+        self.connection = connection
+        if self.activate_on_assignment:
+            connection.active = True
+
+    def is_queued(self) -> bool:
+        return self.connection is None
+
+
+@pytest.mark.anyio
+async def test_connection_pool_removes_connection_before_waking_request() -> None:
+    """
+    A sync request may start using a connection as soon as its event is set.
+    Remove a non-multiplexing connection from the candidate snapshot before
+    assigning it, so that state transition cannot make it eligible again.
+    """
+    connection = RacingConnection()
+    first = RacingPoolRequest(activate_on_assignment=True)
+    second = RacingPoolRequest()
+    pool = httpcore2.AsyncConnectionPool(max_connections=1)
+    pool_state = typing.cast(typing.Any, pool)
+    pool_state._connections = [connection]
+    pool_state._requests = [first, second]
+
+    pool._assign_requests_to_connections()
+
+    assert first.connection is connection
+    assert second.connection is None
+    await pool.aclose()
+
+
+@pytest.mark.anyio
+async def test_connection_pool_excludes_reserved_connection_during_state_transition() -> None:
+    """
+    A connection may transition from IDLE to ACTIVE between `is_available()`
+    and the reservation check. Established reserved HTTP/1.1 connections must
+    remain excluded regardless of that mutable state.
+    """
+    connection = RacingConnection(activate_during_availability_check=True)
+    active = RacingPoolRequest(connection=connection)
+    queued = RacingPoolRequest()
+    pool = httpcore2.AsyncConnectionPool(max_connections=1)
+    pool_state = typing.cast(typing.Any, pool)
+    pool_state._connections = [connection]
+    pool_state._requests = [active, queued]
+    pool_state._request_connections = {connection: 1}
+
+    pool._assign_requests_to_connections()
+
+    assert queued.connection is None
+    await pool.aclose()
diff --git a/tests/httpcore2/_async/test_http_proxy.py b/tests/httpcore2/_async/test_http_proxy.py
index 046e3f01..2aaed3a1 100644
--- a/tests/httpcore2/_async/test_http_proxy.py
+++ b/tests/httpcore2/_async/test_http_proxy.py
@@ -174,6 +174,7 @@ async def test_proxy_tunneling_http2() -> None:
         assert info == ["<AsyncTunnelHTTPConnection ['https://example.com:443', HTTP/2, IDLE, Request Count: 1]>"]
         assert proxy.connections[0].is_idle()
         assert proxy.connections[0].is_available()
+        assert proxy.connections[0].can_multiplex()
         assert not proxy.connections[0].is_closed()
 
         # A connection on a tunneled proxy can only handle HTTPS requests to the same origin.
diff --git a/tests/httpcore2/_async/test_socks_proxy.py b/tests/httpcore2/_async/test_socks_proxy.py
index 3612868f..299d11b1 100644
--- a/tests/httpcore2/_async/test_socks_proxy.py
+++ b/tests/httpcore2/_async/test_socks_proxy.py
@@ -1,3 +1,8 @@
+import ssl
+import typing
+
+import hpack
+import hyperframe.frame
 import pytest
 
 import httpcore2
@@ -49,6 +54,56 @@ async def test_socks5_request() -> None:
         assert not proxy.connections[0].can_handle_request(httpcore2.Origin(b"https", b"other.com", 443))
 
 
+class HTTP2SocksStream(httpcore2.AsyncMockStream):
+    async def start_tls(
+        self,
+        ssl_context: ssl.SSLContext,
+        server_hostname: str | None = None,
+        timeout: float | None = None,
+    ) -> httpcore2.AsyncNetworkStream:
+        self._http2 = True
+        return self
+
+
+class HTTP2SocksBackend(httpcore2.AsyncMockBackend):
+    async def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: float | None = None,
+        local_address: str | None = None,
+        socket_options: typing.Iterable[httpcore2.SOCKET_OPTION] | None = None,
+    ) -> httpcore2.AsyncNetworkStream:
+        return HTTP2SocksStream(list(self._buffer))
+
+
+@pytest.mark.anyio
+async def test_socks5_request_http2_can_multiplex() -> None:
+    network_backend = HTTP2SocksBackend(
+        [
+            b"\x05\x00",
+            b"\x05\x00\x00\x01\xff\x00\x00\x01\x00\x50",
+            hyperframe.frame.SettingsFrame().serialize(),
+            hyperframe.frame.HeadersFrame(
+                stream_id=1,
+                data=hpack.Encoder().encode([(b":status", b"200")]),
+                flags=["END_HEADERS"],
+            ).serialize(),
+            hyperframe.frame.DataFrame(stream_id=1, data=b"Hello, world!", flags=["END_STREAM"]).serialize(),
+        ]
+    )
+
+    async with httpcore2.AsyncConnectionPool(
+        proxy=httpcore2.Proxy("socks5://localhost:8080/"),
+        network_backend=network_backend,
+        http2=True,
+    ) as proxy:
+        response = await proxy.request("GET", "https://example.com/")
+
+        assert response.status == 200
+        assert proxy.connections[0].can_multiplex()
+
+
 @pytest.mark.anyio
 async def test_authenticated_socks5_request() -> None:
     """
diff --git a/tests/httpcore2/_sync/test_connection_pool.py b/tests/httpcore2/_sync/test_connection_pool.py
index 693b7371..d9cef938 100644
--- a/tests/httpcore2/_sync/test_connection_pool.py
+++ b/tests/httpcore2/_sync/test_connection_pool.py
@@ -884,3 +884,104 @@ def fetch(pool: httpcore2.ConnectionPool) -> None:
                 nursery.start_soon(fetch, pool)
 
     assert QueueObservingPool.max_queued_after_pass == 0
+
+
+class RacingConnection(httpcore2.ConnectionInterface):
+    def __init__(self, *, activate_during_availability_check: bool = False) -> None:
+        self.active = False
+        self.closed = False
+        self.activate_during_availability_check = activate_during_availability_check
+
+    def handle_request(self, request: httpcore2.Request) -> httpcore2.Response:
+        raise NotImplementedError
+
+    def close(self) -> None:
+        self.closed = True
+
+    def info(self) -> str:
+        return "racing connection"
+
+    def can_handle_request(self, origin: httpcore2.Origin) -> bool:
+        return True
+
+    def is_connected(self) -> bool:
+        return not self.closed
+
+    def is_available(self) -> bool:
+        available = not self.active and not self.closed
+        if self.activate_during_availability_check:
+            self.active = True
+        return available
+
+    def has_expired(self) -> bool:
+        return False
+
+    def is_idle(self) -> bool:
+        return not self.active and not self.closed
+
+    def is_closed(self) -> bool:
+        return self.closed
+
+
+class RacingPoolRequest:
+    def __init__(
+        self,
+        *,
+        connection: RacingConnection | None = None,
+        activate_on_assignment: bool = False,
+    ) -> None:
+        self.request = httpcore2.Request("GET", "https://example.com/")
+        self.connection = connection
+        self.activate_on_assignment = activate_on_assignment
+
+    def assign_to_connection(self, connection: RacingConnection) -> None:
+        self.connection = connection
+        if self.activate_on_assignment:
+            connection.active = True
+
+    def is_queued(self) -> bool:
+        return self.connection is None
+
+
+
+def test_connection_pool_removes_connection_before_waking_request() -> None:
+    """
+    A sync request may start using a connection as soon as its event is set.
+    Remove a non-multiplexing connection from the candidate snapshot before
+    assigning it, so that state transition cannot make it eligible again.
+    """
+    connection = RacingConnection()
+    first = RacingPoolRequest(activate_on_assignment=True)
+    second = RacingPoolRequest()
+    pool = httpcore2.ConnectionPool(max_connections=1)
+    pool_state = typing.cast(typing.Any, pool)
+    pool_state._connections = [connection]
+    pool_state._requests = [first, second]
+
+    pool._assign_requests_to_connections()
+
+    assert first.connection is connection
+    assert second.connection is None
+    pool.close()
+
+
+
+def test_connection_pool_excludes_reserved_connection_during_state_transition() -> None:
+    """
+    A connection may transition from IDLE to ACTIVE between `is_available()`
+    and the reservation check. Established reserved HTTP/1.1 connections must
+    remain excluded regardless of that mutable state.
+    """
+    connection = RacingConnection(activate_during_availability_check=True)
+    active = RacingPoolRequest(connection=connection)
+    queued = RacingPoolRequest()
+    pool = httpcore2.ConnectionPool(max_connections=1)
+    pool_state = typing.cast(typing.Any, pool)
+    pool_state._connections = [connection]
+    pool_state._requests = [active, queued]
+    pool_state._request_connections = {connection: 1}
+
+    pool._assign_requests_to_connections()
+
+    assert queued.connection is None
+    pool.close()
diff --git a/tests/httpcore2/_sync/test_http_proxy.py b/tests/httpcore2/_sync/test_http_proxy.py
index 5e470f0e..db31f455 100644
--- a/tests/httpcore2/_sync/test_http_proxy.py
+++ b/tests/httpcore2/_sync/test_http_proxy.py
@@ -174,6 +174,7 @@ def test_proxy_tunneling_http2() -> None:
         assert info == ["<TunnelHTTPConnection ['https://example.com:443', HTTP/2, IDLE, Request Count: 1]>"]
         assert proxy.connections[0].is_idle()
         assert proxy.connections[0].is_available()
+        assert proxy.connections[0].can_multiplex()
         assert not proxy.connections[0].is_closed()
 
         # A connection on a tunneled proxy can only handle HTTPS requests to the same origin.
diff --git a/tests/httpcore2/_sync/test_socks_proxy.py b/tests/httpcore2/_sync/test_socks_proxy.py
index aa48c9a7..71139bbc 100644
--- a/tests/httpcore2/_sync/test_socks_proxy.py
+++ b/tests/httpcore2/_sync/test_socks_proxy.py
@@ -1,3 +1,8 @@
+import ssl
+import typing
+
+import hpack
+import hyperframe.frame
 import pytest
 
 import httpcore2
@@ -49,6 +54,56 @@ def test_socks5_request() -> None:
         assert not proxy.connections[0].can_handle_request(httpcore2.Origin(b"https", b"other.com", 443))
 
 
+class HTTP2SocksStream(httpcore2.MockStream):
+    def start_tls(
+        self,
+        ssl_context: ssl.SSLContext,
+        server_hostname: str | None = None,
+        timeout: float | None = None,
+    ) -> httpcore2.NetworkStream:
+        self._http2 = True
+        return self
+
+
+class HTTP2SocksBackend(httpcore2.MockBackend):
+    def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: float | None = None,
+        local_address: str | None = None,
+        socket_options: typing.Iterable[httpcore2.SOCKET_OPTION] | None = None,
+    ) -> httpcore2.NetworkStream:
+        return HTTP2SocksStream(list(self._buffer))
+
+
+
+def test_socks5_request_http2_can_multiplex() -> None:
+    network_backend = HTTP2SocksBackend(
+        [
+            b"\x05\x00",
+            b"\x05\x00\x00\x01\xff\x00\x00\x01\x00\x50",
+            hyperframe.frame.SettingsFrame().serialize(),
+            hyperframe.frame.HeadersFrame(
+                stream_id=1,
+                data=hpack.Encoder().encode([(b":status", b"200")]),
+                flags=["END_HEADERS"],
+            ).serialize(),
+            hyperframe.frame.DataFrame(stream_id=1, data=b"Hello, world!", flags=["END_STREAM"]).serialize(),
+        ]
+    )
+
+    with httpcore2.ConnectionPool(
+        proxy=httpcore2.Proxy("socks5://localhost:8080/"),
+        network_backend=network_backend,
+        http2=True,
+    ) as proxy:
+        response = proxy.request("GET", "https://example.com/")
+
+        assert response.status == 200
+        assert proxy.connections[0].can_multiplex()
+
+
 
 def test_authenticated_socks5_request() -> None:
     """