diff --git a/playwright/_impl/_connection.py b/playwright/_impl/_connection.py
index 12a1c60af..dc04bbad1 100644
--- a/playwright/_impl/_connection.py
+++ b/playwright/_impl/_connection.py
@@ -143,14 +143,10 @@ async def _inner_send(
         callback = self._connection._send_message_to_server(
             self._object, method, augmented_params, timeout
         )
+        # Transport errors reject pending callbacks via Connection.cleanup(); no need
+        # to also race transport.on_error_future here.
         try:
-            done, _ = await asyncio.wait(
-                {
-                    self._connection._transport.on_error_future,
-                    callback.future,
-                },
-                return_when=asyncio.FIRST_COMPLETED,
-            )
+            result = await callback.future
         except asyncio.CancelledError as exc:
             await self._connection._abort(
                 self._object,
@@ -158,9 +154,6 @@ async def _inner_send(
                 str(exc) or "Task was cancelled",
             )
             raise
-        if not callback.future.done():
-            callback.future.cancel()
-        result = next(iter(done)).result()
         # Protocol now has named return values, assume result is one level deeper unless
         # there is explicit ambiguity.
         if not result:
@@ -305,6 +298,7 @@ def __init__(
         self._dispatcher_fiber = dispatcher_fiber
         self._transport = transport
         self._transport.on_message = lambda msg: self.dispatch(msg)
+        self._transport.on_error = self._on_transport_error
         self._waiting_for_object: Dict[str, Callable[[ChannelOwner], None]] = {}
         self._last_id = 0
         self._objects: Dict[str, ChannelOwner] = {}
@@ -358,21 +352,34 @@ async def stop_async(self) -> None:
         await self._transport.wait_until_stopped()
         self.cleanup()
 
+    def _on_transport_error(self, error: Exception) -> None:
+        # Unexpected pipe death: fail every in-flight send so waiters do not hang
+        # and do not need to race transport.on_error_future themselves.
+        self.cleanup(str(error) or None)
+
     def cleanup(self, cause: str = None) -> None:
-        self._closed_error = TargetClosedError(cause) if cause else TargetClosedError()
-        if self._init_task and not self._init_task.done():
-            self._init_task.cancel()
-        for ws_connection in self._child_ws_connections:
-            ws_connection._transport.dispose()
-        for callback in self._callbacks.values():
-            # To prevent 'Future exception was never retrieved' we ignore all callbacks that are no_reply.
-            if callback.no_reply:
-                continue
-            if callback.future.cancelled():
-                continue
-            callback.future.set_exception(self._closed_error)
-        self._callbacks.clear()
-        self.emit("close")
+        if not self._closed_error:
+            self._closed_error = (
+                TargetClosedError(cause) if cause else TargetClosedError()
+            )
+            if self._init_task and not self._init_task.done():
+                self._init_task.cancel()
+            for ws_connection in self._child_ws_connections:
+                ws_connection._transport.dispose()
+            for callback in self._callbacks.values():
+                # To prevent 'Future exception was never retrieved' we ignore all callbacks that are no_reply.
+                if callback.no_reply:
+                    continue
+                if callback.future.cancelled():
+                    continue
+                callback.future.set_exception(self._closed_error)
+            self._callbacks.clear()
+            self.emit("close")
+        # Startup / connect waiters race this future; if nobody awaited it (e.g. the
+        # pipe died after init), mark the exception retrieved so GC stays quiet.
+        on_error_future = self._transport.on_error_future
+        if on_error_future.done() and not on_error_future.cancelled():
+            on_error_future.exception()
 
     def call_on_object_with_known_name(
         self, guid: str, callback: Callable[[ChannelOwner], None]
@@ -441,9 +448,8 @@ def _send_message_to_server(
         if self._tracing_count > 0 and frames and object._guid != "localUtils":
             self.local_utils.add_stack_to_tracing_no_reply(id, frames)
 
-        self._callbacks[id] = callback
         self._transport.send(message)
-
+        self._callbacks[id] = callback
         return callback
 
     async def _abort(
@@ -459,20 +465,15 @@ async def _abort(
             )
         except (Error, OSError):
             pass
+        # Wait for the server abort ack or connection teardown (cleanup rejects us).
+        if callback.future.done():
+            if not callback.future.cancelled():
+                callback.future.exception()
+            return
         try:
-            done, _ = await asyncio.wait(
-                {
-                    self._transport.on_error_future,
-                    callback.future,
-                },
-                return_when=asyncio.FIRST_COMPLETED,
-            )
-        finally:
-            if not callback.future.done():
-                callback.future.cancel()
-        for future in done:
-            if not future.cancelled():
-                future.exception()
+            await callback.future
+        except (Exception, asyncio.CancelledError):
+            pass
 
     def dispatch(self, msg: ParsedMessagePayload) -> None:
         if self._closed_error:
diff --git a/playwright/_impl/_json_pipe.py b/playwright/_impl/_json_pipe.py
index 41973b8c7..5a034a2d0 100644
--- a/playwright/_impl/_json_pipe.py
+++ b/playwright/_impl/_json_pipe.py
@@ -54,10 +54,13 @@ def handle_message(message: Dict) -> None:
             self.on_message(cast(ParsedMessagePayload, message))
 
         def handle_closed(reason: Optional[str]) -> None:
-            self.emit("close", reason)
-            if reason:
+            # Set the error future before notifying listeners so Connection.cleanup()
+            # can mark the exception retrieved if nobody else awaits it.
+            if reason and not self.on_error_future.done():
                 self.on_error_future.set_exception(TargetClosedError(reason))
-            self._stopped_future.set_result(None)
+            self.emit("close", reason)
+            if not self._stopped_future.done():
+                self._stopped_future.set_result(None)
 
         self._pipe_channel.on(
             "message",
diff --git a/playwright/_impl/_transport.py b/playwright/_impl/_transport.py
index 3cc029e18..3bd3ac45c 100644
--- a/playwright/_impl/_transport.py
+++ b/playwright/_impl/_transport.py
@@ -50,6 +50,9 @@ def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
         self._loop = loop
         self.on_message: Callable[[ParsedMessagePayload], None] = lambda _: None
         self.on_error_future: asyncio.Future = loop.create_future()
+        # Called on unexpected transport failure so the connection can reject
+        # in-flight protocol callbacks (see Connection._on_transport_error).
+        self.on_error: Callable[[Exception], None] = lambda _: None
 
     @abstractmethod
     def request_stop(self) -> None:
@@ -129,11 +132,16 @@ async def connect(self) -> None:
                 startupinfo=startupinfo,
             )
         except Exception as exc:
-            self.on_error_future.set_exception(exc)
+            self._handle_error(exc)
             raise exc
 
         self._output = self._proc.stdin
 
+    def _handle_error(self, error: Exception) -> None:
+        if not self.on_error_future.done():
+            self.on_error_future.set_exception(error)
+        self.on_error(error)
+
     async def run(self) -> None:
         assert self._proc.stdout
         assert self._proc.stdin
@@ -162,8 +170,10 @@ async def run(self) -> None:
                     self.on_message(obj)
                 except asyncio.IncompleteReadError:
                     if not self._stopped:
-                        self.on_error_future.set_exception(
-                            Exception("Connection closed while reading from the driver")
+                        self._handle_error(
+                            Exception(
+                                "Connection closed while reading from the driver"
+                            )
                         )
                     break
                 await asyncio.sleep(0)
diff --git a/tests/async/test_asyncio.py b/tests/async/test_asyncio.py
index 243a0eed8..3408efa46 100644
--- a/tests/async/test_asyncio.py
+++ b/tests/async/test_asyncio.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 import asyncio
 import gc
+import os
 import subprocess
 import sys
 import textwrap
@@ -154,3 +155,94 @@ async def test_should_return_proper_api_name_on_error(page: Page) -> None:
     except Exception as error:
         # Each browser returns slightly different error messages, but they should all start with "Page.evaluate:", because that was the Playwright method where the error originated
         assert str(error).startswith("Page.evaluate:")
+
+
+async def test_no_orphaned_future_on_serialization_error(
+    browser_name: str,
+    launch_arguments: Dict,
+) -> None:
+    # When transport.send fails (e.g. non-JSON-serializable params), the protocol
+    # callback must not be left for cleanup() to reject — that yields
+    # "Future exception was never retrieved". See issue #3165.
+    handler_exception = None
+
+    def exception_handler(loop: asyncio.AbstractEventLoop, context: Dict) -> None:
+        nonlocal handler_exception
+        handler_exception = context.get("exception")
+
+    asyncio.get_running_loop().set_exception_handler(exception_handler)
+    try:
+        async with async_playwright() as p:
+            browser = await p[browser_name].launch(**launch_arguments)
+            page = await browser.new_page()
+            with pytest.raises(TypeError, match="JSON serializable"):
+                await page.locator("asdf").highlight(style=object())  # type: ignore[arg-type]
+            await browser.close()
+        gc.collect()
+        assert handler_exception is None
+    finally:
+        asyncio.get_running_loop().set_exception_handler(None)
+
+
+async def test_no_orphaned_future_on_send_may_fail_teardown(
+    browser_name: str,
+    launch_arguments: Dict,
+    tmp_path: Path,
+) -> None:
+    # Mirrors pytest-playwright session teardown: background send_may_fail work,
+    # then browser.close() + playwright.stop(). Also covers unexpected driver death
+    # before stop (transport EOF → connection.cleanup). See issue #3165.
+    script = tmp_path / "repro_may_fail.py"
+    script.write_text(
+        textwrap.dedent(
+            f"""
+            import asyncio
+            import gc
+            import os
+            import signal
+
+            from playwright.async_api import async_playwright
+
+            async def session(kill_driver: bool) -> None:
+                p = await async_playwright().start()
+                browser = await p["{browser_name}"].launch(**{launch_arguments!r})
+                page = await browser.new_page()
+                channel = page._impl_obj._channel
+                for _ in range(20):
+                    channel.send_may_fail("bringToFront", None, {{}})
+                await asyncio.sleep(0)
+                if kill_driver:
+                    os.kill(
+                        page._impl_obj._connection._transport._proc.pid,
+                        signal.SIGKILL,
+                    )
+                    await asyncio.sleep(0.01)
+                else:
+                    await browser.close()
+                try:
+                    await p.stop()
+                except Exception:
+                    pass
+
+            async def main() -> None:
+                for i in range(40):
+                    await session(kill_driver=(i % 2 == 0))
+                    gc.collect()
+                await asyncio.sleep(0.1)
+                gc.collect()
+
+            asyncio.run(main())
+            """
+        )
+    )
+    env = os.environ.copy()
+    env["PYTHONPATH"] = str(Path(__file__).resolve().parents[2])
+    result = subprocess.run(
+        [sys.executable, str(script)],
+        capture_output=True,
+        text=True,
+        timeout=180,
+        env=env,
+    )
+    assert result.returncode == 0, result.stderr
+    assert "Future exception was never retrieved" not in result.stderr
diff --git a/tests/sync/test_sync.py b/tests/sync/test_sync.py
index fb9c7c16e..9fe27f00d 100644
--- a/tests/sync/test_sync.py
+++ b/tests/sync/test_sync.py
@@ -14,7 +14,11 @@
 
 import multiprocessing
 import os
+import subprocess
+import sys
+import textwrap
 from datetime import timedelta
+from pathlib import Path
 from typing import Any, Callable, Dict
 
 import pytest
@@ -364,3 +368,36 @@ def test_should_return_proper_api_name_on_error(page: Page) -> None:
 def test_click_should_accept_timedelta_for_timeout(page: Page) -> None:
     with pytest.raises(TimeoutError, match="Timeout 1ms exceeded"):
         page.click("does-not-exist", timeout=timedelta(milliseconds=1))
+
+
+def test_no_orphaned_future_on_serialization_error(
+    browser_name: str, launch_arguments: Dict[str, Any], tmp_path: Path
+) -> None:
+    # When transport.send fails (e.g. non-JSON-serializable params), the protocol
+    # callback must not be left for cleanup() to reject — that yields
+    # "Future exception was never retrieved". See issue #3165.
+    script = tmp_path / "repro.py"
+    script.write_text(
+        textwrap.dedent(
+            f"""
+            from playwright.sync_api import sync_playwright
+
+            with sync_playwright() as p:
+                browser = p["{browser_name}"].launch(**{launch_arguments!r})
+                page = browser.new_page()
+                try:
+                    page.locator("asdf").highlight(style=object())
+                except TypeError:
+                    pass
+                browser.close()
+            """
+        )
+    )
+    result = subprocess.run(
+        [sys.executable, str(script)],
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
+    assert result.returncode == 0, result.stderr
+    assert "Future exception was never retrieved" not in result.stderr