diff --git a/doc/changelog.rst b/doc/changelog.rst
index 5fb2fb4731..9f16f0ee99 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -7,6 +7,15 @@ Changes in Version 4.19.0 (2026/XX/XX)
 PyMongo 4.19 brings a number of changes including:
 
 - Added support for Python 3.15.
+- Added support for running the synchronous and asynchronous clients in
+  subinterpreters, including inside
+  ``concurrent.futures.InterpreterPoolExecutor`` (Python 3.14+). Only in
+  subinterpreters, which do not allow daemon threads, the synchronous
+  client's monitor threads now start as non-daemon threads and are stopped
+  and joined when the interpreter is torn down. Monitor threads in the main
+  interpreter remain daemon threads and shutdown behavior is unchanged.
+  Note that because these threads are non-daemon, a subinterpreter may block
+  on teardown until any in-flight monitor work completes.
 
 Bug fixes
 .........
diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py
index ba0e3804a6..da261e37ea 100644
--- a/pymongo/asynchronous/monitor.py
+++ b/pymongo/asynchronous/monitor.py
@@ -18,6 +18,7 @@
 
 import asyncio
 import atexit
+import threading
 import time
 import weakref
 from typing import TYPE_CHECKING, Any, Optional
@@ -492,3 +493,8 @@ def _shutdown_resources() -> None:
 
 if _IS_SYNC:
     atexit.register(_shutdown_resources)
+    # In subinterpreters, the executors' threads are non-daemon and are
+    # joined at shutdown, so they must be stopped first. Checking daemon
+    # support preserves the normal atexit ordering in the main interpreter.
+    if not periodic_executor._daemon_threads_allowed() and hasattr(threading, "_register_atexit"):
+        threading._register_atexit(_shutdown_resources)  # type: ignore[attr-defined]
diff --git a/pymongo/periodic_executor.py b/pymongo/periodic_executor.py
index 4b979ca9f9..867fc33e06 100644
--- a/pymongo/periodic_executor.py
+++ b/pymongo/periodic_executor.py
@@ -121,6 +121,23 @@ async def _run(self) -> None:
             self._event = False
 
 
+def _daemon_threads_allowed() -> bool:
+    """Whether daemon threads are allowed in the current interpreter.
+
+    Subinterpreters do not allow daemon threads, so a PeriodicExecutor
+    there runs on a non-daemon thread instead: _shutdown_executors stops
+    and joins it during interpreter shutdown.
+    """
+    # Python 3.12+ exposes the same check CPython itself uses to enforce
+    # the restriction (in the Thread daemon setter), which is reliable in
+    # the current interpreter. On older versions daemon threads are
+    # always allowed.
+    check = getattr(threading, "_daemon_threads_allowed", None)
+    if check is not None:
+        return check()
+    return True
+
+
 class PeriodicExecutor:
     def __init__(
         self,
@@ -185,7 +202,15 @@ def open(self) -> None:
 
         if not started:
             thread = threading.Thread(target=self._run, name=self._name)
-            thread.daemon = True
+            try:
+                # Subinterpreters do not allow daemon threads, so fall back
+                # to a non-daemon thread: _shutdown_executors stops and
+                # joins it during interpreter shutdown.
+                thread.daemon = _daemon_threads_allowed()
+            except RuntimeError:
+                # The restriction is enforced when the thread is started,
+                # handled below.
+                pass
             self._thread = weakref.proxy(thread)
             _register_executor(self)
             # Mitigation to RuntimeError firing when thread starts on shutdown
@@ -196,7 +221,13 @@ def open(self) -> None:
                 if "interpreter shutdown" in str(e) or sys.is_finalizing():
                     self._thread = None
                     return
-                raise
+                if not thread.daemon:
+                    raise
+                # Some Python versions only enforce the subinterpreter daemon
+                # restriction when the thread is started, so fall back to a
+                # non-daemon thread as above.
+                thread.daemon = False
+                thread.start()
 
     def close(self, dummy: Any = None) -> None:
         """Stop. To restart, call open().
diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py
index 9a25757f03..3a1e1a49d0 100644
--- a/pymongo/synchronous/monitor.py
+++ b/pymongo/synchronous/monitor.py
@@ -18,6 +18,7 @@
 
 import asyncio
 import atexit
+import threading
 import time
 import weakref
 from typing import TYPE_CHECKING, Any, Optional
@@ -490,3 +491,8 @@ def _shutdown_resources() -> None:
 
 if _IS_SYNC:
     atexit.register(_shutdown_resources)
+    # In subinterpreters, the executors' threads are non-daemon and are
+    # joined at shutdown, so they must be stopped first. Checking daemon
+    # support preserves the normal atexit ordering in the main interpreter.
+    if not periodic_executor._daemon_threads_allowed() and hasattr(threading, "_register_atexit"):
+        threading._register_atexit(_shutdown_resources)  # type: ignore[attr-defined]
diff --git a/pyproject.toml b/pyproject.toml
index 2adddcd29b..eb8aa80984 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -278,7 +278,9 @@ exclude_lines = [
     "return NotImplemented",
     "_use_c = true",
     "if __name__ == '__main__':",
-    "if TYPE_CHECKING:"
+    "if TYPE_CHECKING:",
+    # Sync-only registration glue, generated by unasync into the sync twin.
+    "^if _IS_SYNC"
     ]
 partial_branches = ["if (.*and +)*not _use_c( and.*)*:"]
 
diff --git a/test/asynchronous/test_periodic_executor.py b/test/asynchronous/test_periodic_executor.py
index 15186c4f88..42d64367ee 100644
--- a/test/asynchronous/test_periodic_executor.py
+++ b/test/asynchronous/test_periodic_executor.py
@@ -17,17 +17,29 @@
 from __future__ import annotations
 
 import asyncio
+import importlib
+import os
 import sys
+import textwrap
 import threading
 import time
+from typing import Any
+from unittest import mock
+
+_interpreters: Any = None
+if sys.version_info >= (3, 14):
+    _interpreters = importlib.import_module("concurrent.interpreters")
 
 sys.path[0:0] = [""]
 
+import pymongo
 from pymongo.periodic_executor import AsyncPeriodicExecutor
 from test.asynchronous import AsyncUnitTest, unittest
 
 _IS_SYNC = False
 
+_PYMONGO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(pymongo.__file__)))
+
 
 class TestAsyncPeriodicExecutor(AsyncUnitTest):
     def _make_executor(self, interval=30.0, min_interval=0.01, target=None, name="test"):
@@ -178,6 +190,76 @@ async def target():
             executor._task.exception()
         self.assertEqual(call_count, 2, "executor should run again after re-open")
 
+    async def test_open_non_daemon_thread(self):
+        # Subinterpreters disallow daemon threads; PeriodicExecutor.open()
+        # must fall back to starting the monitor thread as a non-daemon thread.
+        from pymongo.periodic_executor import PeriodicExecutor
+
+        def set_daemon(self, value):
+            raise RuntimeError("daemon threads are disallowed in subinterpreters")
+
+        daemon = threading.Thread.daemon
+        executor = PeriodicExecutor(
+            interval=30.0, min_interval=0.01, target=lambda: True, name="non-daemon"
+        )
+        self.addCleanup(executor.join, 2)
+        self.addCleanup(executor.close)
+        with mock.patch.object(threading.Thread, "daemon", property(daemon.fget, set_daemon)):
+            executor.open()
+            thread = executor._thread
+            assert thread is not None
+            self.assertFalse(thread.daemon, "thread must not be a daemon thread")
+
+    async def test_subinterpreter_shutdown(self):
+        if _interpreters is None:
+            self.skipTest("concurrent.interpreters requires Python 3.14+")
+            return
+
+        root = _PYMONGO_ROOT
+        # The async executor's open() starts a task, which requires a running
+        # event loop; synchro translates the rest of the block for the sync suite.
+        # Windows proactor loops cannot start in subinterpreters (set_wakeup_fd
+        # is main-interpreter only), so use a selector loop there.
+        run_stmt = (
+            "main()"
+            if _IS_SYNC
+            else 'asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop if sys.platform == "win32" else None)'
+        )
+        # The sync monitor signals its thread through the event; the async
+        # monitor yields to its loop instead. Synchro drops the async yield.
+        wait_stmt = "assert started.wait(10)" if _IS_SYNC else ""
+        code = textwrap.dedent(
+            f"""
+            import asyncio
+            import sys
+            import threading
+            sys.path.insert(0, {root!r})
+            from pymongo.periodic_executor import AsyncPeriodicExecutor
+
+            started = threading.Event()
+
+            async def target():
+                started.set()
+                return True
+
+            async def main():
+                executor = AsyncPeriodicExecutor(
+                    interval=30.0, min_interval=0.05, target=target, name="subinterp"
+                )
+                executor.open()
+                {wait_stmt}
+                await asyncio.sleep(0)
+
+            {run_stmt}
+            """
+        )
+        interp = _interpreters.create()
+        try:
+            interp.exec(code)
+        finally:
+            # Destroying the subinterpreter must not crash or hang.
+            interp.close()
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py
new file mode 100644
index 0000000000..f3e8c2883c
--- /dev/null
+++ b/test/asynchronous/test_subinterpreters.py
@@ -0,0 +1,274 @@
+# Copyright 2026-present MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test running pymongo in subinterpreters."""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+import textwrap
+import threading
+import uuid
+
+sys.path[0:0] = [""]
+
+try:
+    from concurrent import interpreters
+except ImportError:  # pragma: no cover - Python < 3.14
+    interpreters = None  # type: ignore[assignment]
+
+try:
+    from concurrent.futures import InterpreterPoolExecutor
+except ImportError:  # pragma: no cover - Python < 3.14
+    InterpreterPoolExecutor = None  # type: ignore[assignment,misc]
+
+from test.asynchronous import (
+    MONGODB_API_VERSION,
+    AsyncIntegrationTest,
+    async_client_context,
+    unittest,
+)
+
+_IS_SYNC = False
+
+
+class TestSubinterpreters(AsyncIntegrationTest):
+    @staticmethod
+    def _get_n(queue, n, errors):
+        try:
+            return sorted(queue.get(timeout=30) for _ in range(n))
+        except interpreters.QueueEmpty:
+            raise AssertionError(f"subinterpreters failed to run: {errors!r}") from None
+
+    @unittest.skipUnless(
+        sys.version_info >= (3, 14), "concurrent.interpreters requires Python 3.14+"
+    )
+    async def test_subinterpreters(self):
+        if interpreters is None:
+            self.skipTest("concurrent.interpreters is not available")
+
+        # Run live clients in multiple subinterpreters at once, mirroring the
+        # mod_wsgi test, which mounts the same app in two interpreters.
+        n_interpreters = 2
+        coll_name = f"subinterp-{uuid.uuid4().hex}"
+        self.addCleanup(self.db.drop_collection, coll_name)
+
+        ready = interpreters.create_queue()
+        release = interpreters.create_queue()
+        done = interpreters.create_queue()
+        # The async client's constructor starts a task, which requires a
+        # running event loop; synchro translates the rest of the block.
+        # Windows proactor loops cannot start in subinterpreters (set_wakeup_fd
+        # is main-interpreter only), so use a selector loop there.
+        run_stmt = (
+            "main()"
+            if _IS_SYNC
+            else 'asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop if sys.platform == "win32" else None)'
+        )
+        # The sync monitor reads the queue on its thread; the async monitor
+        # reads it off-loop so the client's tasks keep running.
+        release_stmt = (
+            "await asyncio.to_thread(release.get, timeout=60)"
+            if not _IS_SYNC
+            else "release.get(timeout=60)"
+        )
+        code = textwrap.dedent(
+            f"""
+            import asyncio
+            import sys
+            sys.path[:0] = path
+
+            from pymongo import AsyncMongoClient
+            from pymongo.server_api import ServerApi
+
+            async def main():
+                kwargs = {{"serverSelectionTimeoutMS": 30000}}
+                if api_version:
+                    kwargs["server_api"] = ServerApi(api_version)
+                client = AsyncMongoClient(uri, **kwargs)
+                collection = client.get_database(db_name).get_collection(coll_name)
+                await collection.insert_one({{"subinterp": i}})
+                assert await collection.find_one({{"subinterp": i}}) is not None
+                ready.put(i)
+                # Hold the client open until every interpreter has connected.
+                {release_stmt}
+                assert await collection.find_one({{"subinterp": i}}) is not None
+                done.put(i)
+
+            {run_stmt}
+            """
+        )
+
+        errors: list[BaseException] = []
+
+        def run(interp):
+            try:
+                interp.exec(code)
+            except BaseException as exc:
+                errors.append(exc)
+
+        uri = await async_client_context.uri
+        interps = []
+        threads = []
+        try:
+            for i in range(n_interpreters):
+                interp = interpreters.create()
+                interp.prepare_main(
+                    uri=uri,
+                    db_name=self.db.name,
+                    coll_name=coll_name,
+                    i=i,
+                    path=tuple(sys.path),
+                    api_version=MONGODB_API_VERSION,
+                    ready=ready,
+                    release=release,
+                    done=done,
+                )
+                interps.append(interp)
+                thread = threading.Thread(target=run, args=(interp,), name=f"subinterp-{i}")
+                threads.append(thread)
+                thread.start()
+
+            started = (
+                await asyncio.to_thread(self._get_n, ready, n_interpreters, errors)
+                if not _IS_SYNC
+                else self._get_n(ready, n_interpreters, errors)
+            )
+            self.assertEqual(started, list(range(n_interpreters)))
+            for _ in range(n_interpreters):
+                release.put(True)
+
+            for thread in threads:
+                if _IS_SYNC:
+                    thread.join(60)
+                else:
+                    await asyncio.to_thread(thread.join, 60)
+                self.assertFalse(thread.is_alive(), f"{thread.name} did not exit")
+
+            finished = (
+                await asyncio.to_thread(self._get_n, done, n_interpreters, errors)
+                if not _IS_SYNC
+                else self._get_n(done, n_interpreters, errors)
+            )
+            self.assertEqual(finished, list(range(n_interpreters)))
+            if errors:
+                self.fail(f"subinterpreter errors: {errors!r}")
+        finally:
+            # Unblock any interpreter still waiting, join its worker, then
+            # destroy them all.
+            for _ in range(n_interpreters):
+                release.put(True)
+            for thread in threads:
+                if _IS_SYNC:
+                    thread.join(60)
+                else:
+                    await asyncio.to_thread(thread.join, 60)
+            for interp in interps:
+                # Closing an idle interpreter runs threading._shutdown, which
+                # stops and joins pymongo's monitor threads; skip running ones
+                # to avoid masking errors.
+                if not interp.is_running():
+                    interp.close()
+
+        docs = await self.db[coll_name].find({}, {"subinterp": 1}).to_list()
+        found = sorted(doc["subinterp"] for doc in docs)
+        self.assertEqual(found, list(range(n_interpreters)))
+
+    @unittest.skipUnless(
+        sys.version_info >= (3, 14), "InterpreterPoolExecutor requires Python 3.14+"
+    )
+    async def test_interpreter_pool_executor(self):
+        if InterpreterPoolExecutor is None:
+            self.skipTest("InterpreterPoolExecutor is not available")
+
+        # The interpreters disallow daemon threads, so the sync client starts
+        # non-daemon monitor threads; the async client runs tasks on the
+        # interpreter's own event loop.
+        n_interpreters = 2
+        coll_name = f"interp-pool-{uuid.uuid4().hex}"
+        self.addCleanup(self.db.drop_collection, coll_name)
+
+        # The async client's constructor starts a task, which requires a
+        # running event loop; synchro translates the rest of the block.
+        # Windows proactor loops cannot start in subinterpreters (set_wakeup_fd
+        # is main-interpreter only), so use a selector loop there.
+        run_stmt = (
+            "main()"
+            if _IS_SYNC
+            else 'asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop if sys.platform == "win32" else None)'
+        )
+
+        # The worker's sys.path lacks the repo root, so submit the builtin exec
+        # and fix sys.path in the code.
+        code = textwrap.dedent(
+            f"""
+            import asyncio
+            import sys
+            sys.path[:0] = path
+
+            from pymongo import AsyncMongoClient
+            from pymongo.server_api import ServerApi
+
+            async def main():
+                kwargs = {{"serverSelectionTimeoutMS": 30000}}
+                if api_version:
+                    kwargs["server_api"] = ServerApi(api_version)
+                client = AsyncMongoClient(uri, **kwargs)
+                collection = client.get_database(db_name).get_collection(coll_name)
+                await collection.insert_one({{"interp-pool": i}})
+                assert await collection.find_one({{"interp-pool": i}}) is not None
+
+            {run_stmt}
+            """
+        )
+        executor = InterpreterPoolExecutor(max_workers=n_interpreters)
+        try:
+            uri = await async_client_context.uri
+            futures = [
+                executor.submit(
+                    exec,
+                    code,
+                    {
+                        "i": i,
+                        "path": tuple(sys.path),
+                        "uri": uri,
+                        "db_name": self.db.name,
+                        "coll_name": coll_name,
+                        "api_version": MONGODB_API_VERSION,
+                    },
+                )
+                for i in range(n_interpreters)
+            ]
+            for future in futures:
+                if _IS_SYNC:
+                    future.result(timeout=120)
+                else:
+                    # Keep the event loop free while waiting for the workers.
+                    await asyncio.to_thread(future.result, 120)
+        finally:
+            # Worker interpreter teardown joins monitor threads, which can
+            # take a while; keep the event loop free while waiting.
+            if _IS_SYNC:
+                executor.shutdown(wait=True)
+            else:
+                await asyncio.to_thread(executor.shutdown)
+
+        docs = await self.db[coll_name].find({}, {"interp-pool": 1}).to_list()
+        found = sorted(doc["interp-pool"] for doc in docs)
+        self.assertEqual(found, list(range(n_interpreters)))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/test/test_periodic_executor.py b/test/test_periodic_executor.py
index e6bbb31a72..d564042c30 100644
--- a/test/test_periodic_executor.py
+++ b/test/test_periodic_executor.py
@@ -17,17 +17,29 @@
 from __future__ import annotations
 
 import asyncio
+import importlib
+import os
 import sys
+import textwrap
 import threading
 import time
+from typing import Any
+from unittest import mock
+
+_interpreters: Any = None
+if sys.version_info >= (3, 14):
+    _interpreters = importlib.import_module("concurrent.interpreters")
 
 sys.path[0:0] = [""]
 
+import pymongo
 from pymongo.periodic_executor import PeriodicExecutor
 from test import UnitTest, unittest
 
 _IS_SYNC = True
 
+_PYMONGO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(pymongo.__file__)))
+
 
 class TestPeriodicExecutor(UnitTest):
     def _make_executor(self, interval=30.0, min_interval=0.01, target=None, name="test"):
@@ -178,6 +190,75 @@ def target():
             executor._task.exception()
         self.assertEqual(call_count, 2, "executor should run again after re-open")
 
+    def test_open_non_daemon_thread(self):
+        # Subinterpreters disallow daemon threads; PeriodicExecutor.open()
+        # must fall back to starting the monitor thread as a non-daemon thread.
+        from pymongo.periodic_executor import PeriodicExecutor
+
+        def set_daemon(self, value):
+            raise RuntimeError("daemon threads are disallowed in subinterpreters")
+
+        daemon = threading.Thread.daemon
+        executor = PeriodicExecutor(
+            interval=30.0, min_interval=0.01, target=lambda: True, name="non-daemon"
+        )
+        self.addCleanup(executor.join, 2)
+        self.addCleanup(executor.close)
+        with mock.patch.object(threading.Thread, "daemon", property(daemon.fget, set_daemon)):
+            executor.open()
+            thread = executor._thread
+            assert thread is not None
+            self.assertFalse(thread.daemon, "thread must not be a daemon thread")
+
+    def test_subinterpreter_shutdown(self):
+        if _interpreters is None:
+            self.skipTest("concurrent.interpreters requires Python 3.14+")
+            return
+
+        root = _PYMONGO_ROOT
+        # The async executor's open() starts a task, which requires a running
+        # event loop; synchro translates the rest of the block for the sync suite.
+        # Windows proactor loops cannot start in subinterpreters (set_wakeup_fd
+        # is main-interpreter only), so use a selector loop there.
+        run_stmt = (
+            "main()"
+            if _IS_SYNC
+            else 'asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop if sys.platform == "win32" else None)'
+        )
+        # The sync monitor signals its thread through the event; the async
+        # monitor yields to its loop instead. Synchro drops the async yield.
+        wait_stmt = "assert started.wait(10)" if _IS_SYNC else ""
+        code = textwrap.dedent(
+            f"""
+            import asyncio
+            import sys
+            import threading
+            sys.path.insert(0, {root!r})
+            from pymongo.periodic_executor import PeriodicExecutor
+
+            started = threading.Event()
+
+            def target():
+                started.set()
+                return True
+
+            def main():
+                executor = PeriodicExecutor(
+                    interval=30.0, min_interval=0.05, target=target, name="subinterp"
+                )
+                executor.open()
+                {wait_stmt}
+
+            {run_stmt}
+            """
+        )
+        interp = _interpreters.create()
+        try:
+            interp.exec(code)
+        finally:
+            # Destroying the subinterpreter must not crash or hang.
+            interp.close()
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py
new file mode 100644
index 0000000000..e16d8be642
--- /dev/null
+++ b/test/test_subinterpreters.py
@@ -0,0 +1,274 @@
+# Copyright 2026-present MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test running pymongo in subinterpreters."""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+import textwrap
+import threading
+import uuid
+
+sys.path[0:0] = [""]
+
+try:
+    from concurrent import interpreters
+except ImportError:  # pragma: no cover - Python < 3.14
+    interpreters = None  # type: ignore[assignment]
+
+try:
+    from concurrent.futures import InterpreterPoolExecutor
+except ImportError:  # pragma: no cover - Python < 3.14
+    InterpreterPoolExecutor = None  # type: ignore[assignment,misc]
+
+from test import (
+    MONGODB_API_VERSION,
+    IntegrationTest,
+    client_context,
+    unittest,
+)
+
+_IS_SYNC = True
+
+
+class TestSubinterpreters(IntegrationTest):
+    @staticmethod
+    def _get_n(queue, n, errors):
+        try:
+            return sorted(queue.get(timeout=30) for _ in range(n))
+        except interpreters.QueueEmpty:
+            raise AssertionError(f"subinterpreters failed to run: {errors!r}") from None
+
+    @unittest.skipUnless(
+        sys.version_info >= (3, 14), "concurrent.interpreters requires Python 3.14+"
+    )
+    def test_subinterpreters(self):
+        if interpreters is None:
+            self.skipTest("concurrent.interpreters is not available")
+
+        # Run live clients in multiple subinterpreters at once, mirroring the
+        # mod_wsgi test, which mounts the same app in two interpreters.
+        n_interpreters = 2
+        coll_name = f"subinterp-{uuid.uuid4().hex}"
+        self.addCleanup(self.db.drop_collection, coll_name)
+
+        ready = interpreters.create_queue()
+        release = interpreters.create_queue()
+        done = interpreters.create_queue()
+        # The async client's constructor starts a task, which requires a
+        # running event loop; synchro translates the rest of the block.
+        # Windows proactor loops cannot start in subinterpreters (set_wakeup_fd
+        # is main-interpreter only), so use a selector loop there.
+        run_stmt = (
+            "main()"
+            if _IS_SYNC
+            else 'asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop if sys.platform == "win32" else None)'
+        )
+        # The sync monitor reads the queue on its thread; the async monitor
+        # reads it off-loop so the client's tasks keep running.
+        release_stmt = (
+            "asyncio.to_thread(release.get, timeout=60)"
+            if not _IS_SYNC
+            else "release.get(timeout=60)"
+        )
+        code = textwrap.dedent(
+            f"""
+            import asyncio
+            import sys
+            sys.path[:0] = path
+
+            from pymongo import MongoClient
+            from pymongo.server_api import ServerApi
+
+            def main():
+                kwargs = {{"serverSelectionTimeoutMS": 30000}}
+                if api_version:
+                    kwargs["server_api"] = ServerApi(api_version)
+                client = MongoClient(uri, **kwargs)
+                collection = client.get_database(db_name).get_collection(coll_name)
+                collection.insert_one({{"subinterp": i}})
+                assert collection.find_one({{"subinterp": i}}) is not None
+                ready.put(i)
+                # Hold the client open until every interpreter has connected.
+                {release_stmt}
+                assert collection.find_one({{"subinterp": i}}) is not None
+                done.put(i)
+
+            {run_stmt}
+            """
+        )
+
+        errors: list[BaseException] = []
+
+        def run(interp):
+            try:
+                interp.exec(code)
+            except BaseException as exc:
+                errors.append(exc)
+
+        uri = client_context.uri
+        interps = []
+        threads = []
+        try:
+            for i in range(n_interpreters):
+                interp = interpreters.create()
+                interp.prepare_main(
+                    uri=uri,
+                    db_name=self.db.name,
+                    coll_name=coll_name,
+                    i=i,
+                    path=tuple(sys.path),
+                    api_version=MONGODB_API_VERSION,
+                    ready=ready,
+                    release=release,
+                    done=done,
+                )
+                interps.append(interp)
+                thread = threading.Thread(target=run, args=(interp,), name=f"subinterp-{i}")
+                threads.append(thread)
+                thread.start()
+
+            started = (
+                asyncio.to_thread(self._get_n, ready, n_interpreters, errors)
+                if not _IS_SYNC
+                else self._get_n(ready, n_interpreters, errors)
+            )
+            self.assertEqual(started, list(range(n_interpreters)))
+            for _ in range(n_interpreters):
+                release.put(True)
+
+            for thread in threads:
+                if _IS_SYNC:
+                    thread.join(60)
+                else:
+                    asyncio.to_thread(thread.join, 60)
+                self.assertFalse(thread.is_alive(), f"{thread.name} did not exit")
+
+            finished = (
+                asyncio.to_thread(self._get_n, done, n_interpreters, errors)
+                if not _IS_SYNC
+                else self._get_n(done, n_interpreters, errors)
+            )
+            self.assertEqual(finished, list(range(n_interpreters)))
+            if errors:
+                self.fail(f"subinterpreter errors: {errors!r}")
+        finally:
+            # Unblock any interpreter still waiting, join its worker, then
+            # destroy them all.
+            for _ in range(n_interpreters):
+                release.put(True)
+            for thread in threads:
+                if _IS_SYNC:
+                    thread.join(60)
+                else:
+                    asyncio.to_thread(thread.join, 60)
+            for interp in interps:
+                # Closing an idle interpreter runs threading._shutdown, which
+                # stops and joins pymongo's monitor threads; skip running ones
+                # to avoid masking errors.
+                if not interp.is_running():
+                    interp.close()
+
+        docs = self.db[coll_name].find({}, {"subinterp": 1}).to_list()
+        found = sorted(doc["subinterp"] for doc in docs)
+        self.assertEqual(found, list(range(n_interpreters)))
+
+    @unittest.skipUnless(
+        sys.version_info >= (3, 14), "InterpreterPoolExecutor requires Python 3.14+"
+    )
+    def test_interpreter_pool_executor(self):
+        if InterpreterPoolExecutor is None:
+            self.skipTest("InterpreterPoolExecutor is not available")
+
+        # The interpreters disallow daemon threads, so the sync client starts
+        # non-daemon monitor threads; the async client runs tasks on the
+        # interpreter's own event loop.
+        n_interpreters = 2
+        coll_name = f"interp-pool-{uuid.uuid4().hex}"
+        self.addCleanup(self.db.drop_collection, coll_name)
+
+        # The async client's constructor starts a task, which requires a
+        # running event loop; synchro translates the rest of the block.
+        # Windows proactor loops cannot start in subinterpreters (set_wakeup_fd
+        # is main-interpreter only), so use a selector loop there.
+        run_stmt = (
+            "main()"
+            if _IS_SYNC
+            else 'asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop if sys.platform == "win32" else None)'
+        )
+
+        # The worker's sys.path lacks the repo root, so submit the builtin exec
+        # and fix sys.path in the code.
+        code = textwrap.dedent(
+            f"""
+            import asyncio
+            import sys
+            sys.path[:0] = path
+
+            from pymongo import MongoClient
+            from pymongo.server_api import ServerApi
+
+            def main():
+                kwargs = {{"serverSelectionTimeoutMS": 30000}}
+                if api_version:
+                    kwargs["server_api"] = ServerApi(api_version)
+                client = MongoClient(uri, **kwargs)
+                collection = client.get_database(db_name).get_collection(coll_name)
+                collection.insert_one({{"interp-pool": i}})
+                assert collection.find_one({{"interp-pool": i}}) is not None
+
+            {run_stmt}
+            """
+        )
+        executor = InterpreterPoolExecutor(max_workers=n_interpreters)
+        try:
+            uri = client_context.uri
+            futures = [
+                executor.submit(
+                    exec,
+                    code,
+                    {
+                        "i": i,
+                        "path": tuple(sys.path),
+                        "uri": uri,
+                        "db_name": self.db.name,
+                        "coll_name": coll_name,
+                        "api_version": MONGODB_API_VERSION,
+                    },
+                )
+                for i in range(n_interpreters)
+            ]
+            for future in futures:
+                if _IS_SYNC:
+                    future.result(timeout=120)
+                else:
+                    # Keep the event loop free while waiting for the workers.
+                    asyncio.to_thread(future.result, 120)
+        finally:
+            # Worker interpreter teardown joins monitor threads, which can
+            # take a while; keep the event loop free while waiting.
+            if _IS_SYNC:
+                executor.shutdown(wait=True)
+            else:
+                asyncio.to_thread(executor.shutdown)
+
+        docs = self.db[coll_name].find({}, {"interp-pool": 1}).to_list()
+        found = sorted(doc["interp-pool"] for doc in docs)
+        self.assertEqual(found, list(range(n_interpreters)))
+
+
+if __name__ == "__main__":
+    unittest.main()