diff --git a/driver/src/client/executor.rs b/driver/src/client/executor.rs
index 3fb0f25b6..b96e6deb5 100644
--- a/driver/src/client/executor.rs
+++ b/driver/src/client/executor.rs
@@ -839,7 +839,7 @@ impl Client {
                         session.transaction.state,
                         TransactionState::None | TransactionState::Starting
                     )
-                    && op.read_concern().supported()
+                    && (op.read_concern().supported() || op.is_after_cluster_time_write())
                 {
                     cmd.set_after_cluster_time(session);
                 }
diff --git a/driver/src/operation.rs b/driver/src/operation.rs
index 9e7e189a9..7a81593a7 100644
--- a/driver/src/operation.rs
+++ b/driver/src/operation.rs
@@ -193,6 +193,9 @@ pub(crate) trait Operation {
     /// The noun to this operation's verb.
     fn target(&self) -> OperationTarget;
 
+    /// Whether this is a write operation that needs `afterClusterTime` set.
+    fn is_after_cluster_time_write(&self) -> bool;
+
     #[cfg(feature = "opentelemetry")]
     type Otel: crate::otel::OtelWitness<Op = Self>;
 
@@ -407,70 +410,197 @@ pub(crate) trait OperationWithDefaults: Send + Sync {
 
     fn target(&self) -> OperationTarget;
 
+    fn is_after_cluster_time_write(&self) -> bool {
+        self.write_concern().supported()
+    }
+
+    #[cfg(feature = "opentelemetry")]
+    type Otel: crate::otel::OtelWitness<Op = Self>;
+}
+
+pub(crate) trait OperationDispatch<Kind> {
+    type O;
+    const NAME: &'static CStr;
+    const ZERO_COPY: bool;
+    fn build(&mut self, description: &StreamDescription) -> Result<Command>;
+    fn extract_at_cluster_time(&self, response: &RawDocument) -> Result<Option<Timestamp>>;
+    fn handle_response<'a>(
+        &'a self,
+        response: Cow<'a, RawCommandResponse>,
+        context: ExecutionContext<'a>,
+    ) -> BoxFuture<'a, Result<Self::O>>;
+    fn handle_error(&self, error: Error) -> Result<Self::O>;
+    fn selection_criteria(&self) -> Feature<&SelectionCriteria>;
+    fn read_concern(&self) -> Feature<&ReadConcern>;
+    fn write_concern(&self) -> Feature<&WriteConcern>;
+    fn supports_sessions(&self) -> bool;
+    fn retryability(&self, options: &ClientOptions) -> Retryability;
+    fn is_backpressure_retryable(&self, options: &ClientOptions) -> bool;
+    fn update_for_retry(&mut self, retry: Option<&Retry>);
+    fn override_criteria(&self) -> OverrideCriteriaFn;
+    fn pinned_connection(&self) -> Option<&PinnedConnectionHandle>;
+    fn name(&self) -> &CStr;
+    fn target(&self) -> OperationTarget;
+    fn is_after_cluster_time_write(&self) -> bool;
     #[cfg(feature = "opentelemetry")]
     type Otel: crate::otel::OtelWitness<Op = Self>;
 }
 
-impl<T: OperationWithDefaults> Operation for T
+macro_rules! operation_dispatch {
+    ($trait:path, $tag:ty, $handle_response:ident) => {
+        impl<T: $trait> OperationDispatch<$tag> for T {
+            operation_dispatch_body! { $trait, $handle_response }
+        }
+    };
+}
+
+macro_rules! operation_dispatch_body {
+    ($trait:path, $handle_response:ident) => {
+        type O = <T as $trait>::O;
+        const NAME: &'static CStr = <T as $trait>::NAME;
+        const ZERO_COPY: bool = <T as $trait>::ZERO_COPY;
+        fn build(&mut self, description: &StreamDescription) -> Result<Command> {
+            <T as $trait>::build(self, description)
+        }
+        fn extract_at_cluster_time(&self, response: &RawDocument) -> Result<Option<Timestamp>> {
+            <T as $trait>::extract_at_cluster_time(self, response)
+        }
+        fn handle_response<'a>(
+            &'a self,
+            response: Cow<'a, RawCommandResponse>,
+            context: ExecutionContext<'a>,
+        ) -> BoxFuture<'a, Result<Self::O>> {
+            <T as $trait>::$handle_response(self, response, context)
+        }
+        fn handle_error(&self, error: Error) -> Result<Self::O> {
+            <T as $trait>::handle_error(self, error)
+        }
+        fn selection_criteria(&self) -> Feature<&SelectionCriteria> {
+            <T as $trait>::selection_criteria(self)
+        }
+        fn read_concern(&self) -> Feature<&ReadConcern> {
+            <T as $trait>::read_concern(self)
+        }
+        fn write_concern(&self) -> Feature<&WriteConcern> {
+            <T as $trait>::write_concern(self)
+        }
+        fn supports_sessions(&self) -> bool {
+            <T as $trait>::supports_sessions(self)
+        }
+        fn retryability(&self, options: &ClientOptions) -> Retryability {
+            <T as $trait>::retryability(self, options)
+        }
+        fn is_backpressure_retryable(&self, options: &ClientOptions) -> bool {
+            <T as $trait>::is_backpressure_retryable(self, options)
+        }
+        fn update_for_retry(&mut self, retry: Option<&Retry>) {
+            <T as $trait>::update_for_retry(self, retry)
+        }
+        fn override_criteria(&self) -> OverrideCriteriaFn {
+            <T as $trait>::override_criteria(self)
+        }
+        fn pinned_connection(&self) -> Option<&PinnedConnectionHandle> {
+            <T as $trait>::pinned_connection(self)
+        }
+        fn name(&self) -> &CStr {
+            <T as $trait>::name(self)
+        }
+        fn target(&self) -> OperationTarget {
+            <T as $trait>::target(self)
+        }
+        fn is_after_cluster_time_write(&self) -> bool {
+            <T as $trait>::is_after_cluster_time_write(self)
+        }
+        #[cfg(feature = "opentelemetry")]
+        type Otel = <T as $trait>::Otel;
+    };
+}
+
+pub(crate) trait OperationImpl {
+    type Kind;
+}
+
+impl<K, T> Operation for T
 where
-    T: Send + Sync,
+    T: OperationImpl<Kind = K> + OperationDispatch<K> + Send + Sync,
 {
-    type O = T::O;
-    const NAME: &'static CStr = T::NAME;
-    const ZERO_COPY: bool = T::ZERO_COPY;
-    fn build(&mut self, description: &StreamDescription) -> Result<Command> {
-        self.build(description)
-    }
-    fn extract_at_cluster_time(&self, response: &RawDocument) -> Result<Option<Timestamp>> {
-        self.extract_at_cluster_time(response)
-    }
+    operation_dispatch_body! { OperationDispatch<K>, handle_response }
+}
+
+pub(crate) struct WithDefaults;
+
+operation_dispatch! { OperationWithDefaults, WithDefaults, handle_response_async }
+
+pub(crate) trait OperationWrapper {
+    // Required
+    type Wrapped: Operation;
+    type O;
+    #[cfg(feature = "opentelemetry")]
+    type Otel: crate::otel::OtelWitness<Op = Self>;
+
+    fn wrapped(&self) -> &Self::Wrapped;
+    fn wrapped_mut(&mut self) -> &mut Self::Wrapped;
+
     fn handle_response<'a>(
         &'a self,
         response: Cow<'a, RawCommandResponse>,
         context: ExecutionContext<'a>,
-    ) -> BoxFuture<'a, Result<Self::O>> {
-        self.handle_response_async(response, context)
+    ) -> BoxFuture<'a, Result<Self::O>>;
+
+    // Delegated to wrapped
+    const NAME: &'static CStr = Self::Wrapped::NAME;
+    const ZERO_COPY: bool = Self::Wrapped::ZERO_COPY;
+    fn build(&mut self, description: &StreamDescription) -> Result<Command> {
+        self.wrapped_mut().build(description)
     }
     fn handle_error(&self, error: Error) -> Result<Self::O> {
-        self.handle_error(error)
+        Err(error)
+    }
+    fn extract_at_cluster_time(&self, response: &RawDocument) -> Result<Option<Timestamp>> {
+        self.wrapped().extract_at_cluster_time(response)
     }
     fn selection_criteria(&self) -> Feature<&SelectionCriteria> {
-        self.selection_criteria()
+        self.wrapped().selection_criteria()
     }
     fn read_concern(&self) -> Feature<&ReadConcern> {
-        self.read_concern()
+        self.wrapped().read_concern()
     }
     fn write_concern(&self) -> Feature<&WriteConcern> {
-        self.write_concern()
+        self.wrapped().write_concern()
     }
     fn supports_sessions(&self) -> bool {
-        self.supports_sessions()
+        self.wrapped().supports_sessions()
     }
     fn retryability(&self, options: &ClientOptions) -> Retryability {
-        self.retryability(options)
+        self.wrapped().retryability(options)
     }
     fn is_backpressure_retryable(&self, options: &ClientOptions) -> bool {
-        self.is_backpressure_retryable(options)
+        self.wrapped().is_backpressure_retryable(options)
     }
     fn update_for_retry(&mut self, retry: Option<&Retry>) {
-        self.update_for_retry(retry)
+        self.wrapped_mut().update_for_retry(retry);
     }
     fn override_criteria(&self) -> OverrideCriteriaFn {
-        self.override_criteria()
+        self.wrapped().override_criteria()
     }
     fn pinned_connection(&self) -> Option<&PinnedConnectionHandle> {
-        self.pinned_connection()
+        self.wrapped().pinned_connection()
     }
     fn name(&self) -> &CStr {
-        self.name()
+        self.wrapped().name()
     }
     fn target(&self) -> OperationTarget {
-        self.target()
+        self.wrapped().target()
+    }
+    fn is_after_cluster_time_write(&self) -> bool {
+        self.wrapped().is_after_cluster_time_write()
     }
-    #[cfg(feature = "opentelemetry")]
-    type Otel = <Self as OperationWithDefaults>::Otel;
 }
 
+pub(crate) struct Wrapper;
+
+operation_dispatch! { OperationWrapper, Wrapper, handle_response }
+
 fn should_redact_body(body: &RawDocumentBuf) -> bool {
     if let Some(Ok((command_name, _))) = body.into_iter().next() {
         HELLO_COMMAND_NAMES.contains(command_name.to_lowercase().as_str())
diff --git a/driver/src/operation/abort_transaction.rs b/driver/src/operation/abort_transaction.rs
index 8277f7e6f..1a65f771a 100644
--- a/driver/src/operation/abort_transaction.rs
+++ b/driver/src/operation/abort_transaction.rs
@@ -4,7 +4,7 @@ use crate::{
     client::{session::TransactionPin, Retry},
     cmap::{conn::PinnedConnectionHandle, Command, RawCommandResponse, StreamDescription},
     error::Result,
-    operation::Retryability,
+    operation::{OperationImpl, Retryability, WithDefaults},
     options::{ClientOptions, SelectionCriteria, WriteConcern},
     Client,
 };
@@ -85,9 +85,17 @@ impl OperationWithDefaults for AbortTransaction {
         crate::operation::OperationTarget::admin(&self.target)
     }
 
+    fn is_after_cluster_time_write(&self) -> bool {
+        false
+    }
+
     #[cfg(feature = "opentelemetry")]
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for AbortTransaction {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for AbortTransaction {}
diff --git a/driver/src/operation/aggregate.rs b/driver/src/operation/aggregate.rs
index 6c6d0da45..2dd382b8f 100644
--- a/driver/src/operation/aggregate.rs
+++ b/driver/src/operation/aggregate.rs
@@ -7,7 +7,7 @@ use crate::{
     cmap::{Command, RawCommandResponse, StreamDescription},
     cursor::common::CursorSpecification,
     error::Result,
-    operation::{append_options, OperationTarget, Retryability},
+    operation::{append_options, OperationImpl, OperationTarget, Retryability, WithDefaults},
     options::{AggregateOptions, ClientOptions, ReadPreference, SelectionCriteria, WriteConcern},
 };
 
@@ -173,10 +173,18 @@ impl OperationWithDefaults for Aggregate {
         self.target.clone()
     }
 
+    fn is_after_cluster_time_write(&self) -> bool {
+        false
+    }
+
     #[cfg(feature = "opentelemetry")]
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for Aggregate {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for Aggregate {
     fn output_cursor_id(output: &Self::O) -> Option<i64> {
diff --git a/driver/src/operation/aggregate/change_stream.rs b/driver/src/operation/aggregate/change_stream.rs
index 7b66a045a..12ec6a234 100644
--- a/driver/src/operation/aggregate/change_stream.rs
+++ b/driver/src/operation/aggregate/change_stream.rs
@@ -4,12 +4,18 @@ use crate::{
         common::{ChangeStreamData, WatchArgs},
         event::ResumeToken,
     },
-    client::Retry,
     cmap::{Command, RawCommandResponse, StreamDescription},
     cursor::common::CursorSpecification,
     error::Result,
-    operation::{append_options, ExecutionContext, Operation, Retryability},
-    options::{ChangeStreamOptions, ClientOptions, SelectionCriteria, WriteConcern},
+    operation::{
+        append_options,
+        ExecutionContext,
+        Operation,
+        OperationImpl,
+        OperationWrapper,
+        Wrapper,
+    },
+    options::ChangeStreamOptions,
 };
 
 use super::Aggregate;
@@ -43,12 +49,18 @@ impl ChangeStreamAggregate {
     }
 }
 
-impl Operation for ChangeStreamAggregate {
+impl OperationWrapper for ChangeStreamAggregate {
+    type Wrapped = Aggregate;
     type O = (CursorSpecification, ChangeStreamData);
+    const ZERO_COPY: bool = true;
 
-    const NAME: &'static crate::bson_compat::CStr = Aggregate::NAME;
+    fn wrapped(&self) -> &Self::Wrapped {
+        &self.inner
+    }
 
-    const ZERO_COPY: bool = true;
+    fn wrapped_mut(&mut self) -> &mut Self::Wrapped {
+        &mut self.inner
+    }
 
     fn build(&mut self, description: &StreamDescription) -> Result<Command> {
         if let Some(data) = &mut self.resume_data {
@@ -80,13 +92,6 @@ impl Operation for ChangeStreamAggregate {
         self.inner.build(description)
     }
 
-    fn extract_at_cluster_time(
-        &self,
-        response: &crate::bson::RawDocument,
-    ) -> Result<Option<crate::bson::Timestamp>> {
-        self.inner.extract_at_cluster_time(response)
-    }
-
     fn handle_response<'a>(
         &'a self,
         response: std::borrow::Cow<'a, RawCommandResponse>,
@@ -133,57 +138,13 @@ impl Operation for ChangeStreamAggregate {
         .boxed()
     }
 
-    fn selection_criteria(&self) -> crate::operation::Feature<&SelectionCriteria> {
-        self.inner.selection_criteria()
-    }
-
-    fn write_concern(&self) -> crate::operation::Feature<&WriteConcern> {
-        self.inner.write_concern()
-    }
-
-    fn retryability(&self, options: &ClientOptions) -> Retryability {
-        self.inner.retryability(options)
-    }
-
-    fn is_backpressure_retryable(&self, options: &ClientOptions) -> bool {
-        self.inner.is_backpressure_retryable(options)
-    }
-
-    fn target(&self) -> crate::operation::OperationTarget {
-        self.inner.target()
-    }
-
-    fn handle_error(&self, error: crate::error::Error) -> Result<Self::O> {
-        Err(error)
-    }
-
-    fn read_concern(&self) -> crate::operation::Feature<&crate::options::ReadConcern> {
-        self.inner.read_concern()
-    }
-
-    fn supports_sessions(&self) -> bool {
-        self.inner.supports_sessions()
-    }
-
-    fn update_for_retry(&mut self, retry: Option<&Retry>) {
-        self.inner.update_for_retry(retry);
-    }
-
-    fn override_criteria(&self) -> crate::operation::OverrideCriteriaFn {
-        self.inner.override_criteria()
-    }
-
-    fn pinned_connection(&self) -> Option<&crate::cmap::conn::PinnedConnectionHandle> {
-        self.inner.pinned_connection()
-    }
-
-    fn name(&self) -> &crate::bson_compat::CStr {
-        self.inner.name()
-    }
-
     #[cfg(feature = "opentelemetry")]
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for ChangeStreamAggregate {
+    type Kind = Wrapper;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for ChangeStreamAggregate {}
diff --git a/driver/src/operation/bulk_write.rs b/driver/src/operation/bulk_write.rs
index d95a96646..ec8b05114 100644
--- a/driver/src/operation/bulk_write.rs
+++ b/driver/src/operation/bulk_write.rs
@@ -17,7 +17,9 @@ use crate::{
     operation::{
         run_command::RunCommand,
         GetMore,
+        OperationImpl,
         OperationWithDefaults,
+        WithDefaults,
         MAX_ENCRYPTED_WRITE_SIZE,
     },
     options::{BulkWriteOptions, ClientOptions, OperationType, WriteModel},
@@ -528,5 +530,9 @@ where
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl<R: BulkWriteResult> OperationImpl for BulkWrite<'_, R> {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl<R: BulkWriteResult> crate::otel::OtelInfoDefaults for BulkWrite<'_, R> {}
diff --git a/driver/src/operation/commit_transaction.rs b/driver/src/operation/commit_transaction.rs
index 291b7b04e..dcf8d3488 100644
--- a/driver/src/operation/commit_transaction.rs
+++ b/driver/src/operation/commit_transaction.rs
@@ -6,7 +6,13 @@ use crate::{
     client::Retry,
     cmap::{Command, RawCommandResponse, StreamDescription},
     error::Result,
-    operation::{append_options_to_raw_document, OperationWithDefaults, Retryability},
+    operation::{
+        append_options_to_raw_document,
+        OperationImpl,
+        OperationWithDefaults,
+        Retryability,
+        WithDefaults,
+    },
     options::{Acknowledgment, ClientOptions, TransactionOptions, WriteConcern},
     Client,
 };
@@ -92,9 +98,17 @@ impl OperationWithDefaults for CommitTransaction {
         super::OperationTarget::admin(&self.target)
     }
 
+    fn is_after_cluster_time_write(&self) -> bool {
+        false
+    }
+
     #[cfg(feature = "opentelemetry")]
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for CommitTransaction {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for CommitTransaction {}
diff --git a/driver/src/operation/count.rs b/driver/src/operation/count.rs
index 09ef28da2..9481625c1 100644
--- a/driver/src/operation/count.rs
+++ b/driver/src/operation/count.rs
@@ -1,5 +1,6 @@
 use crate::{
     bson::rawdoc,
+    operation::{OperationImpl, WithDefaults},
     options::{ClientOptions, SelectionCriteria},
     Collection,
 };
@@ -88,6 +89,10 @@ impl OperationWithDefaults for Count {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for Count {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for Count {}
 
diff --git a/driver/src/operation/count_documents.rs b/driver/src/operation/count_documents.rs
index 86a8af143..eae9e54d7 100644
--- a/driver/src/operation/count_documents.rs
+++ b/driver/src/operation/count_documents.rs
@@ -3,12 +3,11 @@ use std::convert::TryInto;
 use serde::Deserialize;
 
 use crate::{
-    bson::{doc, Document, RawDocument},
-    client::Retry,
-    cmap::{Command, RawCommandResponse, StreamDescription},
+    bson::{doc, Document},
+    cmap::RawCommandResponse,
     error::{Error, ErrorKind, Result},
-    operation::{aggregate::Aggregate, Operation},
-    options::{AggregateOptions, ClientOptions, CountOptions, SelectionCriteria},
+    operation::{aggregate::Aggregate, OperationImpl, OperationWrapper, Wrapper},
+    options::{AggregateOptions, ClientOptions, CountOptions},
 };
 
 use super::{ExecutionContext, Retryability, SingleCursorResult};
@@ -73,22 +72,17 @@ impl CountDocuments {
     }
 }
 
-impl Operation for CountDocuments {
+impl OperationWrapper for CountDocuments {
+    type Wrapped = Aggregate;
     type O = u64;
-
-    const NAME: &'static crate::bson_compat::CStr = Aggregate::NAME;
-
     const ZERO_COPY: bool = false;
 
-    fn build(&mut self, description: &StreamDescription) -> Result<Command> {
-        self.aggregate.build(description)
+    fn wrapped(&self) -> &Self::Wrapped {
+        &self.aggregate
     }
 
-    fn extract_at_cluster_time(
-        &self,
-        response: &RawDocument,
-    ) -> Result<Option<crate::bson::Timestamp>> {
-        self.aggregate.extract_at_cluster_time(response)
+    fn wrapped_mut(&mut self) -> &mut Self::Wrapped {
+        &mut self.aggregate
     }
 
     fn handle_response<'a>(
@@ -104,10 +98,6 @@ impl Operation for CountDocuments {
         .boxed()
     }
 
-    fn selection_criteria(&self) -> super::Feature<&SelectionCriteria> {
-        self.aggregate.selection_criteria()
-    }
-
     fn retryability(&self, options: &ClientOptions) -> Retryability {
         Retryability::read(options)
     }
@@ -116,46 +106,14 @@ impl Operation for CountDocuments {
         options.retry_reads != Some(false)
     }
 
-    fn target(&self) -> super::OperationTarget {
-        self.aggregate.target()
-    }
-
-    fn read_concern(&self) -> super::Feature<&crate::options::ReadConcern> {
-        self.aggregate.read_concern()
-    }
-
-    fn handle_error(&self, error: Error) -> Result<Self::O> {
-        Err(error)
-    }
-
-    fn write_concern(&self) -> super::Feature<&crate::options::WriteConcern> {
-        self.aggregate.write_concern()
-    }
-
-    fn supports_sessions(&self) -> bool {
-        self.aggregate.supports_sessions()
-    }
-
-    fn update_for_retry(&mut self, retry: Option<&Retry>) {
-        self.aggregate.update_for_retry(retry);
-    }
-
-    fn override_criteria(&self) -> super::OverrideCriteriaFn {
-        self.aggregate.override_criteria()
-    }
-
-    fn pinned_connection(&self) -> Option<&crate::cmap::conn::PinnedConnectionHandle> {
-        self.aggregate.pinned_connection()
-    }
-
-    fn name(&self) -> &crate::bson_compat::CStr {
-        self.aggregate.name()
-    }
-
     #[cfg(feature = "opentelemetry")]
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for CountDocuments {
+    type Kind = Wrapper;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for CountDocuments {}
 
diff --git a/driver/src/operation/create.rs b/driver/src/operation/create.rs
index dc45bdda0..3f6cd6d09 100644
--- a/driver/src/operation/create.rs
+++ b/driver/src/operation/create.rs
@@ -5,7 +5,13 @@ use crate::{
     bson_compat::{cstr, CStr},
     cmap::{Command, RawCommandResponse, StreamDescription},
     error::Result,
-    operation::{append_options_to_raw_document, OperationWithDefaults, WriteConcernOnlyBody},
+    operation::{
+        append_options_to_raw_document,
+        OperationImpl,
+        OperationWithDefaults,
+        WithDefaults,
+        WriteConcernOnlyBody,
+    },
     options::{CreateCollectionOptions, WriteConcern},
 };
 
@@ -65,6 +71,10 @@ impl OperationWithDefaults for Create {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for Create {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for Create {
     fn log_name(&self) -> &str {
diff --git a/driver/src/operation/create_indexes.rs b/driver/src/operation/create_indexes.rs
index 5a97c2cd3..6b1980964 100644
--- a/driver/src/operation/create_indexes.rs
+++ b/driver/src/operation/create_indexes.rs
@@ -7,7 +7,12 @@ use crate::{
     cmap::{Command, RawCommandResponse, StreamDescription},
     error::{ErrorKind, Result},
     index::IndexModel,
-    operation::{append_options_to_raw_document, OperationWithDefaults},
+    operation::{
+        append_options_to_raw_document,
+        OperationImpl,
+        OperationWithDefaults,
+        WithDefaults,
+    },
     options::{CreateIndexOptions, WriteConcern},
     results::CreateIndexesResult,
 };
@@ -97,5 +102,9 @@ impl OperationWithDefaults for CreateIndexes {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for CreateIndexes {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for CreateIndexes {}
diff --git a/driver/src/operation/delete.rs b/driver/src/operation/delete.rs
index ee289ad21..bbef5eb1a 100644
--- a/driver/src/operation/delete.rs
+++ b/driver/src/operation/delete.rs
@@ -4,7 +4,14 @@ use crate::{
     cmap::{Command, RawCommandResponse, StreamDescription},
     collation::Collation,
     error::{convert_insert_many_error, Result},
-    operation::{append_options, OperationWithDefaults, Retryability, WriteResponseBody},
+    operation::{
+        append_options,
+        OperationImpl,
+        OperationWithDefaults,
+        Retryability,
+        WithDefaults,
+        WriteResponseBody,
+    },
     options::{ClientOptions, DeleteOptions, Hint, WriteConcern},
     results::DeleteResult,
     Collection,
@@ -113,5 +120,9 @@ impl OperationWithDefaults for Delete {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for Delete {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for Delete {}
diff --git a/driver/src/operation/distinct.rs b/driver/src/operation/distinct.rs
index 8ea17420f..bf65aae9f 100644
--- a/driver/src/operation/distinct.rs
+++ b/driver/src/operation/distinct.rs
@@ -6,7 +6,7 @@ use crate::{
     cmap::{Command, RawCommandResponse, StreamDescription},
     coll::options::DistinctOptions,
     error::Result,
-    operation::{OperationWithDefaults, Retryability},
+    operation::{OperationImpl, OperationWithDefaults, Retryability, WithDefaults},
     options::{ClientOptions, SelectionCriteria},
     Collection,
 };
@@ -97,6 +97,10 @@ impl OperationWithDefaults for Distinct {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for Distinct {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for Distinct {}
 
diff --git a/driver/src/operation/drop_collection.rs b/driver/src/operation/drop_collection.rs
index f3194a5a2..b5e26afeb 100644
--- a/driver/src/operation/drop_collection.rs
+++ b/driver/src/operation/drop_collection.rs
@@ -5,7 +5,13 @@ use crate::{
     bson_compat::{cstr, CStr},
     cmap::{Command, RawCommandResponse, StreamDescription},
     error::{Error, Result},
-    operation::{append_options_to_raw_document, OperationWithDefaults, WriteConcernOnlyBody},
+    operation::{
+        append_options_to_raw_document,
+        OperationImpl,
+        OperationWithDefaults,
+        WithDefaults,
+        WriteConcernOnlyBody,
+    },
     options::{DropCollectionOptions, WriteConcern},
 };
 
@@ -73,6 +79,10 @@ impl OperationWithDefaults for DropCollection {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for DropCollection {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for DropCollection {
     fn log_name(&self) -> &str {
diff --git a/driver/src/operation/drop_database.rs b/driver/src/operation/drop_database.rs
index d455a8d96..1fc26594e 100644
--- a/driver/src/operation/drop_database.rs
+++ b/driver/src/operation/drop_database.rs
@@ -5,7 +5,13 @@ use crate::{
     cmap::{Command, RawCommandResponse, StreamDescription},
     db::options::DropDatabaseOptions,
     error::Result,
-    operation::{append_options_to_raw_document, OperationWithDefaults, WriteConcernOnlyBody},
+    operation::{
+        append_options_to_raw_document,
+        OperationImpl,
+        OperationWithDefaults,
+        WithDefaults,
+        WriteConcernOnlyBody,
+    },
     options::WriteConcern,
 };
 
@@ -62,5 +68,9 @@ impl OperationWithDefaults for DropDatabase {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for DropDatabase {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for DropDatabase {}
diff --git a/driver/src/operation/drop_indexes.rs b/driver/src/operation/drop_indexes.rs
index 1e7b0e0a7..239130be5 100644
--- a/driver/src/operation/drop_indexes.rs
+++ b/driver/src/operation/drop_indexes.rs
@@ -5,7 +5,12 @@ use crate::{
     bson_compat::{cstr, CStr},
     cmap::{Command, RawCommandResponse, StreamDescription},
     error::Result,
-    operation::{append_options_to_raw_document, OperationWithDefaults},
+    operation::{
+        append_options_to_raw_document,
+        OperationImpl,
+        OperationWithDefaults,
+        WithDefaults,
+    },
     options::{DropIndexOptions, WriteConcern},
 };
 
@@ -73,5 +78,9 @@ impl OperationWithDefaults for DropIndexes {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for DropIndexes {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for DropIndexes {}
diff --git a/driver/src/operation/find.rs b/driver/src/operation/find.rs
index 3d38e6216..f99453006 100644
--- a/driver/src/operation/find.rs
+++ b/driver/src/operation/find.rs
@@ -4,7 +4,13 @@ use crate::{
     cmap::{Command, RawCommandResponse, StreamDescription},
     cursor::common::CursorSpecification,
     error::{Error, Result},
-    operation::{OperationWithDefaults, Retryability, SERVER_4_4_0_WIRE_VERSION},
+    operation::{
+        OperationImpl,
+        OperationWithDefaults,
+        Retryability,
+        WithDefaults,
+        SERVER_4_4_0_WIRE_VERSION,
+    },
     options::{ClientOptions, CursorType, FindOptions, SelectionCriteria},
     Collection,
 };
@@ -137,6 +143,10 @@ impl OperationWithDefaults for Find {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for Find {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for Find {
     fn output_cursor_id(output: &Self::O) -> Option<i64> {
diff --git a/driver/src/operation/find_and_modify.rs b/driver/src/operation/find_and_modify.rs
index 93c1e5562..ab4639a9c 100644
--- a/driver/src/operation/find_and_modify.rs
+++ b/driver/src/operation/find_and_modify.rs
@@ -15,8 +15,10 @@ use crate::{
     operation::{
         append_options_to_raw_document,
         find_and_modify::options::Modification,
+        OperationImpl,
         OperationWithDefaults,
         Retryability,
+        WithDefaults,
     },
     options::{ClientOptions, WriteConcern},
     Collection,
@@ -131,5 +133,9 @@ impl<T: DeserializeOwned> OperationWithDefaults for FindAndModify<T> {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl<T: DeserializeOwned> OperationImpl for FindAndModify<T> {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl<T: DeserializeOwned> crate::otel::OtelInfoDefaults for FindAndModify<T> {}
diff --git a/driver/src/operation/get_more.rs b/driver/src/operation/get_more.rs
index f06518572..3557e71cf 100644
--- a/driver/src/operation/get_more.rs
+++ b/driver/src/operation/get_more.rs
@@ -7,7 +7,7 @@ use crate::{
     cmap::{conn::PinnedConnectionHandle, Command, RawCommandResponse, StreamDescription},
     cursor::common::{CursorInformation, CursorReply},
     error::Result,
-    operation::OperationWithDefaults,
+    operation::{OperationImpl, OperationWithDefaults, WithDefaults},
     options::SelectionCriteria,
     results::GetMoreResult,
     Namespace,
@@ -124,6 +124,10 @@ impl OperationWithDefaults for GetMore<'_> {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for GetMore<'_> {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for GetMore<'_> {
     #[cfg(feature = "opentelemetry")]
diff --git a/driver/src/operation/insert.rs b/driver/src/operation/insert.rs
index 13b639167..04dbb593a 100644
--- a/driver/src/operation/insert.rs
+++ b/driver/src/operation/insert.rs
@@ -12,7 +12,13 @@ use crate::{
     checked::Checked,
     cmap::{Command, RawCommandResponse, StreamDescription},
     error::{Error, ErrorKind, InsertManyError, Result},
-    operation::{OperationWithDefaults, Retryability, WriteResponseBody},
+    operation::{
+        OperationImpl,
+        OperationWithDefaults,
+        Retryability,
+        WithDefaults,
+        WriteResponseBody,
+    },
     options::{ClientOptions, InsertManyOptions, WriteConcern},
     results::InsertManyResult,
     Collection,
@@ -186,5 +192,9 @@ impl OperationWithDefaults for Insert<'_> {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for Insert<'_> {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for Insert<'_> {}
diff --git a/driver/src/operation/list_collections.rs b/driver/src/operation/list_collections.rs
index dd7651317..4eeb0ba03 100644
--- a/driver/src/operation/list_collections.rs
+++ b/driver/src/operation/list_collections.rs
@@ -4,7 +4,7 @@ use crate::{
     cmap::{Command, RawCommandResponse, StreamDescription},
     cursor::common::CursorSpecification,
     error::Result,
-    operation::{OperationWithDefaults, Retryability},
+    operation::{OperationImpl, OperationWithDefaults, Retryability, WithDefaults},
     options::{ClientOptions, ListCollectionsOptions, ReadPreference, SelectionCriteria},
     Database,
 };
@@ -91,6 +91,10 @@ impl OperationWithDefaults for ListCollections {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for ListCollections {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for ListCollections {
     fn output_cursor_id(output: &Self::O) -> Option<i64> {
diff --git a/driver/src/operation/list_databases.rs b/driver/src/operation/list_databases.rs
index f7271372f..c9e504c88 100644
--- a/driver/src/operation/list_databases.rs
+++ b/driver/src/operation/list_databases.rs
@@ -7,7 +7,7 @@ use crate::{
     cmap::{Command, RawCommandResponse, StreamDescription},
     db::options::ListDatabasesOptions,
     error::Result,
-    operation::{OperationWithDefaults, Retryability},
+    operation::{OperationImpl, OperationWithDefaults, Retryability, WithDefaults},
     selection_criteria::{ReadPreference, SelectionCriteria},
 };
 
@@ -71,6 +71,10 @@ impl OperationWithDefaults for ListDatabases {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for ListDatabases {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for ListDatabases {}
 
diff --git a/driver/src/operation/list_indexes.rs b/driver/src/operation/list_indexes.rs
index e8ba28ab2..60d24d742 100644
--- a/driver/src/operation/list_indexes.rs
+++ b/driver/src/operation/list_indexes.rs
@@ -5,7 +5,7 @@ use crate::{
     cmap::{Command, RawCommandResponse, StreamDescription},
     cursor::common::CursorSpecification,
     error::Result,
-    operation::OperationWithDefaults,
+    operation::{OperationImpl, OperationWithDefaults, WithDefaults},
     options::{ClientOptions, ListIndexesOptions},
     selection_criteria::{ReadPreference, SelectionCriteria},
     Collection,
@@ -78,6 +78,10 @@ impl OperationWithDefaults for ListIndexes {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for ListIndexes {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for ListIndexes {
     fn output_cursor_id(output: &Self::O) -> Option<i64> {
diff --git a/driver/src/operation/raw_output.rs b/driver/src/operation/raw_output.rs
index 78e19c152..f3526b6df 100644
--- a/driver/src/operation/raw_output.rs
+++ b/driver/src/operation/raw_output.rs
@@ -1,11 +1,9 @@
 use futures_util::FutureExt;
 
 use crate::{
-    bson_compat::CStr,
-    client::Retry,
-    cmap::{Command, RawCommandResponse, StreamDescription},
+    cmap::RawCommandResponse,
     error::Result,
-    options::{ClientOptions, SelectionCriteria},
+    operation::{OperationImpl, OperationWrapper, Wrapper},
     BoxFuture,
 };
 
@@ -16,20 +14,17 @@ use super::{ExecutionContext, Operation};
 #[derive(Clone)]
 pub(crate) struct RawOutput<Op>(pub(crate) Op);
 
-impl<Op: Operation> Operation for RawOutput<Op> {
+impl<Op: Operation + Sync + Send> OperationWrapper for RawOutput<Op> {
+    type Wrapped = Op;
     type O = RawCommandResponse;
-    const NAME: &'static CStr = Op::NAME;
     const ZERO_COPY: bool = true;
 
-    fn build(&mut self, description: &StreamDescription) -> Result<Command> {
-        self.0.build(description)
+    fn wrapped(&self) -> &Self::Wrapped {
+        &self.0
     }
 
-    fn extract_at_cluster_time(
-        &self,
-        response: &crate::bson::RawDocument,
-    ) -> Result<Option<crate::bson::Timestamp>> {
-        self.0.extract_at_cluster_time(response)
+    fn wrapped_mut(&mut self) -> &mut Self::Wrapped {
+        &mut self.0
     }
 
     fn handle_response<'a>(
@@ -40,60 +35,16 @@ impl<Op: Operation> Operation for RawOutput<Op> {
         async move { Ok(response.into_owned()) }.boxed()
     }
 
-    fn handle_error(&self, error: crate::error::Error) -> Result<Self::O> {
-        Err(error)
-    }
-
-    fn selection_criteria(&self) -> super::Feature<&SelectionCriteria> {
-        self.0.selection_criteria()
-    }
-
-    fn read_concern(&self) -> super::Feature<&crate::options::ReadConcern> {
-        self.0.read_concern()
-    }
-
-    fn write_concern(&self) -> super::Feature<&crate::options::WriteConcern> {
-        self.0.write_concern()
-    }
-
-    fn supports_sessions(&self) -> bool {
-        self.0.supports_sessions()
-    }
-
-    fn retryability(&self, options: &ClientOptions) -> super::Retryability {
-        self.0.retryability(options)
-    }
-
-    fn is_backpressure_retryable(&self, options: &ClientOptions) -> bool {
-        self.0.is_backpressure_retryable(options)
-    }
-
-    fn update_for_retry(&mut self, retry: Option<&Retry>) {
-        self.0.update_for_retry(retry)
-    }
-
-    fn override_criteria(&self) -> super::OverrideCriteriaFn {
-        self.0.override_criteria()
-    }
-
-    fn pinned_connection(&self) -> Option<&crate::cmap::conn::PinnedConnectionHandle> {
-        self.0.pinned_connection()
-    }
-
-    fn name(&self) -> &CStr {
-        self.0.name()
-    }
-
-    fn target(&self) -> super::OperationTarget {
-        self.0.target()
-    }
-
     #[cfg(feature = "opentelemetry")]
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl<Op> OperationImpl for RawOutput<Op> {
+    type Kind = Wrapper;
+}
+
 #[cfg(feature = "opentelemetry")]
-impl<Op: Operation> crate::otel::OtelInfo for RawOutput<Op> {
+impl<Op: Operation + Sync + Send> crate::otel::OtelInfo for RawOutput<Op> {
     fn log_name(&self) -> &str {
         self.0.otel().log_name()
     }
diff --git a/driver/src/operation/run_command.rs b/driver/src/operation/run_command.rs
index 956903751..7a2f258f4 100644
--- a/driver/src/operation/run_command.rs
+++ b/driver/src/operation/run_command.rs
@@ -6,6 +6,7 @@ use crate::{
     client::SESSIONS_UNSUPPORTED_COMMANDS,
     cmap::{conn::PinnedConnectionHandle, Command, RawCommandResponse, StreamDescription},
     error::{Error, Result},
+    operation::{OperationImpl, WithDefaults},
     options::ClientOptions,
     selection_criteria::SelectionCriteria,
     Database,
@@ -118,5 +119,9 @@ impl OperationWithDefaults for RunCommand<'_> {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for RunCommand<'_> {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for RunCommand<'_> {}
diff --git a/driver/src/operation/run_cursor_command.rs b/driver/src/operation/run_cursor_command.rs
index ffa19d648..7415a1927 100644
--- a/driver/src/operation/run_cursor_command.rs
+++ b/driver/src/operation/run_cursor_command.rs
@@ -2,13 +2,18 @@ use futures_util::FutureExt;
 
 use crate::{
     bson_compat::{cstr, CStr},
-    client::Retry,
-    cmap::{conn::PinnedConnectionHandle, Command, RawCommandResponse, StreamDescription},
-    concern::WriteConcern,
+    cmap::RawCommandResponse,
     cursor::common::CursorSpecification,
-    error::{Error, Result},
-    operation::{run_command::RunCommand, Operation, SERVER_4_4_0_WIRE_VERSION},
-    options::{ClientOptions, RunCursorCommandOptions, SelectionCriteria},
+    error::Result,
+    operation::{
+        run_command::RunCommand,
+        Operation,
+        OperationImpl,
+        OperationWrapper,
+        Wrapper,
+        SERVER_4_4_0_WIRE_VERSION,
+    },
+    options::RunCursorCommandOptions,
     BoxFuture,
 };
 
@@ -32,70 +37,18 @@ impl<'conn> RunCursorCommand<'conn> {
     }
 }
 
-impl Operation for RunCursorCommand<'_> {
+impl<'conn> OperationWrapper for RunCursorCommand<'conn> {
+    type Wrapped = RunCommand<'conn>;
     type O = CursorSpecification;
-
     const NAME: &'static CStr = cstr!("run_cursor_command");
-
     const ZERO_COPY: bool = true;
 
-    fn build(&mut self, description: &StreamDescription) -> Result<Command> {
-        self.run_command.build(description)
-    }
-
-    fn extract_at_cluster_time(
-        &self,
-        response: &crate::bson::RawDocument,
-    ) -> Result<Option<crate::bson::Timestamp>> {
-        self.run_command.extract_at_cluster_time(response)
-    }
-
-    fn handle_error(&self, error: Error) -> Result<Self::O> {
-        Err(error)
-    }
-
-    fn selection_criteria(&self) -> super::Feature<&SelectionCriteria> {
-        self.run_command.selection_criteria()
-    }
-
-    fn read_concern(&self) -> super::Feature<&crate::options::ReadConcern> {
-        self.run_command.read_concern()
-    }
-
-    fn write_concern(&self) -> super::Feature<&WriteConcern> {
-        self.run_command.write_concern()
-    }
-
-    fn supports_sessions(&self) -> bool {
-        self.run_command.supports_sessions()
+    fn wrapped(&self) -> &Self::Wrapped {
+        &self.run_command
     }
 
-    fn retryability(&self, options: &ClientOptions) -> crate::operation::Retryability {
-        self.run_command.retryability(options)
-    }
-
-    fn is_backpressure_retryable(&self, options: &ClientOptions) -> bool {
-        self.run_command.is_backpressure_retryable(options)
-    }
-
-    fn update_for_retry(&mut self, retry: Option<&Retry>) {
-        self.run_command.update_for_retry(retry)
-    }
-
-    fn override_criteria(&self) -> super::OverrideCriteriaFn {
-        self.run_command.override_criteria()
-    }
-
-    fn pinned_connection(&self) -> Option<&PinnedConnectionHandle> {
-        self.run_command.pinned_connection()
-    }
-
-    fn name(&self) -> &CStr {
-        self.run_command.name()
-    }
-
-    fn target(&self) -> super::OperationTarget {
-        self.run_command.target()
+    fn wrapped_mut(&mut self) -> &mut Self::Wrapped {
+        &mut self.run_command
     }
 
     fn handle_response<'a>(
@@ -128,6 +81,10 @@ impl Operation for RunCursorCommand<'_> {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for RunCursorCommand<'_> {
+    type Kind = Wrapper;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfo for RunCursorCommand<'_> {
     fn log_name(&self) -> &str {
diff --git a/driver/src/operation/search_index.rs b/driver/src/operation/search_index.rs
index df3d73069..6067378ab 100644
--- a/driver/src/operation/search_index.rs
+++ b/driver/src/operation/search_index.rs
@@ -10,7 +10,7 @@ use crate::{
     SearchIndexModel,
 };
 
-use super::{ExecutionContext, OperationWithDefaults};
+use super::{ExecutionContext, OperationImpl, OperationWithDefaults, WithDefaults};
 
 #[derive(Debug)]
 pub(crate) struct CreateSearchIndexes {
@@ -76,6 +76,10 @@ impl OperationWithDefaults for CreateSearchIndexes {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for CreateSearchIndexes {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for CreateSearchIndexes {}
 
@@ -135,6 +139,10 @@ impl OperationWithDefaults for UpdateSearchIndex {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for UpdateSearchIndex {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for UpdateSearchIndex {}
 
@@ -192,5 +200,9 @@ impl OperationWithDefaults for DropSearchIndex {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for DropSearchIndex {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for DropSearchIndex {}
diff --git a/driver/src/operation/update.rs b/driver/src/operation/update.rs
index 97c213d84..e57adbb94 100644
--- a/driver/src/operation/update.rs
+++ b/driver/src/operation/update.rs
@@ -8,7 +8,13 @@ use crate::{
     bson_util,
     cmap::{Command, RawCommandResponse, StreamDescription},
     error::{convert_insert_many_error, Result},
-    operation::{OperationWithDefaults, Retryability, WriteResponseBody},
+    operation::{
+        OperationImpl,
+        OperationWithDefaults,
+        Retryability,
+        WithDefaults,
+        WriteResponseBody,
+    },
     options::{ClientOptions, UpdateModifications, UpdateOptions, WriteConcern},
     results::UpdateResult,
     Collection,
@@ -217,6 +223,10 @@ impl OperationWithDefaults for Update {
     type Otel = crate::otel::Witness<Self>;
 }
 
+impl OperationImpl for Update {
+    type Kind = WithDefaults;
+}
+
 #[cfg(feature = "opentelemetry")]
 impl crate::otel::OtelInfoDefaults for Update {}
 
diff --git a/driver/src/test/spec.rs b/driver/src/test/spec.rs
index d944a8ba9..499886122 100644
--- a/driver/src/test/spec.rs
+++ b/driver/src/test/spec.rs
@@ -1,4 +1,5 @@
 mod auth;
+mod causal_consistency;
 mod change_streams;
 mod collection_management;
 mod command_monitoring;
diff --git a/driver/src/test/spec/causal_consistency.rs b/driver/src/test/spec/causal_consistency.rs
new file mode 100644
index 000000000..4bd489c76
--- /dev/null
+++ b/driver/src/test/spec/causal_consistency.rs
@@ -0,0 +1,6 @@
+use crate::test::spec::unified_runner::run_unified_tests;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn run_unified() {
+    run_unified_tests(&["causal-consistency"]).await;
+}
diff --git a/etc/update-spec-tests.sh b/etc/update-spec-tests.sh
index 2d54eda55..b2396f059 100755
--- a/etc/update-spec-tests.sh
+++ b/etc/update-spec-tests.sh
@@ -32,7 +32,7 @@ EXCLUDE=""
 if [ "$1" = "client-side-encryption" ]; then
   EXCLUDE="--exclude=legacy/"
 fi
-rsync -ah "$tmpdir/specifications-$REF"*"/source/$1/tests/" "${DEST}/$1" --delete "${EXCLUDE}"
+rsync -ah $EXCLUDE --delete "$tmpdir/specifications-$REF"*"/source/$1/tests/" "${DEST}/$1"
 
 if [ "$1" = "client-side-encryption" ]; then
     mkdir -p "${DEST}/testdata/$1/data"
diff --git a/spec/causal-consistency/causal-consistency-clientBulkWrite.json b/spec/causal-consistency/causal-consistency-clientBulkWrite.json
new file mode 100644
index 000000000..c2a04422d
--- /dev/null
+++ b/spec/causal-consistency/causal-consistency-clientBulkWrite.json
@@ -0,0 +1,151 @@
+{
+  "description": "causal consistency bulkWrite include afterClusterTime",
+  "schemaVersion": "1.3",
+  "runOnRequirements": [
+    {
+      "minServerVersion": "8.0",
+      "topologies": [
+        "replicaset",
+        "sharded",
+        "load-balanced"
+      ]
+    }
+  ],
+  "createEntities": [
+    {
+      "client": {
+        "id": "client0",
+        "useMultipleMongoses": false,
+        "uriOptions": {
+          "retryWrites": false
+        },
+        "observeEvents": [
+          "commandStartedEvent"
+        ]
+      }
+    },
+    {
+      "database": {
+        "id": "database0",
+        "client": "client0",
+        "databaseName": "causal-consistency-tests"
+      }
+    },
+    {
+      "collection": {
+        "id": "collection0",
+        "database": "database0",
+        "collectionName": "test"
+      }
+    },
+    {
+      "session": {
+        "id": "session0",
+        "client": "client0",
+        "sessionOptions": {
+          "causalConsistency": true
+        }
+      }
+    }
+  ],
+  "initialData": [
+    {
+      "collectionName": "test",
+      "databaseName": "causal-consistency-tests",
+      "documents": [
+        {
+          "_id": 1,
+          "x": 11
+        },
+        {
+          "_id": 2,
+          "x": 22
+        },
+        {
+          "_id": 3,
+          "x": 33
+        }
+      ]
+    }
+  ],
+  "tests": [
+    {
+      "description": "clientBulkWrite includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "clientBulkWrite",
+          "object": "client0",
+          "arguments": {
+            "session": "session0",
+            "models": [
+              {
+                "insertOne": {
+                  "namespace": "causal-consistency-tests.test",
+                  "document": {
+                    "_id": 4
+                  }
+                }
+              }
+            ]
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "bulkWrite",
+                "command": {
+                  "bulkWrite": 1,
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}
diff --git a/spec/causal-consistency/causal-consistency-clientBulkWrite.yml b/spec/causal-consistency/causal-consistency-clientBulkWrite.yml
new file mode 100644
index 000000000..d1902e188
--- /dev/null
+++ b/spec/causal-consistency/causal-consistency-clientBulkWrite.yml
@@ -0,0 +1,75 @@
+description: "causal consistency bulkWrite include afterClusterTime"
+
+schemaVersion: "1.3"
+
+runOnRequirements:
+  - minServerVersion: "8.0"
+    topologies: [replicaset, sharded, load-balanced]
+
+createEntities:
+  - client:
+      id: &client0 client0
+      useMultipleMongoses: false
+      uriOptions:
+        retryWrites: false
+      observeEvents: [commandStartedEvent]
+  - database:
+      id: &database0 database0
+      client: *client0
+      databaseName: &databaseName causal-consistency-tests
+  - collection:
+      id: &collection0 collection0
+      database: *database0
+      collectionName: &collectionName test
+  - session:
+      id: &session0 session0
+      client: *client0
+      sessionOptions:
+        causalConsistency: true
+
+initialData:
+  - collectionName: *collectionName
+    databaseName: *databaseName
+    documents:
+      - { _id: 1, x: 11 }
+      - { _id: 2, x: 22 }
+      - { _id: 3, x: 33 }
+
+# In a causally consistent session, once an operationTime has been established by a prior
+# operation, subsequent write commands MUST include readConcern.afterClusterTime so the
+# server can apply the write causally after the previously-observed data.
+
+tests:
+  - description: "clientBulkWrite includes afterClusterTime in causally consistent session"
+    operations:
+      - name: find
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: 1 }
+        expectResult: [{ _id: 1, x: 11 }]
+      - name: clientBulkWrite
+        object: *client0
+        arguments:
+          session: *session0
+          models:
+          - insertOne:
+              namespace: causal-consistency-tests.test
+              document: { _id: 4 }
+    expectEvents:
+      - client: *client0
+        events:
+          - commandStartedEvent:
+              commandName: find
+              command:
+                find: *collectionName
+                readConcern: { $$exists: false }
+                lsid: { $$sessionLsid: *session0 }
+          - commandStartedEvent:
+              commandName: bulkWrite
+              command:
+                bulkWrite: 1
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
diff --git a/spec/causal-consistency/causal-consistency-write-commands.json b/spec/causal-consistency/causal-consistency-write-commands.json
new file mode 100644
index 000000000..a8140c2b3
--- /dev/null
+++ b/spec/causal-consistency/causal-consistency-write-commands.json
@@ -0,0 +1,1396 @@
+{
+  "description": "causal consistency write commands include afterClusterTime",
+  "schemaVersion": "1.3",
+  "runOnRequirements": [
+    {
+      "topologies": [
+        "replicaset",
+        "sharded",
+        "load-balanced"
+      ]
+    }
+  ],
+  "createEntities": [
+    {
+      "client": {
+        "id": "client0",
+        "useMultipleMongoses": false,
+        "uriOptions": {
+          "retryWrites": false
+        },
+        "observeEvents": [
+          "commandStartedEvent"
+        ]
+      }
+    },
+    {
+      "database": {
+        "id": "database0",
+        "client": "client0",
+        "databaseName": "causal-consistency-tests"
+      }
+    },
+    {
+      "collection": {
+        "id": "collection0",
+        "database": "database0",
+        "collectionName": "test"
+      }
+    },
+    {
+      "session": {
+        "id": "session0",
+        "client": "client0",
+        "sessionOptions": {
+          "causalConsistency": true
+        }
+      }
+    }
+  ],
+  "initialData": [
+    {
+      "collectionName": "test",
+      "databaseName": "causal-consistency-tests",
+      "documents": [
+        {
+          "_id": 1,
+          "x": 11
+        },
+        {
+          "_id": 2,
+          "x": 22
+        },
+        {
+          "_id": 3,
+          "x": 33
+        }
+      ]
+    }
+  ],
+  "tests": [
+    {
+      "description": "insertOne includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "insertOne",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "document": {
+              "_id": 4
+            }
+          },
+          "expectResult": {
+            "$$unsetOrMatches": {
+              "insertedId": {
+                "$$unsetOrMatches": 4
+              }
+            }
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "insert",
+                "command": {
+                  "insert": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "insertMany includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "insertMany",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "documents": [
+              {
+                "_id": 4
+              },
+              {
+                "_id": 5
+              }
+            ]
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "insert",
+                "command": {
+                  "insert": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "updateOne includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "updateOne",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            },
+            "update": {
+              "$set": {
+                "x": 100
+              }
+            }
+          },
+          "expectResult": {
+            "matchedCount": 1,
+            "modifiedCount": 1,
+            "upsertedCount": 0
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "update",
+                "command": {
+                  "update": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "updateMany includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "updateMany",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": {
+                "$gt": 0
+              }
+            },
+            "update": {
+              "$set": {
+                "updated": true
+              }
+            }
+          },
+          "expectResult": {
+            "matchedCount": 3,
+            "modifiedCount": 3,
+            "upsertedCount": 0
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "update",
+                "command": {
+                  "update": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "replaceOne includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "replaceOne",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            },
+            "replacement": {
+              "x": 100
+            }
+          },
+          "expectResult": {
+            "matchedCount": 1,
+            "modifiedCount": 1,
+            "upsertedCount": 0
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "update",
+                "command": {
+                  "update": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "deleteOne includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "deleteOne",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": {
+            "deletedCount": 1
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "delete",
+                "command": {
+                  "delete": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "deleteMany includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "deleteMany",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": {
+                "$gt": 0
+              }
+            }
+          },
+          "expectResult": {
+            "deletedCount": 3
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "delete",
+                "command": {
+                  "delete": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "findOneAndUpdate includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "findOneAndUpdate",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            },
+            "update": {
+              "$set": {
+                "x": 100
+              }
+            }
+          },
+          "expectResult": {
+            "_id": 1,
+            "x": 11
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "findAndModify",
+                "command": {
+                  "findAndModify": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "findOneAndDelete includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "findOneAndDelete",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": {
+            "_id": 1,
+            "x": 11
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "findAndModify",
+                "command": {
+                  "findAndModify": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "findOneAndReplace includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "findOneAndReplace",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            },
+            "replacement": {
+              "x": 100
+            }
+          },
+          "expectResult": {
+            "_id": 1,
+            "x": 11
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "findAndModify",
+                "command": {
+                  "findAndModify": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "bulkWrite includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "bulkWrite",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "requests": [
+              {
+                "insertOne": {
+                  "document": {
+                    "_id": 4
+                  }
+                }
+              },
+              {
+                "updateOne": {
+                  "filter": {
+                    "_id": 2
+                  },
+                  "update": {
+                    "$set": {
+                      "x": 100
+                    }
+                  }
+                }
+              },
+              {
+                "deleteOne": {
+                  "filter": {
+                    "_id": 3
+                  }
+                }
+              }
+            ]
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "insert",
+                "command": {
+                  "insert": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "update",
+                "command": {
+                  "update": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "delete",
+                "command": {
+                  "delete": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "create includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "dropCollection",
+          "object": "database0",
+          "arguments": {
+            "session": "session0",
+            "collection": "causal-consistency-createCollection-test"
+          }
+        },
+        {
+          "name": "createCollection",
+          "object": "database0",
+          "arguments": {
+            "session": "session0",
+            "collection": "causal-consistency-createCollection-test"
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "drop",
+                "command": {
+                  "drop": "causal-consistency-createCollection-test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "create",
+                "command": {
+                  "create": "causal-consistency-createCollection-test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "createIndexes includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "createIndex",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "keys": {
+              "x": 1
+            },
+            "name": "x_1"
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "createIndexes",
+                "command": {
+                  "createIndexes": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "drop includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "dropCollection",
+          "object": "database0",
+          "arguments": {
+            "session": "session0",
+            "collection": "test"
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "drop",
+                "command": {
+                  "drop": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "dropDatabase includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "dropDatabase",
+          "object": "client0",
+          "arguments": {
+            "session": "session0",
+            "database": "causal-consistency-dropDatabase-test"
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "dropDatabase",
+                "command": {
+                  "dropDatabase": 1,
+                  "$db": "causal-consistency-dropDatabase-test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "dropIndexes includes afterClusterTime in causally consistent session",
+      "operations": [
+        {
+          "name": "find",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "filter": {
+              "_id": 1
+            }
+          },
+          "expectResult": [
+            {
+              "_id": 1,
+              "x": 11
+            }
+          ]
+        },
+        {
+          "name": "dropIndexes",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0"
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "find",
+                "command": {
+                  "find": "test",
+                  "readConcern": {
+                    "$$exists": false
+                  },
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  }
+                }
+              }
+            },
+            {
+              "commandStartedEvent": {
+                "commandName": "dropIndexes",
+                "command": {
+                  "dropIndexes": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "description": "first write command in a causally consistent session does not include afterClusterTime",
+      "operations": [
+        {
+          "name": "insertOne",
+          "object": "collection0",
+          "arguments": {
+            "session": "session0",
+            "document": {
+              "_id": 4
+            }
+          },
+          "expectResult": {
+            "$$unsetOrMatches": {
+              "insertedId": {
+                "$$unsetOrMatches": 4
+              }
+            }
+          }
+        }
+      ],
+      "expectEvents": [
+        {
+          "client": "client0",
+          "events": [
+            {
+              "commandStartedEvent": {
+                "commandName": "insert",
+                "command": {
+                  "insert": "test",
+                  "lsid": {
+                    "$$sessionLsid": "session0"
+                  },
+                  "readConcern": {
+                    "$$exists": false
+                  }
+                }
+              }
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}
diff --git a/spec/causal-consistency/causal-consistency-write-commands.yml b/spec/causal-consistency/causal-consistency-write-commands.yml
new file mode 100644
index 000000000..a33ca1743
--- /dev/null
+++ b/spec/causal-consistency/causal-consistency-write-commands.yml
@@ -0,0 +1,472 @@
+description: "causal consistency write commands include afterClusterTime"
+
+schemaVersion: "1.3"
+
+runOnRequirements:
+  - topologies: [replicaset, sharded, load-balanced]
+
+createEntities:
+  - client:
+      id: &client0 client0
+      useMultipleMongoses: false
+      uriOptions:
+        retryWrites: false
+      observeEvents: [commandStartedEvent]
+  - database:
+      id: &database0 database0
+      client: *client0
+      databaseName: &databaseName causal-consistency-tests
+  - collection:
+      id: &collection0 collection0
+      database: *database0
+      collectionName: &collectionName test
+  - session:
+      id: &session0 session0
+      client: *client0
+      sessionOptions:
+        causalConsistency: true
+
+initialData:
+  - collectionName: *collectionName
+    databaseName: *databaseName
+    documents:
+      - { _id: 1, x: 11 }
+      - { _id: 2, x: 22 }
+      - { _id: 3, x: 33 }
+
+# In a causally consistent session, once an operationTime has been established by a prior
+# operation, subsequent write commands MUST include readConcern.afterClusterTime so the
+# server can apply the write causally after the previously-observed data.
+
+tests:
+  - description: "insertOne includes afterClusterTime in causally consistent session"
+    operations:
+      - &find
+        name: find
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: 1 }
+        expectResult: [{ _id: 1, x: 11 }]
+      - name: insertOne
+        object: *collection0
+        arguments:
+          session: *session0
+          document: { _id: 4 }
+        expectResult:
+          $$unsetOrMatches: { insertedId: { $$unsetOrMatches: 4 } }
+    expectEvents:
+      - client: *client0
+        events:
+          - &findEvent
+            commandStartedEvent:
+              commandName: find
+              command:
+                find: *collectionName
+                readConcern: { $$exists: false }
+                lsid: { $$sessionLsid: *session0 }
+          - commandStartedEvent:
+              commandName: insert
+              command:
+                insert: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "insertMany includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: insertMany
+        object: *collection0
+        arguments:
+          session: *session0
+          documents:
+            - { _id: 4 }
+            - { _id: 5 }
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: insert
+              command:
+                insert: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "updateOne includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: updateOne
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: 1 }
+          update: { $set: { x: 100 } }
+        expectResult:
+          matchedCount: 1
+          modifiedCount: 1
+          upsertedCount: 0
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: update
+              command:
+                update: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "updateMany includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: updateMany
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: { $gt: 0 } }
+          update: { $set: { updated: true } }
+        expectResult:
+          matchedCount: 3
+          modifiedCount: 3
+          upsertedCount: 0
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: update
+              command:
+                update: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "replaceOne includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: replaceOne
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: 1 }
+          replacement: { x: 100 }
+        expectResult:
+          matchedCount: 1
+          modifiedCount: 1
+          upsertedCount: 0
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: update
+              command:
+                update: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "deleteOne includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: deleteOne
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: 1 }
+        expectResult:
+          deletedCount: 1
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: delete
+              command:
+                delete: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "deleteMany includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: deleteMany
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: { $gt: 0 } }
+        expectResult:
+          deletedCount: 3
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: delete
+              command:
+                delete: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "findOneAndUpdate includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: findOneAndUpdate
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: 1 }
+          update: { $set: { x: 100 } }
+        expectResult: { _id: 1, x: 11 }
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: findAndModify
+              command:
+                findAndModify: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "findOneAndDelete includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: findOneAndDelete
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: 1 }
+        expectResult: { _id: 1, x: 11 }
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: findAndModify
+              command:
+                findAndModify: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "findOneAndReplace includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: findOneAndReplace
+        object: *collection0
+        arguments:
+          session: *session0
+          filter: { _id: 1 }
+          replacement: { x: 100 }
+        expectResult: { _id: 1, x: 11 }
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: findAndModify
+              command:
+                findAndModify: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "bulkWrite includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: bulkWrite
+        object: *collection0
+        arguments:
+          session: *session0
+          requests:
+            - insertOne:
+                document: { _id: 4 }
+            - updateOne:
+                filter: { _id: 2 }
+                update: { $set: { x: 100 } }
+            - deleteOne:
+                filter: { _id: 3 }
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: insert
+              command:
+                insert: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+          - commandStartedEvent:
+              commandName: update
+              command:
+                update: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+          - commandStartedEvent:
+              commandName: delete
+              command:
+                delete: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "create includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      # Drop the collection first to make sure there's no name conflict during
+      # createCollection.
+      - name: dropCollection
+        object: *database0
+        arguments:
+          session: *session0
+          collection: &newCollectionName causal-consistency-createCollection-test
+      - name: createCollection
+        object: *database0
+        arguments:
+          session: *session0
+          collection: *newCollectionName
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: drop
+              command:
+                drop: *newCollectionName
+                lsid: { $$sessionLsid: *session0 }
+          - commandStartedEvent:
+              commandName: create
+              command:
+                create: *newCollectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "createIndexes includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: createIndex
+        object: *collection0
+        arguments:
+          session: *session0
+          keys: { x: 1 }
+          name: x_1
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: createIndexes
+              command:
+                createIndexes: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "drop includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: dropCollection
+        object: *database0
+        arguments:
+          session: *session0
+          collection: *collectionName
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: drop
+              command:
+                drop: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "dropDatabase includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: dropDatabase
+        object: *client0
+        arguments:
+          session: *session0
+          database: causal-consistency-dropDatabase-test
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: dropDatabase
+              command:
+                dropDatabase: 1
+                $db: causal-consistency-dropDatabase-test
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  - description: "dropIndexes includes afterClusterTime in causally consistent session"
+    operations:
+      - *find
+      - name: dropIndexes
+        object: *collection0
+        arguments:
+          session: *session0
+    expectEvents:
+      - client: *client0
+        events:
+          - *findEvent
+          - commandStartedEvent:
+              commandName: dropIndexes
+              command:
+                dropIndexes: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
+
+  # Covers the same condition as Causal Consistency prose test #2, but for a write operation.
+  - description: "first write command in a causally consistent session does not include afterClusterTime"
+    operations:
+      - name: insertOne
+        object: *collection0
+        arguments:
+          session: *session0
+          document: { _id: 4 }
+        expectResult:
+          $$unsetOrMatches: { insertedId: { $$unsetOrMatches: 4 } }
+    expectEvents:
+      - client: *client0
+        events:
+          - commandStartedEvent:
+              commandName: insert
+              command:
+                insert: *collectionName
+                lsid: { $$sessionLsid: *session0 }
+                readConcern: { $$exists: false }
diff --git a/spec/transactions-convenient-api/README.md b/spec/transactions-convenient-api/README.md
index a797a3182..d8ec2b479 100644
--- a/spec/transactions-convenient-api/README.md
+++ b/spec/transactions-convenient-api/README.md
@@ -29,20 +29,97 @@ Write a callback that returns a custom value (e.g. boolean, string, object). Exe
 Drivers should test that `withTransaction` enforces a non-configurable timeout before retrying both commits and entire
 transactions. Specifically, three cases should be checked:
 
-- If the callback raises an error with the TransientTransactionError label and the retry timeout has been exceeded,
-    `withTransaction` should propagate the error to its caller.
-- If committing raises an error with the UnknownTransactionCommitResult label, and the retry timeout has been exceeded,
-    `withTransaction` should propagate the error to its caller.
-- If committing raises an error with the TransientTransactionError label and the retry timeout has been exceeded,
-    `withTransaction` should propagate the error to its caller. This case may occur if the commit was internally retried
-    against a new primary after a failover and the second primary returned a NoSuchTransaction error response.
+- If the callback raises an error with the `TransientTransactionError` label and the retry timeout has been exceeded,
+    `withTransaction` should propagate the error as described in the
+    [propagation mechanism](../transactions-convenient-api.md#timeout-error-propagation) to its caller.
+- If committing raises an error with the `UnknownTransactionCommitResult` label, and the retry timeout has been
+    exceeded, `withTransaction` should propagate the error as described in the
+    [propagation mechanism](../transactions-convenient-api.md#timeout-error-propagation) to its caller
+- If committing raises an error with the `TransientTransactionError` label and the retry timeout has been exceeded,
+    `withTransaction` should propagate the error as described in the
+    [propagation mechanism](../transactions-convenient-api.md#timeout-error-propagation) to its caller. This case may
+    occur if the commit was internally retried against a new primary after a failover and the second primary returned a
+    `NoSuchTransaction` error response.
 
 If possible, drivers should implement these tests without requiring the test runner to block for the full duration of
 the retry timeout. This might be done by internally modifying the timeout value used by `withTransaction` with some
 private API or using a mock timer.
 
+The drivers should assert that the timeout error propagated has the same labels as the error it wraps.
+
+### Retry Backoff is Enforced
+
+Drivers should test that retries within `withTransaction` do not occur immediately.
+
+1. let `client` be a `MongoClient`
+2. let `coll` be a collection
+3. Now, run transactions without backoff:
+    1. Configure the random number generator used for jitter to always return `0` -- this effectively disables backoff.
+
+    2. Configure a fail point that forces 13 retries like so:
+
+        ```python
+           set_fail_point(
+              {
+                  "configureFailPoint": "failCommand",
+                  "mode": {
+                      "times": 13
+                  },  # sufficiently high enough such that the time effect of backoff is noticeable
+                  "data": {
+                      "failCommands": ["commitTransaction"],
+                      "errorCode": 251,
+                  },
+              }
+           )
+        ```
+
+        > Note: errorCode 251 is NoSuchTransaction.
+
+    3. Define the callback for the transaction as follows:
+
+        ```python
+           def callback(session):
+              coll.insert_one({}, session=session)
+        ```
+
+    4. Let `no_backoff_time` be the duration of the withTransaction API call:
+
+        ```python
+           start = time.monotonic()
+           with client.start_session() as s:
+              s.with_transaction(callback)
+           end = time.monotonic()
+           no_backoff_time = end - start
+        ```
+4. Now run the command with backoff:
+    1. Configure the random number generator used for jitter to always return a number as close as possible to `1`.
+    2. Configure a fail point that forces 13 retries like in step 3.2.
+    3. Use the same callback defined in 3.3.
+    4. Let `with_backoff_time` be the duration of the withTransaction API call:
+        ```python
+           start = time.monotonic()
+           with client.start_session() as s:
+              s.with_transaction(callback)
+           end = time.monotonic()
+           no_backoff_time = end - start
+        ```
+5. Compare the durations of the two runs.
+    ```python
+    assertTrue(absolute_value(with_backoff_time - (no_backoff_time + 3.6 seconds)) < 0.5 seconds)
+    ```
+    The sum of 13 backoffs is roughly 3.6 seconds. There is a half-second window to account for potential variance
+    between the two runs.
+
 ## Changelog
 
+- 2206-07-08: Update Backoff test to use updated exponential formula.
+- 2026-04-02: [DRIVERS-3436](https://github.com/mongodb/specifications/pull/1920) Refine withTransaction timeout error
+    wrapping semantics and label propagation in spec and prose tests
+- 2026-03-03: Clarify exponential backoff jitter upper bound.
+- 2026-02-17: Clarify expected error when timeout is reached
+    [DRIVERS-3391](https://jira.mongodb.org/browse/DRIVERS-3391).
+- 2026-01-07: Fixed Retry Backoff is Enforced test accordingly to the updated spec.
+- 2025-11-18: Added Backoff test.
 - 2024-09-06: Migrated from reStructuredText to Markdown.
 - 2024-02-08: Converted legacy tests to unified format.
 - 2021-04-29: Remove text about write concern timeouts from prose test.
diff --git a/spec/transactions-convenient-api/unified/callback-aborts.json b/spec/transactions-convenient-api/unified/callback-aborts.json
index 206428715..dc8f7fb5a 100644
--- a/spec/transactions-convenient-api/unified/callback-aborts.json
+++ b/spec/transactions-convenient-api/unified/callback-aborts.json
@@ -308,10 +308,12 @@
                   "lsid": {
                     "$$sessionLsid": "session0"
                   },
-                  "autocommit": {
-                    "$$exists": false
-                  },
                   "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    }
+                  },
+                  "autocommit": {
                     "$$exists": false
                   },
                   "startTransaction": {
diff --git a/spec/transactions-convenient-api/unified/callback-aborts.yml b/spec/transactions-convenient-api/unified/callback-aborts.yml
index 9414040ec..d5a95d777 100644
--- a/spec/transactions-convenient-api/unified/callback-aborts.yml
+++ b/spec/transactions-convenient-api/unified/callback-aborts.yml
@@ -164,9 +164,10 @@ tests:
                   - { _id: 2 }
                 ordered: true
                 lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
                 # omitted fields
                 autocommit: { $$exists: false }
-                readConcern: { $$exists: false }
                 startTransaction: { $$exists: false }
                 writeConcern: { $$exists: false }
               commandName: insert
diff --git a/spec/transactions-convenient-api/unified/callback-commits.json b/spec/transactions-convenient-api/unified/callback-commits.json
index 06f791e9a..edda38641 100644
--- a/spec/transactions-convenient-api/unified/callback-commits.json
+++ b/spec/transactions-convenient-api/unified/callback-commits.json
@@ -381,10 +381,12 @@
                   "lsid": {
                     "$$sessionLsid": "session0"
                   },
-                  "autocommit": {
-                    "$$exists": false
-                  },
                   "readConcern": {
+                    "afterClusterTime": {
+                      "$$exists": true
+                    }
+                  },
+                  "autocommit": {
                     "$$exists": false
                   },
                   "startTransaction": {
diff --git a/spec/transactions-convenient-api/unified/callback-commits.yml b/spec/transactions-convenient-api/unified/callback-commits.yml
index b5cbb0415..7bd3e7fea 100644
--- a/spec/transactions-convenient-api/unified/callback-commits.yml
+++ b/spec/transactions-convenient-api/unified/callback-commits.yml
@@ -193,9 +193,10 @@ tests:
                   - { _id: 3 }
                 ordered: true
                 lsid: { $$sessionLsid: *session0 }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
                 # omitted fields
                 autocommit: { $$exists: false }
-                readConcern: { $$exists: false }
                 startTransaction: { $$exists: false }
                 writeConcern: { $$exists: false }
               commandName: insert
diff --git a/spec/transactions/unified/commit.json b/spec/transactions/unified/commit.json
index ab778d8df..de3be3047 100644
--- a/spec/transactions/unified/commit.json
+++ b/spec/transactions/unified/commit.json
@@ -209,6 +209,9 @@
                   "readConcern": {
                     "afterClusterTime": {
                       "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
                     }
                   },
                   "lsid": {
@@ -870,6 +873,9 @@
                   "readConcern": {
                     "afterClusterTime": {
                       "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
                     }
                   },
                   "lsid": {
@@ -1040,7 +1046,12 @@
                   ],
                   "ordered": true,
                   "readConcern": {
-                    "$$exists": false
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
                   },
                   "lsid": {
                     "$$sessionLsid": "session1"
@@ -1196,7 +1207,12 @@
                   ],
                   "ordered": true,
                   "readConcern": {
-                    "$$exists": false
+                    "afterClusterTime": {
+                      "$$exists": true
+                    },
+                    "level": {
+                      "$$exists": false
+                    }
                   },
                   "lsid": {
                     "$$sessionLsid": "session1"
diff --git a/spec/transactions/unified/commit.yml b/spec/transactions/unified/commit.yml
index d9af08489..e0f12fdb0 100644
--- a/spec/transactions/unified/commit.yml
+++ b/spec/transactions/unified/commit.yml
@@ -136,6 +136,7 @@ tests:
                 ordered: true
                 readConcern:
                   afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
                 lsid: { $$sessionLsid: *session0 }
                 txnNumber: { $numberLong: '2' }
                 startTransaction: true
@@ -522,6 +523,7 @@ tests:
                 ordered: true
                 readConcern:
                   afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
                 lsid: { $$sessionLsid: *session0 }
                 # txnNumber 2 was skipped
                 txnNumber: { $numberLong: '3' }
@@ -615,7 +617,9 @@ tests:
                 documents:
                   - { _id: 2 }
                 ordered: true
-                readConcern: { $$exists: false }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
                 lsid: { $$sessionLsid: *session1 }
                 txnNumber: { $$exists: false }
                 startTransaction: { $$exists: false }
@@ -698,7 +702,9 @@ tests:
                 documents:
                   - { _id: 2 }
                 ordered: true
-                readConcern: { $$exists: false }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
+                  level: { $$exists: false }
                 lsid: { $$sessionLsid: *session1 }
                 txnNumber: { $$exists: false }
                 startTransaction: { $$exists: false }
diff --git a/spec/transactions/unified/retryable-writes.json b/spec/transactions/unified/retryable-writes.json
index c196e6862..21ef22b8a 100644
--- a/spec/transactions/unified/retryable-writes.json
+++ b/spec/transactions/unified/retryable-writes.json
@@ -217,7 +217,9 @@
                   ],
                   "ordered": true,
                   "readConcern": {
-                    "$$exists": false
+                    "afterClusterTime": {
+                      "$$exists": true
+                    }
                   },
                   "lsid": {
                     "$$sessionLsid": "session0"
@@ -306,7 +308,9 @@
                   ],
                   "ordered": true,
                   "readConcern": {
-                    "$$exists": false
+                    "afterClusterTime": {
+                      "$$exists": true
+                    }
                   },
                   "lsid": {
                     "$$sessionLsid": "session0"
diff --git a/spec/transactions/unified/retryable-writes.yml b/spec/transactions/unified/retryable-writes.yml
index aa9c037d4..81339c9ad 100644
--- a/spec/transactions/unified/retryable-writes.yml
+++ b/spec/transactions/unified/retryable-writes.yml
@@ -132,7 +132,8 @@ tests:
                 documents:
                   - { _id: 2 }
                 ordered: true
-                readConcern: { $$exists: false }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
                 lsid: { $$sessionLsid: *session0 }
                 txnNumber: { $numberLong: '2' }
                 startTransaction: { $$exists: false }
@@ -175,7 +176,8 @@ tests:
                   - { _id: 4 }
                   - { _id: 5 }
                 ordered: true
-                readConcern: { $$exists: false }
+                readConcern:
+                  afterClusterTime: { $$exists: true }
                 lsid: { $$sessionLsid: *session0 }
                 txnNumber: { $numberLong: '4' }
                 startTransaction: { $$exists: false }