diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py
index 471a013eb9..168d2a07fb 100644
--- a/pymongo/_telemetry.py
+++ b/pymongo/_telemetry.py
@@ -86,7 +86,7 @@ def _emit_log(self, message: _CommandStatusMessage, **extra: Any) -> None:
             commandName=self._name,
             databaseName=self._dbname,
             requestId=self._request_id,
-            operationId=self._request_id,
+            operationId=self._op_id if self._op_id is not None else self._request_id,
             driverConnectionId=self._conn.id,
             serverConnectionId=self._conn.server_connection_id,
             serverHost=self._conn.address[0],
diff --git a/pymongo/asynchronous/command_runner.py b/pymongo/asynchronous/command_runner.py
index 3fc048fc39..7900a6c958 100644
--- a/pymongo/asynchronous/command_runner.py
+++ b/pymongo/asynchronous/command_runner.py
@@ -325,6 +325,7 @@ async def run_cursor_command(
         more_to_come=more_to_come,
         unpack_res=unpack_res,
         cursor_id=cursor_id,
+        op_id=conn.op_id,
     )
 
 
@@ -428,6 +429,7 @@ async def run_command(
         codec_options=codec_options,
         user_fields=user_fields,
         orig=orig,
+        op_id=conn.op_id,
         check=check,
         allowable_errors=allowable_errors,
         parse_write_concern_error=parse_write_concern_error,
diff --git a/pymongo/asynchronous/helpers.py b/pymongo/asynchronous/helpers.py
index 0ee99a441a..03215af785 100644
--- a/pymongo/asynchronous/helpers.py
+++ b/pymongo/asynchronous/helpers.py
@@ -68,7 +68,13 @@ async def inner(*args: Any, **kwargs: Any) -> Any:
                         conn = arg.conn  # type: ignore[assignment]
                         break
                 if conn:
-                    await conn.authenticate(reauthenticate=True)
+                    # Don't let reauth's auth commands inherit the in-flight op's id.
+                    prev_op_id = conn.op_id
+                    conn.op_id = None
+                    try:
+                        await conn.authenticate(reauthenticate=True)
+                    finally:
+                        conn.op_id = prev_op_id
                 else:
                     raise
                 return await func(*args, **kwargs)
diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py
index c745720ed9..8cf6d4a05f 100644
--- a/pymongo/asynchronous/mongo_client.py
+++ b/pymongo/asynchronous/mongo_client.py
@@ -95,7 +95,7 @@
     _log_client_error,
     _log_or_warn,
 )
-from pymongo.message import _CursorAddress, _GetMore, _Query
+from pymongo.message import _CursorAddress, _GetMore, _Query, _randint
 from pymongo.monitoring import ConnectionClosedReason, _EventListeners
 from pymongo.operations import (
     DeleteMany,
@@ -1793,10 +1793,13 @@ async def _checkout(
         async with _MongoClientErrorHandler(self, server, session) as err_handler:
             # Reuse the pinned connection, if it exists.
             if in_txn and session and session._pinned_connection:
-                err_handler.contribute_socket(session._pinned_connection)
-                yield session._pinned_connection
+                conn = session._pinned_connection
+                conn.op_id = None
+                err_handler.contribute_socket(conn)
+                yield conn
                 return
             async with await server.checkout(handler=err_handler) as conn:
+                conn.op_id = None  # Only retryable read/write logic sets an op_id.
                 # Pin this session to the selected server or connection.
                 if (
                     in_txn
@@ -1837,6 +1840,8 @@ async def _select_server(
             be pinned to a mongos server address.
           - `address` (optional): Address when sending a message
             to a specific server, used for getMore.
+          - `operation_id` (optional): Stable operation id shared across retries,
+            used for command monitoring.
         """
         try:
             topology = await self._get_topology()
@@ -1933,6 +1938,8 @@ async def _run_operation(
             async with operation.conn_mgr._lock:
                 async with _MongoClientErrorHandler(self, server, operation.session) as err_handler:  # type: ignore[arg-type]
                     err_handler.contribute_socket(operation.conn_mgr.conn)
+                    # Non-retryable getMore on a pinned conn; no shared op_id.
+                    operation.conn_mgr.conn.op_id = None
                     return await run_with_conn(
                         operation.conn_mgr.conn, operation, operation.read_preference
                     )
@@ -2012,6 +2019,7 @@ async def _retry_internal(
         :param retryable: If the operation should be retried once, defaults to None
         :param is_run_command: If this is a runCommand operation, defaults to False
         :param is_aggregate_write: If this is a aggregate operation with a write, defaults to False.
+        :param operation_id: Stable operation id shared across retries, defaults to None
 
         :return: Output of the calling func()
         """
@@ -2058,6 +2066,7 @@ async def _retryable_read(
             (may not always be supported even if supplied), defaults to False
         :param is_run_command: If this is a runCommand operation, defaults to False.
         :param is_aggregate_write: If this is a aggregate operation with a write, defaults to False.
+        :param operation_id: Stable operation id shared across retries, defaults to None
         """
 
         # Ensure that the client supports retrying on reads and there is no session in
@@ -2101,6 +2110,7 @@ async def _retryable_write(
         :param session: Client session we will use to execute write operation
         :param operation: The name of the operation that the server is being selected for
         :param bulk: bulk abstraction to execute operations in bulk, defaults to None
+        :param operation_id: Stable operation id shared across retries, defaults to None
         """
         async with self._tmp_session(session) as s:
             return await self._retry_with_session(retryable, func, s, bulk, operation, operation_id)
@@ -2217,6 +2227,8 @@ async def _kill_cursor_impl(
         namespace = address.namespace
         db, coll = namespace.split(".", 1)
         spec = {"killCursors": coll, "cursors": cursor_ids}
+        # killCursors is its own op; drop any op_id left on a pinned cursor conn.
+        conn.op_id = None
         await conn.command(db, spec, session=session, client=self)
 
     async def _process_kill_cursors(self) -> None:
@@ -2784,7 +2796,7 @@ def __init__(
         self._server: Server = None  # type: ignore
         self._deprioritized_servers: list[Server] = []
         self._operation = operation
-        self._operation_id = operation_id
+        self._operation_id = operation_id if operation_id is not None else _randint()
         self._attempt_number = 0
         self._is_run_command = is_run_command
         self._is_aggregate_write = is_aggregate_write
@@ -2990,6 +3002,7 @@ async def _write(self) -> T:
             is_mongos = False
             self._server = await self._get_server()
             async with self._client._checkout(self._server, self._session) as conn:
+                conn.op_id = self._operation_id
                 max_wire_version = conn.max_wire_version
                 sessions_supported = (
                     self._session
@@ -3029,6 +3042,7 @@ async def _read(self) -> T:
             conn,
             read_pref,
         ):
+            conn.op_id = self._operation_id
             if self._retrying and not self._retryable and not self._always_retryable:
                 self._check_last_error()
             if self._retrying:
diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py
index 212f35f0d1..ab99eb150d 100644
--- a/pymongo/asynchronous/pool.py
+++ b/pymongo/asynchronous/pool.py
@@ -177,6 +177,13 @@ def __init__(
         self.creation_time = time.monotonic()
         # For gossiping $clusterTime from the connection handshake to the client.
         self._cluster_time = None
+        # APM operation id of the op currently using this connection, read by
+        # command(). Kept here for simplicity: retries reuse it without threading
+        # an op_id kwarg through every command. Reset to None on checkout; only
+        # the retryable read/write logic sets a real value. An op running a
+        # command directly on a pinned connection (see killCursors, reauth) must
+        # clear it so it isn't attributed to an unrelated command.
+        self.op_id: Optional[int] = None
 
     def set_conn_timeout(self, timeout: Optional[float]) -> None:
         """Cache last timeout to avoid duplicate calls to conn.settimeout."""
diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py
index eec58722bb..31fdd72ad8 100644
--- a/pymongo/synchronous/command_runner.py
+++ b/pymongo/synchronous/command_runner.py
@@ -325,6 +325,7 @@ def run_cursor_command(
         more_to_come=more_to_come,
         unpack_res=unpack_res,
         cursor_id=cursor_id,
+        op_id=conn.op_id,
     )
 
 
@@ -428,6 +429,7 @@ def run_command(
         codec_options=codec_options,
         user_fields=user_fields,
         orig=orig,
+        op_id=conn.op_id,
         check=check,
         allowable_errors=allowable_errors,
         parse_write_concern_error=parse_write_concern_error,
diff --git a/pymongo/synchronous/helpers.py b/pymongo/synchronous/helpers.py
index f24d72aea9..877e2c2da5 100644
--- a/pymongo/synchronous/helpers.py
+++ b/pymongo/synchronous/helpers.py
@@ -68,7 +68,13 @@ def inner(*args: Any, **kwargs: Any) -> Any:
                         conn = arg.conn  # type: ignore[assignment]
                         break
                 if conn:
-                    conn.authenticate(reauthenticate=True)
+                    # Don't let reauth's auth commands inherit the in-flight op's id.
+                    prev_op_id = conn.op_id
+                    conn.op_id = None
+                    try:
+                        conn.authenticate(reauthenticate=True)
+                    finally:
+                        conn.op_id = prev_op_id
                 else:
                     raise
                 return func(*args, **kwargs)
diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py
index 90912dd6f5..0ff845321d 100644
--- a/pymongo/synchronous/mongo_client.py
+++ b/pymongo/synchronous/mongo_client.py
@@ -85,7 +85,7 @@
     _log_client_error,
     _log_or_warn,
 )
-from pymongo.message import _CursorAddress, _GetMore, _Query
+from pymongo.message import _CursorAddress, _GetMore, _Query, _randint
 from pymongo.monitoring import ConnectionClosedReason, _EventListeners
 from pymongo.operations import (
     DeleteMany,
@@ -1790,10 +1790,13 @@ def _checkout(
         with _MongoClientErrorHandler(self, server, session) as err_handler:
             # Reuse the pinned connection, if it exists.
             if in_txn and session and session._pinned_connection:
-                err_handler.contribute_socket(session._pinned_connection)
-                yield session._pinned_connection
+                conn = session._pinned_connection
+                conn.op_id = None
+                err_handler.contribute_socket(conn)
+                yield conn
                 return
             with server.checkout(handler=err_handler) as conn:
+                conn.op_id = None  # Only retryable read/write logic sets an op_id.
                 # Pin this session to the selected server or connection.
                 if (
                     in_txn
@@ -1834,6 +1837,8 @@ def _select_server(
             be pinned to a mongos server address.
           - `address` (optional): Address when sending a message
             to a specific server, used for getMore.
+          - `operation_id` (optional): Stable operation id shared across retries,
+            used for command monitoring.
         """
         try:
             topology = self._get_topology()
@@ -1930,6 +1935,8 @@ def _run_operation(
             with operation.conn_mgr._lock:
                 with _MongoClientErrorHandler(self, server, operation.session) as err_handler:  # type: ignore[arg-type]
                     err_handler.contribute_socket(operation.conn_mgr.conn)
+                    # Non-retryable getMore on a pinned conn; no shared op_id.
+                    operation.conn_mgr.conn.op_id = None
                     return run_with_conn(
                         operation.conn_mgr.conn, operation, operation.read_preference
                     )
@@ -2009,6 +2016,7 @@ def _retry_internal(
         :param retryable: If the operation should be retried once, defaults to None
         :param is_run_command: If this is a runCommand operation, defaults to False
         :param is_aggregate_write: If this is a aggregate operation with a write, defaults to False.
+        :param operation_id: Stable operation id shared across retries, defaults to None
 
         :return: Output of the calling func()
         """
@@ -2055,6 +2063,7 @@ def _retryable_read(
             (may not always be supported even if supplied), defaults to False
         :param is_run_command: If this is a runCommand operation, defaults to False.
         :param is_aggregate_write: If this is a aggregate operation with a write, defaults to False.
+        :param operation_id: Stable operation id shared across retries, defaults to None
         """
 
         # Ensure that the client supports retrying on reads and there is no session in
@@ -2098,6 +2107,7 @@ def _retryable_write(
         :param session: Client session we will use to execute write operation
         :param operation: The name of the operation that the server is being selected for
         :param bulk: bulk abstraction to execute operations in bulk, defaults to None
+        :param operation_id: Stable operation id shared across retries, defaults to None
         """
         with self._tmp_session(session) as s:
             return self._retry_with_session(retryable, func, s, bulk, operation, operation_id)
@@ -2214,6 +2224,8 @@ def _kill_cursor_impl(
         namespace = address.namespace
         db, coll = namespace.split(".", 1)
         spec = {"killCursors": coll, "cursors": cursor_ids}
+        # killCursors is its own op; drop any op_id left on a pinned cursor conn.
+        conn.op_id = None
         conn.command(db, spec, session=session, client=self)
 
     def _process_kill_cursors(self) -> None:
@@ -2775,7 +2787,7 @@ def __init__(
         self._server: Server = None  # type: ignore
         self._deprioritized_servers: list[Server] = []
         self._operation = operation
-        self._operation_id = operation_id
+        self._operation_id = operation_id if operation_id is not None else _randint()
         self._attempt_number = 0
         self._is_run_command = is_run_command
         self._is_aggregate_write = is_aggregate_write
@@ -2981,6 +2993,7 @@ def _write(self) -> T:
             is_mongos = False
             self._server = self._get_server()
             with self._client._checkout(self._server, self._session) as conn:
+                conn.op_id = self._operation_id
                 max_wire_version = conn.max_wire_version
                 sessions_supported = (
                     self._session
@@ -3020,6 +3033,7 @@ def _read(self) -> T:
             conn,
             read_pref,
         ):
+            conn.op_id = self._operation_id
             if self._retrying and not self._retryable and not self._always_retryable:
                 self._check_last_error()
             if self._retrying:
diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py
index f34a3341f3..84c7432079 100644
--- a/pymongo/synchronous/pool.py
+++ b/pymongo/synchronous/pool.py
@@ -177,6 +177,13 @@ def __init__(
         self.creation_time = time.monotonic()
         # For gossiping $clusterTime from the connection handshake to the client.
         self._cluster_time = None
+        # APM operation id of the op currently using this connection, read by
+        # command(). Kept here for simplicity: retries reuse it without threading
+        # an op_id kwarg through every command. Reset to None on checkout; only
+        # the retryable read/write logic sets a real value. An op running a
+        # command directly on a pinned connection (see killCursors, reauth) must
+        # clear it so it isn't attributed to an unrelated command.
+        self.op_id: Optional[int] = None
 
     def set_conn_timeout(self, timeout: Optional[float]) -> None:
         """Cache last timeout to avoid duplicate calls to conn.settimeout."""
diff --git a/test/asynchronous/test_operation_id_retry.py b/test/asynchronous/test_operation_id_retry.py
new file mode 100644
index 0000000000..0fd91a2611
--- /dev/null
+++ b/test/asynchronous/test_operation_id_retry.py
@@ -0,0 +1,156 @@
+# 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 that retry attempts reuse a single stable CommandStartedEvent.operation_id."""
+
+from __future__ import annotations
+
+import sys
+
+sys.path[0:0] = [""]
+
+import pymongo
+from pymongo.errors import ConnectionFailure
+from pymongo.operations import InsertOne
+from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest
+from test.utils_shared import AllowListEventListener
+
+_IS_SYNC = False
+
+_APP_NAME = "operationIdRetryTest"
+
+_RETRYABLE_WRITES = [
+    ("insert", lambda c: c.insert_one({"_id": 100})),
+    ("update", lambda c: c.update_one({"_id": 1}, {"$set": {"y": 1}})),
+    ("update", lambda c: c.replace_one({"_id": 2}, {"x": 9})),
+    ("delete", lambda c: c.delete_one({"_id": 3})),
+    ("findAndModify", lambda c: c.find_one_and_update({"_id": 4}, {"$set": {"y": 2}})),
+    ("insert", lambda c: c.bulk_write([InsertOne({"_id": 200}), InsertOne({"_id": 201})])),
+]
+
+
+_RETRYABLE_READS = [
+    ("find", lambda c: c.find({"x": 1}).to_list()),
+    ("find", lambda c: c.find_one({"_id": 1})),
+    ("aggregate", lambda c: _agg(c)),
+    ("aggregate", lambda c: c.count_documents({"x": 1})),
+    ("distinct", lambda c: c.distinct("x")),
+    ("listIndexes", lambda c: _list_indexes(c)),
+]
+
+_COMMANDS = {name for name, _ in _RETRYABLE_WRITES + _RETRYABLE_READS}
+
+
+async def _agg(coll):
+    cursor = await coll.aggregate([{"$match": {"x": 1}}])
+    return await cursor.to_list()
+
+
+async def _list_indexes(coll):
+    cursor = await coll.list_indexes()
+    return await cursor.to_list()
+
+
+class TestOperationIdRetry(AsyncIntegrationTest):
+    RETRIES = 2  # fail this many attempts; the (RETRIES + 1)th succeeds.
+
+    @async_client_context.require_failCommand_fail_point
+    async def asyncSetUp(self) -> None:
+        await super().asyncSetUp()
+        self.listener = AllowListEventListener(*_COMMANDS)
+        self.client = await self.async_rs_or_single_client(
+            event_listeners=[self.listener], appname=_APP_NAME
+        )
+        self.coll = self.client.pymongo_test.test_operation_id_retry
+
+    async def _seed(self):
+        await self.coll.drop()
+        await self.coll.insert_many([{"_id": i, "x": i % 3} for i in range(5)])
+        await self.coll.create_index("x")
+
+    async def _run_under_failpoint(self, command_name, action, times, expected_error=None):
+        """Seed, force ``times`` closeConnection failures of ``command_name``,
+        run ``action``, and return its ``(started, failed, succeeded)`` events."""
+        await self._seed()
+        self.listener.reset()
+        fail_point = {
+            "mode": {"times": times},
+            "data": {
+                "failCommands": [command_name],
+                "closeConnection": True,
+                "appName": _APP_NAME,
+            },
+        }
+        async with self.fail_point(fail_point):
+            # A CSOT timeout lets a single operation retry more than once.
+            with pymongo.timeout(60):
+                if expected_error is not None:
+                    with self.assertRaises(expected_error):
+                        await action(self.coll)
+                else:
+                    await action(self.coll)
+
+        def of(events):
+            return [e for e in events if e.command_name == command_name]
+
+        return (
+            of(self.listener.started_events),
+            of(self.listener.failed_events),
+            of(self.listener.succeeded_events),
+        )
+
+    async def _check_stable_operation_id(self, command_name, action, retries):
+        """Assert every command event for ``command_name`` shares one integer operation_id."""
+        started, failed, succeeded = await self._run_under_failpoint(command_name, action, retries)
+        op_ids = [e.operation_id for e in started + failed + succeeded]
+
+        self.assertEqual(len(started), retries + 1, "expected one started event per attempt")
+        self.assertEqual(len(failed), retries)
+        self.assertEqual(len(succeeded), 1)
+        self.assertTrue(all(isinstance(op, int) for op in op_ids))
+        self.assertEqual(
+            len(set(op_ids)),
+            1,
+            f"operation_id not stable across retries for {command_name}: {op_ids}",
+        )
+
+    @async_client_context.require_no_standalone
+    async def test_retryable_writes_reuse_operation_id(self):
+        for command_name, action in _RETRYABLE_WRITES:
+            with self.subTest(command=command_name):
+                await self._check_stable_operation_id(command_name, action, self.RETRIES)
+
+    async def test_retryable_reads_reuse_operation_id(self):
+        for command_name, action in _RETRYABLE_READS:
+            with self.subTest(command=command_name):
+                await self._check_stable_operation_id(command_name, action, self.RETRIES)
+
+    @async_client_context.require_no_standalone
+    async def test_non_retryable_write_is_not_retried(self):
+        # Multi-document writes are not retryable: a single network error must
+        # surface immediately, with exactly one attempt.
+        for command_name, action in [
+            ("update", lambda c: c.update_many({"x": 1}, {"$set": {"z": 1}})),
+            ("delete", lambda c: c.delete_many({"x": 2})),
+        ]:
+            with self.subTest(command=command_name):
+                started, _, _ = await self._run_under_failpoint(
+                    command_name, action, times=1, expected_error=ConnectionFailure
+                )
+                self.assertEqual(len(started), 1, "must not retry")
+                self.assertIsInstance(started[0].operation_id, int)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/test/test_operation_id_retry.py b/test/test_operation_id_retry.py
new file mode 100644
index 0000000000..b73b4b24aa
--- /dev/null
+++ b/test/test_operation_id_retry.py
@@ -0,0 +1,154 @@
+# 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 that retry attempts reuse a single stable CommandStartedEvent.operation_id."""
+
+from __future__ import annotations
+
+import sys
+
+sys.path[0:0] = [""]
+
+import pymongo
+from pymongo.errors import ConnectionFailure
+from pymongo.operations import InsertOne
+from test import IntegrationTest, client_context, unittest
+from test.utils_shared import AllowListEventListener
+
+_IS_SYNC = True
+
+_APP_NAME = "operationIdRetryTest"
+
+_RETRYABLE_WRITES = [
+    ("insert", lambda c: c.insert_one({"_id": 100})),
+    ("update", lambda c: c.update_one({"_id": 1}, {"$set": {"y": 1}})),
+    ("update", lambda c: c.replace_one({"_id": 2}, {"x": 9})),
+    ("delete", lambda c: c.delete_one({"_id": 3})),
+    ("findAndModify", lambda c: c.find_one_and_update({"_id": 4}, {"$set": {"y": 2}})),
+    ("insert", lambda c: c.bulk_write([InsertOne({"_id": 200}), InsertOne({"_id": 201})])),
+]
+
+
+_RETRYABLE_READS = [
+    ("find", lambda c: c.find({"x": 1}).to_list()),
+    ("find", lambda c: c.find_one({"_id": 1})),
+    ("aggregate", lambda c: _agg(c)),
+    ("aggregate", lambda c: c.count_documents({"x": 1})),
+    ("distinct", lambda c: c.distinct("x")),
+    ("listIndexes", lambda c: _list_indexes(c)),
+]
+
+_COMMANDS = {name for name, _ in _RETRYABLE_WRITES + _RETRYABLE_READS}
+
+
+def _agg(coll):
+    cursor = coll.aggregate([{"$match": {"x": 1}}])
+    return cursor.to_list()
+
+
+def _list_indexes(coll):
+    cursor = coll.list_indexes()
+    return cursor.to_list()
+
+
+class TestOperationIdRetry(IntegrationTest):
+    RETRIES = 2  # fail this many attempts; the (RETRIES + 1)th succeeds.
+
+    @client_context.require_failCommand_fail_point
+    def setUp(self) -> None:
+        super().setUp()
+        self.listener = AllowListEventListener(*_COMMANDS)
+        self.client = self.rs_or_single_client(event_listeners=[self.listener], appname=_APP_NAME)
+        self.coll = self.client.pymongo_test.test_operation_id_retry
+
+    def _seed(self):
+        self.coll.drop()
+        self.coll.insert_many([{"_id": i, "x": i % 3} for i in range(5)])
+        self.coll.create_index("x")
+
+    def _run_under_failpoint(self, command_name, action, times, expected_error=None):
+        """Seed, force ``times`` closeConnection failures of ``command_name``,
+        run ``action``, and return its ``(started, failed, succeeded)`` events."""
+        self._seed()
+        self.listener.reset()
+        fail_point = {
+            "mode": {"times": times},
+            "data": {
+                "failCommands": [command_name],
+                "closeConnection": True,
+                "appName": _APP_NAME,
+            },
+        }
+        with self.fail_point(fail_point):
+            # A CSOT timeout lets a single operation retry more than once.
+            with pymongo.timeout(60):
+                if expected_error is not None:
+                    with self.assertRaises(expected_error):
+                        action(self.coll)
+                else:
+                    action(self.coll)
+
+        def of(events):
+            return [e for e in events if e.command_name == command_name]
+
+        return (
+            of(self.listener.started_events),
+            of(self.listener.failed_events),
+            of(self.listener.succeeded_events),
+        )
+
+    def _check_stable_operation_id(self, command_name, action, retries):
+        """Assert every command event for ``command_name`` shares one integer operation_id."""
+        started, failed, succeeded = self._run_under_failpoint(command_name, action, retries)
+        op_ids = [e.operation_id for e in started + failed + succeeded]
+
+        self.assertEqual(len(started), retries + 1, "expected one started event per attempt")
+        self.assertEqual(len(failed), retries)
+        self.assertEqual(len(succeeded), 1)
+        self.assertTrue(all(isinstance(op, int) for op in op_ids))
+        self.assertEqual(
+            len(set(op_ids)),
+            1,
+            f"operation_id not stable across retries for {command_name}: {op_ids}",
+        )
+
+    @client_context.require_no_standalone
+    def test_retryable_writes_reuse_operation_id(self):
+        for command_name, action in _RETRYABLE_WRITES:
+            with self.subTest(command=command_name):
+                self._check_stable_operation_id(command_name, action, self.RETRIES)
+
+    def test_retryable_reads_reuse_operation_id(self):
+        for command_name, action in _RETRYABLE_READS:
+            with self.subTest(command=command_name):
+                self._check_stable_operation_id(command_name, action, self.RETRIES)
+
+    @client_context.require_no_standalone
+    def test_non_retryable_write_is_not_retried(self):
+        # Multi-document writes are not retryable: a single network error must
+        # surface immediately, with exactly one attempt.
+        for command_name, action in [
+            ("update", lambda c: c.update_many({"x": 1}, {"$set": {"z": 1}})),
+            ("delete", lambda c: c.delete_many({"x": 2})),
+        ]:
+            with self.subTest(command=command_name):
+                started, _, _ = self._run_under_failpoint(
+                    command_name, action, times=1, expected_error=ConnectionFailure
+                )
+                self.assertEqual(len(started), 1, "must not retry")
+                self.assertIsInstance(started[0].operation_id, int)
+
+
+if __name__ == "__main__":
+    unittest.main()