diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
index 6a0caab80f1..fc6fb790ea9 100644
--- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs
@@ -8,7 +8,10 @@ use crate::{
         BatchResponse, ContainerProperties, CosmosResponse, ItemResponse, ResourceResponse,
         ThroughputProperties,
     },
-    options::{BatchOptions, QueryOptions, ReadContainerOptions, ReadFeedRangesOptions},
+    options::{
+        BatchOptions, Precondition, QueryOptions, ReadContainerOptions, ReadFeedRangesOptions,
+        SessionToken,
+    },
     resource_context::{ResourceLink, ResourceType},
     transactional_batch::TransactionalBatch,
     DeleteContainerOptions, FeedItemIterator, ItemReadOptions, ItemWriteOptions, PartitionKey,
@@ -322,16 +325,9 @@ impl ContainerClient {
         let driver_pk = partition_key.into().into_driver_partition_key();
 
         // Create the driver operation and apply ItemWriteOptions fields.
-        let mut operation =
+        let operation =
             CosmosOperation::create_item(self.container_ref.clone(), driver_pk).with_body(body);
-
-        // Wire session token and precondition from SDK options onto the operation.
-        if let Some(session_token) = options.session_token {
-            operation = operation.with_session_token(session_token);
-        }
-        if let Some(precondition) = options.precondition {
-            operation = operation.with_precondition(precondition);
-        }
+        let operation = apply_item_options(operation, options.session_token, options.precondition);
 
         // Execute through the driver.
         let driver_response = self
@@ -417,20 +413,31 @@ impl ContainerClient {
         item: T,
         options: Option<ItemWriteOptions>,
     ) -> azure_core::Result<ItemResponse<()>> {
-        let link = self.items_link.item(item_id);
-        let options = options.clone().unwrap_or_default();
-        let excluded_regions = options.operation.excluded_regions.clone();
-        let mut cosmos_request = CosmosRequest::builder(OperationType::Replace, link)
-            .json(&item)
-            .partition_key(partition_key.into())
-            .excluded_regions(excluded_regions)
-            .build()?;
-        options.apply_headers(&mut cosmos_request.headers);
+        let options = options.unwrap_or_default();
+        let body = serde_json::to_vec(&item)?;
 
-        self.container_connection
-            .send(cosmos_request, Context::default())
-            .await
-            .map(ItemResponse::new)
+        // Build the driver's item reference from our stored container metadata.
+        let item_ref = ItemReference::from_name(
+            &self.container_ref,
+            partition_key.into().into_driver_partition_key(),
+            item_id.to_owned(),
+        );
+
+        // Create the driver operation and apply ItemWriteOptions fields.
+        let operation = CosmosOperation::replace_item(item_ref).with_body(body);
+        let operation = apply_item_options(operation, options.session_token, options.precondition);
+
+        // Execute through the driver.
+        let driver_response = self
+            .context
+            .driver
+            .execute_operation(operation, options.operation)
+            .await?;
+
+        // Bridge the driver response to the SDK response type.
+        Ok(ItemResponse::new(
+            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+        ))
     }
 
     /// Creates or replaces an item in the container.
@@ -506,21 +513,26 @@ impl ContainerClient {
         item: T,
         options: Option<ItemWriteOptions>,
     ) -> azure_core::Result<ItemResponse<()>> {
-        let options = options.clone().unwrap_or_default();
-        let excluded_regions = options.operation.excluded_regions.clone();
-        let mut cosmos_request =
-            CosmosRequest::builder(OperationType::Upsert, self.items_link.clone())
-                .json(&item)
-                .partition_key(partition_key.into())
-                .excluded_regions(excluded_regions)
-                .build()?;
-        options.apply_headers(&mut cosmos_request.headers);
+        let options = options.unwrap_or_default();
+        let body = serde_json::to_vec(&item)?;
+        let driver_pk = partition_key.into().into_driver_partition_key();
 
-        return self
-            .container_connection
-            .send(cosmos_request, Context::default())
-            .await
-            .map(ItemResponse::new);
+        // Create the driver operation and apply ItemWriteOptions fields.
+        let operation =
+            CosmosOperation::upsert_item(self.container_ref.clone(), driver_pk).with_body(body);
+        let operation = apply_item_options(operation, options.session_token, options.precondition);
+
+        // Execute through the driver.
+        let driver_response = self
+            .context
+            .driver
+            .execute_operation(operation, options.operation)
+            .await?;
+
+        // Bridge the driver response to the SDK response type.
+        Ok(ItemResponse::new(
+            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+        ))
     }
 
     /// Reads a specific item from the container.
@@ -567,15 +579,8 @@ impl ContainerClient {
         );
 
         // Create the driver operation.
-        let mut operation = CosmosOperation::read_item(item_ref);
-
-        // Wire session token and precondition from SDK options onto the operation.
-        if let Some(session_token) = options.session_token {
-            operation = operation.with_session_token(session_token);
-        }
-        if let Some(precondition) = options.precondition {
-            operation = operation.with_precondition(precondition);
-        }
+        let operation = CosmosOperation::read_item(item_ref);
+        let operation = apply_item_options(operation, options.session_token, options.precondition);
 
         // Execute through the driver.
         let driver_response = self
@@ -616,19 +621,30 @@ impl ContainerClient {
         item_id: &str,
         options: Option<ItemWriteOptions>,
     ) -> azure_core::Result<ItemResponse<()>> {
-        let link = self.items_link.item(item_id);
-        let options = options.clone().unwrap_or_default();
-        let excluded_regions = options.operation.excluded_regions.clone();
-        let mut cosmos_request = CosmosRequest::builder(OperationType::Delete, link)
-            .partition_key(partition_key.into())
-            .excluded_regions(excluded_regions)
-            .build()?;
-        options.apply_headers(&mut cosmos_request.headers);
+        let options = options.unwrap_or_default();
 
-        self.container_connection
-            .send(cosmos_request, Context::default())
-            .await
-            .map(ItemResponse::new)
+        // Build the driver's item reference from our stored container metadata.
+        let item_ref = ItemReference::from_name(
+            &self.container_ref,
+            partition_key.into().into_driver_partition_key(),
+            item_id.to_owned(),
+        );
+
+        // Create the driver operation (no body for delete).
+        let operation = CosmosOperation::delete_item(item_ref);
+        let operation = apply_item_options(operation, options.session_token, options.precondition);
+
+        // Execute through the driver.
+        let driver_response = self
+            .context
+            .driver
+            .execute_operation(operation, options.operation)
+            .await?;
+
+        // Bridge the driver response to the SDK response type.
+        Ok(ItemResponse::new(
+            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+        ))
     }
 
     /// Executes a single-partition query against items in the container.
@@ -906,6 +922,24 @@ impl ContainerClient {
     }
 }
 
+/// Applies optional `session_token` and `precondition` to a [`CosmosOperation`].
+///
+/// Both [`ItemReadOptions`] and [`ItemWriteOptions`] carry these fields;
+/// this helper avoids duplicating the wiring logic in every item operation.
+fn apply_item_options(
+    mut operation: CosmosOperation,
+    session_token: Option<SessionToken>,
+    precondition: Option<Precondition>,
+) -> CosmosOperation {
+    if let Some(session_token) = session_token {
+        operation = operation.with_session_token(session_token);
+    }
+    if let Some(precondition) = precondition {
+        operation = operation.with_precondition(precondition);
+    }
+    operation
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/cosmos_client_builder.rs b/sdk/cosmos/azure_data_cosmos/src/clients/cosmos_client_builder.rs
index 9ae7b6b39f3..ff7d6953bb3 100644
--- a/sdk/cosmos/azure_data_cosmos/src/clients/cosmos_client_builder.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/clients/cosmos_client_builder.rs
@@ -359,7 +359,7 @@ impl CosmosClientBuilder {
 
         let global_endpoint_manager = GlobalEndpointManager::new(
             endpoint.clone(),
-            preferred_regions,
+            preferred_regions.clone(),
             Vec::new(),
             pipeline_core.clone(),
         );
@@ -443,8 +443,12 @@ impl CosmosClientBuilder {
                 })?;
         }
         let driver_runtime = driver_runtime_builder.build().await?;
+        let driver_options =
+            azure_data_cosmos_driver::options::DriverOptions::builder(driver_account)
+                .with_preferred_regions(preferred_regions)
+                .build();
         let driver = driver_runtime
-            .get_or_create_driver(driver_account, None)
+            .get_or_create_driver(driver_options.account().clone(), Some(driver_options))
             .await?;
 
         Ok(CosmosClient {
diff --git a/sdk/cosmos/azure_data_cosmos/src/cosmos_request.rs b/sdk/cosmos/azure_data_cosmos/src/cosmos_request.rs
index f8b3a11b4c1..4ee5e0f1a59 100644
--- a/sdk/cosmos/azure_data_cosmos/src/cosmos_request.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/cosmos_request.rs
@@ -292,10 +292,7 @@ mod tests {
     use super::*;
     use crate::operation_context::OperationType;
     use crate::resource_context::ResourceType;
-    use crate::{
-        constants, CosmosClientOptions, ItemWriteOptions, OperationOptions, PartitionKey,
-        SessionToken,
-    };
+    use crate::{constants, PartitionKey};
 
     fn make_base_request(op: OperationType) -> CosmosRequest {
         let req = CosmosRequest::builder(op, ResourceLink::root(ResourceType::Documents))
@@ -402,59 +399,6 @@ mod tests {
         assert!(has_upsert);
     }
 
-    #[test]
-    fn prioritize_request_headers_over_client_headers() {
-        let mut request_custom_headers = std::collections::HashMap::new();
-        request_custom_headers.insert(
-            HeaderName::from_static("x-custom-header"),
-            HeaderValue::from_static("custom_value"),
-        );
-
-        let operation = OperationOptions::default().with_custom_headers(request_custom_headers);
-        let item_options = ItemWriteOptions {
-            operation,
-            ..Default::default()
-        }
-        .with_session_token(SessionToken::from("RequestSession"));
-        let mut req = CosmosRequest::builder(
-            OperationType::Create,
-            ResourceLink::root(ResourceType::Documents),
-        )
-        .build()
-        .unwrap();
-        item_options.apply_headers(&mut req.headers);
-
-        req.request_context.location_endpoint_to_route =
-            Some("https://example.com/".parse().unwrap());
-
-        let mut client_custom_headers = std::collections::HashMap::new();
-        client_custom_headers.insert(
-            HeaderName::from_static("x-custom-header"),
-            HeaderValue::from_static("custom_value-2"),
-        );
-
-        let client_operation =
-            OperationOptions::default().with_custom_headers(client_custom_headers);
-        let client_options =
-            CosmosClientOptions::default().with_operation_options(client_operation);
-        client_options.apply_headers(&mut req.headers);
-
-        let raw = req.into_raw_request();
-        let get_header = |name: &HeaderName| {
-            raw.headers()
-                .iter()
-                .find(|(n, _)| n == &name)
-                .map(|(_, v)| v.as_str())
-                .unwrap()
-        };
-
-        assert_eq!(get_header(&constants::SESSION_TOKEN), "RequestSession");
-        assert_eq!(
-            get_header(&HeaderName::from_static("x-custom-header")),
-            "custom_value"
-        );
-    }
-
     #[test]
     fn to_raw_request_batch_sets_batch_headers() {
         let req = make_base_request(OperationType::Batch);
diff --git a/sdk/cosmos/azure_data_cosmos/src/options/mod.rs b/sdk/cosmos/azure_data_cosmos/src/options/mod.rs
index 15f8deea49b..6486acc4d6c 100644
--- a/sdk/cosmos/azure_data_cosmos/src/options/mod.rs
+++ b/sdk/cosmos/azure_data_cosmos/src/options/mod.rs
@@ -19,21 +19,6 @@ pub use azure_data_cosmos_driver::options::{
     ThroughputControlGroupOptions,
 };
 
-// Temporary: these helpers allow the SDK pipeline to apply OperationOptions values
-// as HTTP headers. They will be removed when individual operations use the internal
-// pipeline directly.
-fn apply_precondition_headers(precondition: &Precondition, headers: &mut Headers) {
-    match precondition {
-        Precondition::IfMatch(etag) => {
-            headers.insert(headers::IF_MATCH, etag.to_string());
-        }
-        Precondition::IfNoneMatch(etag) => {
-            headers.insert(constants::IF_NONE_MATCH, etag.to_string());
-        }
-        _ => {}
-    }
-}
-
 // Temporary: applies the prefer header based on the content_response_on_write option.
 // Will be removed when write operations use the internal pipeline directly.
 fn apply_content_response_on_write_header(
@@ -238,31 +223,6 @@ impl ItemWriteOptions {
     }
 }
 
-impl ItemWriteOptions {
-    // Temporary: applies option values as HTTP headers for the SDK pipeline.
-    // Will be removed when write operations use the internal pipeline directly.
-    pub(crate) fn apply_headers(&self, headers: &mut Headers) {
-        if let Some(custom_headers) = self.operation.custom_headers() {
-            for (name, value) in custom_headers {
-                // Only insert if not already set — SDK/request headers take priority.
-                if headers.get_optional_str(name).is_none() {
-                    headers.insert(name.clone(), value.clone());
-                }
-            }
-        }
-        if let Some(session_token) = &self.session_token {
-            headers.insert(constants::SESSION_TOKEN, session_token.to_string());
-        }
-        if let Some(precondition) = &self.precondition {
-            apply_precondition_headers(precondition, headers);
-        }
-        apply_content_response_on_write_header(
-            self.operation.content_response_on_write.as_ref(),
-            headers,
-        );
-    }
-}
-
 /// Options for transactional batch operations.
 ///
 /// Used by [`ContainerClient::execute_transactional_batch()`](crate::clients::ContainerClient::execute_transactional_batch()).
@@ -407,69 +367,6 @@ mod tests {
         headers.into_iter().collect()
     }
 
-    #[test]
-    fn item_write_options_as_headers() {
-        let mut custom_headers = HashMap::new();
-        custom_headers.insert(
-            HeaderName::from_static("x-custom-header"),
-            HeaderValue::from_static("custom_value"),
-        );
-
-        let operation = OperationOptions::default().with_custom_headers(custom_headers);
-
-        let options = ItemWriteOptions {
-            operation,
-            ..Default::default()
-        }
-        .with_session_token("SessionToken".to_string())
-        .with_precondition(Precondition::IfMatch(ETag::from("etag_value")));
-
-        let mut headers_result = Headers::new();
-        options.apply_headers(&mut headers_result);
-
-        let headers_expected: Vec<(HeaderName, HeaderValue)> = vec![
-            ("x-custom-header".into(), "custom_value".into()),
-            (constants::SESSION_TOKEN, "SessionToken".into()),
-            (headers::IF_MATCH, "etag_value".into()),
-            (headers::PREFER, constants::PREFER_MINIMAL),
-        ];
-
-        assert_eq!(
-            headers_to_map(headers_result),
-            headers_to_map(headers_expected)
-        );
-    }
-
-    #[test]
-    fn custom_headers_should_not_override_sdk_set_headers() {
-        let mut custom_headers = HashMap::new();
-        custom_headers.insert(
-            constants::SESSION_TOKEN,
-            HeaderValue::from_static("CustomSession"),
-        );
-
-        let operation = OperationOptions::default().with_custom_headers(custom_headers);
-
-        let options = ItemWriteOptions {
-            operation,
-            ..Default::default()
-        }
-        .with_session_token("RealSessionToken".to_string());
-
-        let mut headers_result = Headers::new();
-        options.apply_headers(&mut headers_result);
-
-        let headers_expected: Vec<(HeaderName, HeaderValue)> = vec![
-            (constants::SESSION_TOKEN, "RealSessionToken".into()),
-            (headers::PREFER, constants::PREFER_MINIMAL),
-        ];
-
-        assert_eq!(
-            headers_to_map(headers_result),
-            headers_to_map(headers_expected)
-        );
-    }
-
     #[test]
     fn client_options_as_headers() {
         let mut custom_headers = HashMap::new();
@@ -497,39 +394,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn item_write_options_default_as_headers() {
-        let options = ItemWriteOptions::default();
-
-        let mut headers_result = Headers::new();
-        options.apply_headers(&mut headers_result);
-        let headers_result: Vec<(HeaderName, HeaderValue)> = headers_result.into_iter().collect();
-
-        let headers_expected: Vec<(HeaderName, HeaderValue)> =
-            vec![(headers::PREFER, constants::PREFER_MINIMAL)];
-
-        assert_eq!(headers_result, headers_expected);
-    }
-
-    #[test]
-    fn item_write_options_with_content_response_enabled() {
-        let mut operation = OperationOptions::default();
-        operation.content_response_on_write = Some(ContentResponseOnWrite::Enabled);
-
-        let options = ItemWriteOptions {
-            operation,
-            ..Default::default()
-        };
-
-        let mut headers_result = Headers::new();
-        options.apply_headers(&mut headers_result);
-        let headers_result: Vec<(HeaderName, HeaderValue)> = headers_result.into_iter().collect();
-
-        let headers_expected: Vec<(HeaderName, HeaderValue)> = vec![];
-
-        assert_eq!(headers_result, headers_expected);
-    }
-
     #[test]
     fn batch_options_as_headers() {
         let mut custom_headers = HashMap::new();
@@ -623,21 +487,4 @@ mod tests {
 
         assert_eq!(headers_result, headers_expected);
     }
-
-    #[test]
-    fn no_throughput_control_headers_from_apply_headers_alone() {
-        // apply_headers() does not set throughput control headers — those are
-        // applied by the driver pipeline. Verify that priority and bucket
-        // headers are absent after apply_headers() only.
-        let options = ItemWriteOptions::default();
-        let mut headers = Headers::new();
-        options.apply_headers(&mut headers);
-
-        assert!(headers
-            .get_optional_str(&constants::PRIORITY_LEVEL)
-            .is_none());
-        assert!(headers
-            .get_optional_str(&constants::THROUGHPUT_BUCKET)
-            .is_none());
-    }
 }
diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs
index b9f4a4ad76e..ebb2908a68c 100644
--- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs
+++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs
@@ -996,86 +996,3 @@ pub async fn fault_injection_enable_disable_rule() -> Result<(), Box<dyn Error>>
     )
     .await
 }
-
-/// Verifies that the pkranges request uses the correct URL and returns 200.
-///
-/// Uses a "spy" fault injection rule with no error or custom response. The rule
-/// matches `MetadataPartitionKeyRanges` requests but lets them pass through to the
-/// real service, recording the response status code. This proves:
-/// 1. The pkranges code path is exercised (`hit_count >= 1`).
-/// 2. The service returns 200 (the URL uses name-based addressing, not mixed RID).
-/// 3. The item operation succeeds end-to-end with correct partition key routing.
-#[tokio::test]
-#[cfg_attr(
-    not(test_category = "emulator"),
-    ignore = "requires test_category 'emulator'"
-)]
-pub async fn fault_injection_pkrange_readfeed_succeeds() -> Result<(), Box<dyn Error>> {
-    // Spy rule: no error, no custom response → passthrough with status recording.
-    let spy_result = FaultInjectionResultBuilder::new().build();
-
-    let condition = FaultInjectionConditionBuilder::new()
-        .with_operation_type(FaultOperationType::MetadataPartitionKeyRanges)
-        .build();
-
-    let rule = Arc::new(
-        FaultInjectionRuleBuilder::new("pkrange-spy", spy_result)
-            .with_condition(condition)
-            .build(),
-    );
-
-    let rule_handle = Arc::clone(&rule);
-
-    let fault_builder = FaultInjectionClientBuilder::new().with_rule(rule);
-
-    TestClient::run_with_unique_db(
-        async move |run_context, db_client| {
-            let container_id = format!("PkRangeTest-{}", Uuid::new_v4());
-            let _container_client = run_context
-                .create_container_with_throughput(
-                    db_client,
-                    ContainerProperties::new(container_id.clone(), "/partition_key".into()),
-                    ThroughputProperties::manual(400),
-                )
-                .await?;
-
-            let unique_id = Uuid::new_v4().to_string();
-            let item = create_test_item(&unique_id);
-            let pk = format!("Partition-{}", unique_id);
-
-            // Use the fault client so the spy rule is in the HTTP pipeline.
-            let fault_client = run_context
-                .fault_client()
-                .expect("fault client should be available");
-            let fault_db_client = fault_client.database_client(db_client.id());
-            let fault_container_client = fault_db_client.container_client(&container_id).await?;
-
-            // Upsert an item — this triggers partition key range resolution
-            // through the fault client's SDK pipeline where the spy rule can observe.
-            // NOTE: create_item is cut over to the driver which handles routing
-            // via the gateway and does not resolve pkranges through the SDK pipeline.
-            fault_container_client
-                .upsert_item(&pk, item, None)
-                .await?;
-
-            // The spy rule should have been hit during the pkrange fetch.
-            assert!(
-                rule_handle.hit_count() >= 1,
-                "Expected the MetadataPartitionKeyRanges spy rule to be hit at least once, but hit_count was {}",
-                rule_handle.hit_count()
-            );
-
-            // Verify the pkranges request returned 200 from the service.
-            let statuses = rule_handle.passthrough_statuses();
-            assert!(
-                statuses.contains(&StatusCode::Ok),
-                "Expected pkranges request to return 200, got: {:?}",
-                statuses
-            );
-
-            Ok(())
-        },
-        Some(TestOptions::new().with_fault_injection_builder(fault_builder)),
-    )
-    .await
-}
diff --git a/sdk/cosmos/azure_data_cosmos/tests/framework/test_client.rs b/sdk/cosmos/azure_data_cosmos/tests/framework/test_client.rs
index f1261bb020e..ff193a2eb0d 100644
--- a/sdk/cosmos/azure_data_cosmos/tests/framework/test_client.rs
+++ b/sdk/cosmos/azure_data_cosmos/tests/framework/test_client.rs
@@ -746,15 +746,20 @@ impl TestRunContext {
 
             let container_id = &created_properties.id;
 
-            // Wait for hub region client to successfully read the container
+            // Wait for hub region client to successfully resolve and read the container.
+            // Both `container_client()` (which resolves metadata via the driver) and
+            // `read()` can fail with 404 while the container replicates.
             loop {
-                match hub_client
-                    .database_client(db_client.id())
-                    .container_client(container_id)
-                    .await?
-                    .read(None)
-                    .await
-                {
+                let result = async {
+                    hub_client
+                        .database_client(db_client.id())
+                        .container_client(container_id)
+                        .await?
+                        .read(None)
+                        .await
+                }
+                .await;
+                match result {
                     Ok(_) => break,
                     Err(e) => {
                         println!(
@@ -767,15 +772,18 @@ impl TestRunContext {
                 }
             }
 
-            // Wait for satellite region client to successfully read the container
+            // Wait for satellite region client to successfully resolve and read the container.
             loop {
-                match satellite_client
-                    .database_client(db_client.id())
-                    .container_client(container_id)
-                    .await?
-                    .read(None)
-                    .await
-                {
+                let result = async {
+                    satellite_client
+                        .database_client(db_client.id())
+                        .container_client(container_id)
+                        .await?
+                        .read(None)
+                        .await
+                }
+                .await;
+                match result {
                     Ok(_) => break,
                     Err(e) => {
                         println!(
diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs
index 67a9323f32b..86afec06471 100644
--- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs
+++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs
@@ -53,13 +53,14 @@ impl tracing::field::Visit for StringVisitor {
     }
 }
 
-/// Finds log lines containing the expected operation and returns them
-fn find_upsert_document_logs(logs: &[String]) -> Vec<String> {
+/// Finds log lines from the driver's operation pipeline containing routing decisions.
+///
+/// The driver emits `tracing::debug!(routing_decision = %routing, "routing decision made")`
+/// which includes the resolved region and endpoint URL.
+fn find_upsert_routing_logs(logs: &[String]) -> Vec<String> {
     logs.iter()
         .filter(|line| {
-            line.contains("azure_data_cosmos::retry_handler")
-                && line.contains("Upsert")
-                && line.contains("Document")
+            line.contains("azure_data_cosmos_driver") && line.contains("routing decision made")
         })
         .cloned()
         .collect()
@@ -86,7 +87,7 @@ fn create_container_and_write_item<'a>(
             .create_container_with_throughput(db_client, properties, throughput)
             .await?;
 
-        // This upsert operation should be logged by the retry_handler
+        // This upsert operation triggers a routing decision log in the driver
         container_client
             .upsert_item(
                 "item1",
@@ -142,7 +143,7 @@ pub async fn multi_write_preferred_locations() -> Result<(), Box<dyn Error>> {
     // Check that the upsert went to the hub region
     {
         let logs = log_buffer.lock().unwrap();
-        let upsert_logs = find_upsert_document_logs(&logs);
+        let upsert_logs = find_upsert_routing_logs(&logs);
         println!("Hub region upsert logs: {:?}", upsert_logs);
 
         assert!(
@@ -184,7 +185,7 @@ pub async fn multi_write_preferred_locations() -> Result<(), Box<dyn Error>> {
     // Check that the upsert went to the satellite region
     {
         let logs = log_buffer.lock().unwrap();
-        let upsert_logs = find_upsert_document_logs(&logs);
+        let upsert_logs = find_upsert_routing_logs(&logs);
         println!("Satellite region upsert logs: {:?}", upsert_logs);
 
         assert!(
diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs
index b26c4adae9f..ad151e362c1 100644
--- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs
+++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs
@@ -348,13 +348,19 @@ pub async fn fault_injection_read_region_retry_503() -> Result<(), Box<dyn Error
     .await
 }
 
-/// Test write region retries - inject 503 for primary region, verify cross region retries.
+/// Test that a transport-generated 503 on a non-idempotent write aborts (not retried).
+///
+/// Fault injection simulates transport-level failures (e.g. connection drops) which
+/// produce a synthetic 503/20003 (`TransportGenerated503`). These are distinct from
+/// HTTP-level 503s returned by the service: transport errors on non-idempotent writes
+/// are NOT retried because we cannot know whether the server processed the request.
+/// HTTP-level 503s ARE retried for all operations (see retry_evaluation.rs Block 2).
 #[tokio::test]
 #[cfg_attr(
     not(test_category = "multi_write"),
     ignore = "requires test_category 'multi_write'"
 )]
-pub async fn fault_injection_write_region_retry_503() -> Result<(), Box<dyn Error>> {
+pub async fn fault_injection_transport_generated_503_write_aborts() -> Result<(), Box<dyn Error>> {
     let server_error = FaultInjectionResultBuilder::new()
         .with_error(FaultInjectionErrorType::ServiceUnavailable)
         .build();
@@ -364,7 +370,7 @@ pub async fn fault_injection_write_region_retry_503() -> Result<(), Box<dyn Erro
         .with_region(HUB_REGION)
         .build();
 
-    let rule = FaultInjectionRuleBuilder::new("write-region-503", server_error)
+    let rule = FaultInjectionRuleBuilder::new("write-region-transport-503", server_error)
         .with_condition(condition)
         .with_hit_limit(1)
         .build();
@@ -400,24 +406,15 @@ pub async fn fault_injection_write_region_retry_503() -> Result<(), Box<dyn Erro
             };
             let pk = format!("Partition-{}", unique_id);
 
-            // Try to create using fault client - should  succeed via retry
+            // Transport-generated 503 on a non-idempotent write (upsert) should NOT
+            // be retried — the driver cannot know if the server processed the request.
             let result = fault_container_client.upsert_item(&pk, &item, None).await;
 
             assert!(
-                result.is_ok(),
-                "Write should succeed via retry, but got error: {:?}",
-                result.err()
+                result.is_err(),
+                "Transport-generated 503 on non-idempotent write should abort, not retry"
             );
 
-            let response = result.unwrap();
-            if let Some(request_url) = response.request_url().map(|u| u.to_string()) {
-                // Verify the request went to a different endpoint than the faulted one
-                assert!(
-                    request_url.contains(SATELLITE_REGION.as_str()),
-                    "request should have failed over to secondary region"
-                );
-            }
-
             Ok(())
         },
         Some(
diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/cosmos_driver.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/cosmos_driver.rs
index 2343a4f63bd..5f74982b439 100644
--- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/cosmos_driver.rs
+++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/cosmos_driver.rs
@@ -752,6 +752,7 @@ impl CosmosDriver {
             refresh_callback,
             runtime.connection_pool().is_gateway20_allowed(),
             endpoint_unavailability_ttl,
+            options.preferred_regions().to_vec(),
         ));
 
         Self {
diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs
index 1e130fccd9a..d4cd938e5c0 100644
--- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs
+++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs
@@ -20,7 +20,7 @@ use crate::{
     },
     models::{
         request_header_names, AccountEndpoint, ActivityId, CosmosOperation, CosmosResponse,
-        Credential, DefaultConsistencyLevel, SessionToken, SubStatusCode,
+        Credential, DefaultConsistencyLevel, OperationType, SessionToken, SubStatusCode,
     },
     options::{OperationOptionsView, ReadConsistencyStrategy, ThroughputControlGroupSnapshot},
 };
@@ -495,6 +495,15 @@ fn build_transport_request(
         }
     }
 
+    // Cosmos DB uses POST for both create and upsert; the service
+    // distinguishes them via this header.
+    if operation.operation_type() == OperationType::Upsert {
+        headers.insert(
+            HeaderName::from_static(request_header_names::IS_UPSERT),
+            HeaderValue::from_static("true"),
+        );
+    }
+
     // Add operation type header for fault injection rule matching
     #[cfg(feature = "fault_injection")]
     {
@@ -1238,6 +1247,58 @@ mod tests {
         assert_eq!(routing.endpoint, fallback_endpoint);
     }
 
+    #[test]
+    fn build_transport_request_sets_is_upsert_header() {
+        let operation = CosmosOperation::upsert_item(test_container(), PartitionKey::from("pk1"))
+            .with_body(b"{}".to_vec());
+
+        let routing = test_routing();
+        let activity_id = ActivityId::from_string("default-activity".to_string());
+        let ctx = TransportRequestContext {
+            routing: &routing,
+            activity_id: &activity_id,
+            execution_context: ExecutionContext::Initial,
+            deadline: None,
+            resolved_session_token: None,
+            throughput_control: None,
+        };
+        let request =
+            build_transport_request(&operation, None, &ctx).expect("request should build");
+
+        let is_upsert = request
+            .headers
+            .get_optional_str(&HeaderName::from_static("x-ms-documentdb-is-upsert"))
+            .expect("is-upsert header should be set");
+        assert_eq!(is_upsert, "true");
+    }
+
+    #[test]
+    fn build_transport_request_omits_is_upsert_header_for_create() {
+        let operation = CosmosOperation::create_item(test_container(), PartitionKey::from("pk1"))
+            .with_body(b"{}".to_vec());
+
+        let routing = test_routing();
+        let activity_id = ActivityId::from_string("default-activity".to_string());
+        let ctx = TransportRequestContext {
+            routing: &routing,
+            activity_id: &activity_id,
+            execution_context: ExecutionContext::Initial,
+            deadline: None,
+            resolved_session_token: None,
+            throughput_control: None,
+        };
+        let request =
+            build_transport_request(&operation, None, &ctx).expect("request should build");
+
+        assert!(
+            request
+                .headers
+                .get_optional_str(&HeaderName::from_static("x-ms-documentdb-is-upsert"))
+                .is_none(),
+            "is-upsert header should not be set for create"
+        );
+    }
+
     #[test]
     fn build_transport_request_sets_priority_level_header() {
         let container = test_container();
diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/retry_evaluation.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/retry_evaluation.rs
index b034a19dc06..8184d0068cb 100644
--- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/retry_evaluation.rs
+++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/retry_evaluation.rs
@@ -10,8 +10,9 @@
 //! - Transport error (Sent/Unknown, non-idempotent) → Abort
 //! - 403/3 WriteForbidden → FailoverRetry + refresh + mark unavailable
 //! - 404/1002 ReadSessionNotAvailable → SessionRetry (advances region)
-//! - 503, 429/3092, 410 → FailoverRetry + mark partition/endpoint unavailable
-//! - 500 (reads only) → FailoverRetry
+//! - 429/3092, 410 → FailoverRetry + mark partition/endpoint unavailable (idempotency-gated)
+//! - ≥500, 408 → FailoverRetry for all ops (reads + writes)
+//!   - 503 additionally marks partition unavailable
 //! - Other HTTP errors → Abort
 
 use azure_core::http::headers::Headers;
@@ -87,15 +88,12 @@ pub(crate) fn evaluate_transport_result(
                 );
             }
 
+            // 429/3092 (SystemResourceUnavailable) and 410 (Gone).
             let is_system_resource_unavailable = status.is_throttled()
                 && status.sub_status() == Some(SubStatusCode::SYSTEM_RESOURCE_UNAVAILABLE);
-            let is_service_unavailable =
-                status.status_code() == azure_core::http::StatusCode::ServiceUnavailable;
             let is_gone = status.is_gone();
 
-            if (is_system_resource_unavailable || is_service_unavailable || is_gone)
-                && retry_state.can_retry_failover()
-            {
+            if (is_system_resource_unavailable || is_gone) && retry_state.can_retry_failover() {
                 if request_definitely_not_sent {
                     return (
                         OperationAction::FailoverRetry {
@@ -138,19 +136,36 @@ pub(crate) fn evaluate_transport_result(
                 );
             }
 
-            if status.status_code() == azure_core::http::StatusCode::InternalServerError
-                && operation.is_read_only()
+            // Server errors (≥500) and request timeouts (408).
+            // Retries ALL operations (reads + writes).
+            // Cross-region retry will be attempted, if not possible in-region retry will be attempted.
+            let status_code = status.status_code();
+            if (status_code.is_server_error()
+                || status_code == azure_core::http::StatusCode::RequestTimeout)
                 && retry_state.can_retry_failover()
             {
+                let mut effects = vec![LocationEffect::MarkEndpointUnavailable {
+                    endpoint: endpoint.clone(),
+                    reason: UnavailableReason::InternalServerError,
+                }];
+                if status_code == azure_core::http::StatusCode::ServiceUnavailable {
+                    effects.push(LocationEffect::MarkPartitionUnavailable(
+                        UnavailablePartition {
+                            // TODO(partition-routing): Wire the actual partition key range ID from
+                            // TransportResult or CosmosOperation once partition-level
+                            // routing is implemented.
+                            partition_key_range_id: String::new(),
+                            region: endpoint.region().cloned(),
+                            is_read: operation.is_read_only(),
+                        },
+                    ));
+                }
                 return (
                     OperationAction::FailoverRetry {
                         new_state: retry_state.clone().advance_failover(),
                         delay: None,
                     },
-                    vec![LocationEffect::MarkEndpointUnavailable {
-                        endpoint: endpoint.clone(),
-                        reason: UnavailableReason::InternalServerError,
-                    }],
+                    effects,
                 );
             }
 
@@ -543,7 +558,7 @@ mod tests {
     }
 
     #[test]
-    fn service_unavailable_sent_non_idempotent_aborts() {
+    fn service_unavailable_write_retries_and_marks_partition() {
         let op = make_create_operation();
         let result = make_http_error(StatusCode::ServiceUnavailable);
         let state = OperationRetryState::initial(0, false, Vec::new(), 3, 1);
@@ -552,8 +567,13 @@ mod tests {
         );
 
         let (action, effects) = evaluate_transport_result(&op, &endpoint, result, &state);
-        assert!(matches!(action, OperationAction::Abort { .. }));
-        assert!(effects.is_empty());
+        assert!(matches!(action, OperationAction::FailoverRetry { .. }));
+        assert!(effects
+            .iter()
+            .any(|e| matches!(e, LocationEffect::MarkEndpointUnavailable { .. })));
+        assert!(effects
+            .iter()
+            .any(|e| matches!(e, LocationEffect::MarkPartitionUnavailable(_))));
     }
 
     #[test]
@@ -599,4 +619,56 @@ mod tests {
             .iter()
             .any(|e| matches!(e, LocationEffect::MarkEndpointUnavailable { .. })));
     }
+
+    #[test]
+    fn internal_server_error_on_write_retries() {
+        let op = make_create_operation();
+        let result = make_http_error(StatusCode::InternalServerError);
+        let state = OperationRetryState::initial(0, false, Vec::new(), 3, 1);
+        let endpoint = CosmosEndpoint::global(
+            url::Url::parse("https://test.documents.azure.com:443/").unwrap(),
+        );
+
+        let (action, effects) = evaluate_transport_result(&op, &endpoint, result, &state);
+        assert!(matches!(action, OperationAction::FailoverRetry { .. }));
+        assert!(effects
+            .iter()
+            .any(|e| matches!(e, LocationEffect::MarkEndpointUnavailable { .. })));
+        // 500 should NOT mark partition unavailable (only 503 does)
+        assert!(!effects
+            .iter()
+            .any(|e| matches!(e, LocationEffect::MarkPartitionUnavailable(_))));
+    }
+
+    #[test]
+    fn request_timeout_read_retries() {
+        let op = make_read_operation();
+        let result = make_http_error(StatusCode::RequestTimeout);
+        let state = OperationRetryState::initial(0, false, Vec::new(), 3, 1);
+        let endpoint = CosmosEndpoint::global(
+            url::Url::parse("https://test.documents.azure.com:443/").unwrap(),
+        );
+
+        let (action, effects) = evaluate_transport_result(&op, &endpoint, result, &state);
+        assert!(matches!(action, OperationAction::FailoverRetry { .. }));
+        assert!(effects
+            .iter()
+            .any(|e| matches!(e, LocationEffect::MarkEndpointUnavailable { .. })));
+    }
+
+    #[test]
+    fn request_timeout_write_retries() {
+        let op = make_create_operation();
+        let result = make_http_error(StatusCode::RequestTimeout);
+        let state = OperationRetryState::initial(0, false, Vec::new(), 3, 1);
+        let endpoint = CosmosEndpoint::global(
+            url::Url::parse("https://test.documents.azure.com:443/").unwrap(),
+        );
+
+        let (action, effects) = evaluate_transport_result(&op, &endpoint, result, &state);
+        assert!(matches!(action, OperationAction::FailoverRetry { .. }));
+        assert!(effects
+            .iter()
+            .any(|e| matches!(e, LocationEffect::MarkEndpointUnavailable { .. })));
+    }
 }
diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/routing/location_state_store.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/routing/location_state_store.rs
index 69d7bd5d5a6..44c48cbee18 100644
--- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/routing/location_state_store.rs
+++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/routing/location_state_store.rs
@@ -17,6 +17,7 @@ use futures::future::BoxFuture;
 use crate::{
     driver::cache::{AccountMetadataCache, AccountProperties},
     models::AccountEndpoint,
+    options::Region,
 };
 
 use super::{
@@ -62,6 +63,7 @@ pub(crate) struct LocationStateStore {
     account_endpoint: AccountEndpoint,
     account_refresh_fn: AccountRefreshFn,
     default_endpoint: CosmosEndpoint,
+    preferred_regions: Vec<Region>,
     gateway20_enabled: bool,
     endpoint_unavailability_ttl: Duration,
     refresh_interval: Duration,
@@ -126,6 +128,7 @@ impl LocationStateStore {
         account_refresh_fn: AccountRefreshFn,
         gateway20_enabled: bool,
         endpoint_unavailability_ttl: Duration,
+        preferred_regions: Vec<Region>,
     ) -> Self {
         let account_state = AccountEndpointState::single(default_endpoint.clone());
 
@@ -141,6 +144,7 @@ impl LocationStateStore {
             account_endpoint,
             account_refresh_fn,
             default_endpoint,
+            preferred_regions,
             gateway20_enabled,
             endpoint_unavailability_ttl,
             // TODO(refresh-config): Make refresh interval configurable.
@@ -342,6 +346,7 @@ impl LocationStateStore {
                 default_endpoint.clone(),
                 Some(current.generation),
                 self.gateway20_enabled,
+                &self.preferred_regions,
             );
             // Carry forward unavailability marks from the current state,
             // filtering out entries that have expired past the configured TTL.
@@ -437,6 +442,7 @@ mod tests {
             refresh,
             false,
             Duration::from_secs(60),
+            Vec::new(),
         );
 
         store
@@ -473,6 +479,7 @@ mod tests {
             refresh,
             false,
             Duration::from_secs(60),
+            Vec::new(),
         );
 
         store
@@ -503,6 +510,7 @@ mod tests {
             refresh,
             false,
             Duration::from_secs(60),
+            Vec::new(),
         );
 
         let properties = Arc::new(test_refresh_payload());
diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/routing/routing_systems.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/routing/routing_systems.rs
index 1a934086e9a..8ab9d704bcb 100644
--- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/routing/routing_systems.rs
+++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/routing/routing_systems.rs
@@ -12,22 +12,22 @@ use std::{
 use tracing::warn;
 
 use crate::driver::cache::AccountProperties;
+use crate::options::Region;
 
 use super::{AccountEndpointState, CosmosEndpoint, UnavailableReason};
 
 /// Builds account endpoint state from account metadata.
 ///
-/// TODO: Accept `preferred_locations: &[Region]` to reorder endpoint lists
-/// based on user configuration (derived from `application_region` via
-/// `generate_preferred_region_list` in `azure_data_cosmos`). Wire this when
-/// operations in `azure_data_cosmos` are migrated to the driver's
-/// `execute_operation` API — that cross-crate change is the natural point
-/// to thread preferred regions through `DriverOptions` → `LocationStateStore`.
+/// When `preferred_regions` is non-empty, read and write endpoint lists are
+/// reordered so that endpoints matching the preferred regions appear first
+/// (in preference order). Endpoints whose region is not in the preferred
+/// list are appended in their original account-metadata order.
 pub(crate) fn build_account_endpoint_state(
     properties: &AccountProperties,
     default_endpoint: CosmosEndpoint,
     previous_generation: Option<u64>,
     gateway20_enabled: bool,
+    preferred_regions: &[Region],
 ) -> AccountEndpointState {
     let generation = previous_generation.map_or(0, |g| g.saturating_add(1));
 
@@ -43,6 +43,13 @@ pub(crate) fn build_account_endpoint_state(
         gateway20_enabled,
     );
 
+    if !preferred_regions.is_empty() {
+        preferred_read_endpoints =
+            reorder_by_preferred_regions(preferred_read_endpoints, preferred_regions);
+        preferred_write_endpoints =
+            reorder_by_preferred_regions(preferred_write_endpoints, preferred_regions);
+    }
+
     if preferred_read_endpoints.is_empty() {
         preferred_read_endpoints.push(default_endpoint.clone());
     }
@@ -128,6 +135,36 @@ fn parse_thin_client_locations(
     urls
 }
 
+/// Reorders endpoints so that those matching `preferred_regions` appear first,
+/// in the same order as the preference list. Endpoints whose region is not in
+/// the preference list (or that have no region, e.g. global endpoints) are
+/// appended in their original order.
+fn reorder_by_preferred_regions(
+    endpoints: Vec<CosmosEndpoint>,
+    preferred_regions: &[Region],
+) -> Vec<CosmosEndpoint> {
+    let mut ordered = Vec::with_capacity(endpoints.len());
+    let mut remaining: Vec<Option<CosmosEndpoint>> = endpoints.into_iter().map(Some).collect();
+
+    for region in preferred_regions {
+        for slot in remaining.iter_mut() {
+            if let Some(ep) = slot {
+                if ep.region().is_some_and(|r| r == region) {
+                    ordered.push(slot.take().unwrap());
+                    break;
+                }
+            }
+        }
+    }
+
+    // Append any endpoints not matched by the preferred list.
+    for ep in remaining.into_iter().flatten() {
+        ordered.push(ep);
+    }
+
+    ordered
+}
+
 /// Returns a new state with an endpoint marked unavailable.
 pub(crate) fn mark_endpoint_unavailable(
     state: &AccountEndpointState,
@@ -176,6 +213,7 @@ pub(crate) fn expire_unavailable_endpoints(
 mod tests {
     use super::*;
     use crate::driver::cache::AccountProperties;
+    use crate::options::Region;
 
     fn default_endpoint() -> CosmosEndpoint {
         CosmosEndpoint::global(url::Url::parse("https://test.documents.azure.com:443/").unwrap())
@@ -204,7 +242,7 @@ mod tests {
     #[test]
     fn build_state_uses_metadata_locations() {
         let state =
-            build_account_endpoint_state(&test_properties(), default_endpoint(), None, false);
+            build_account_endpoint_state(&test_properties(), default_endpoint(), None, false, &[]);
         assert_eq!(state.generation, 0);
         assert_eq!(state.preferred_write_endpoints.len(), 1);
         assert_eq!(state.preferred_read_endpoints.len(), 1);
@@ -232,7 +270,7 @@ mod tests {
         }))
         .unwrap();
 
-        let state = build_account_endpoint_state(&properties, default_endpoint(), None, true);
+        let state = build_account_endpoint_state(&properties, default_endpoint(), None, true, &[]);
 
         assert!(state.preferred_read_endpoints[0].gateway20_url().is_some());
         assert!(state.preferred_write_endpoints[0].gateway20_url().is_none());
@@ -260,7 +298,7 @@ mod tests {
         }))
         .unwrap();
 
-        let state = build_account_endpoint_state(&properties, default_endpoint(), None, true);
+        let state = build_account_endpoint_state(&properties, default_endpoint(), None, true, &[]);
 
         assert!(state.preferred_read_endpoints[0].gateway20_url().is_some());
         assert!(state.preferred_write_endpoints[0].gateway20_url().is_some());
@@ -269,7 +307,7 @@ mod tests {
     #[test]
     fn mark_and_expire_unavailable_endpoint() {
         let state =
-            build_account_endpoint_state(&test_properties(), default_endpoint(), None, false);
+            build_account_endpoint_state(&test_properties(), default_endpoint(), None, false, &[]);
         let endpoint = state.preferred_read_endpoints[0].clone();
         let marked =
             mark_endpoint_unavailable(&state, &endpoint, UnavailableReason::TransportError);
@@ -282,4 +320,135 @@ mod tests {
         );
         assert!(expired.unavailable_endpoints.is_empty());
     }
+
+    fn multi_region_properties() -> AccountProperties {
+        serde_json::from_value(serde_json::json!({
+            "_self": "",
+            "id": "test",
+            "_rid": "test.documents.azure.com",
+            "media": "//media/",
+            "addresses": "//addresses/",
+            "_dbs": "//dbs/",
+            "writableLocations": [
+                { "name": "eastus", "databaseAccountEndpoint": "https://test-eastus.documents.azure.com:443/" },
+                { "name": "westus2", "databaseAccountEndpoint": "https://test-westus2.documents.azure.com:443/" },
+                { "name": "westus3", "databaseAccountEndpoint": "https://test-westus3.documents.azure.com:443/" }
+            ],
+            "readableLocations": [
+                { "name": "eastus", "databaseAccountEndpoint": "https://test-eastus.documents.azure.com:443/" },
+                { "name": "westus2", "databaseAccountEndpoint": "https://test-westus2.documents.azure.com:443/" },
+                { "name": "westus3", "databaseAccountEndpoint": "https://test-westus3.documents.azure.com:443/" }
+            ],
+            "enableMultipleWriteLocations": true,
+            "userReplicationPolicy": { "minReplicaSetSize": 3, "maxReplicasetSize": 4 },
+            "userConsistencyPolicy": { "defaultConsistencyLevel": "Session" },
+            "systemReplicationPolicy": { "minReplicaSetSize": 3, "maxReplicasetSize": 4 },
+            "readPolicy": { "primaryReadCoefficient": 1, "secondaryReadCoefficient": 1 },
+            "queryEngineConfiguration": "{}"
+        }))
+        .unwrap()
+    }
+
+    #[test]
+    fn preferred_regions_reorder_write_endpoints() {
+        let preferred = vec![Region::WEST_US_3, Region::EAST_US];
+        let state = build_account_endpoint_state(
+            &multi_region_properties(),
+            default_endpoint(),
+            None,
+            false,
+            &preferred,
+        );
+
+        assert_eq!(state.preferred_write_endpoints.len(), 3);
+        assert_eq!(
+            state.preferred_write_endpoints[0].region().unwrap(),
+            &Region::WEST_US_3
+        );
+        assert_eq!(
+            state.preferred_write_endpoints[1].region().unwrap(),
+            &Region::EAST_US
+        );
+        assert_eq!(
+            state.preferred_write_endpoints[2].region().unwrap(),
+            &Region::WEST_US_2
+        );
+    }
+
+    #[test]
+    fn preferred_regions_reorder_read_endpoints() {
+        let preferred = vec![Region::WEST_US_2, Region::WEST_US_3];
+        let state = build_account_endpoint_state(
+            &multi_region_properties(),
+            default_endpoint(),
+            None,
+            false,
+            &preferred,
+        );
+
+        assert_eq!(state.preferred_read_endpoints.len(), 3);
+        assert_eq!(
+            state.preferred_read_endpoints[0].region().unwrap(),
+            &Region::WEST_US_2
+        );
+        assert_eq!(
+            state.preferred_read_endpoints[1].region().unwrap(),
+            &Region::WEST_US_3
+        );
+        assert_eq!(
+            state.preferred_read_endpoints[2].region().unwrap(),
+            &Region::EAST_US
+        );
+    }
+
+    #[test]
+    fn preferred_regions_unknown_regions_are_skipped() {
+        let preferred = vec![Region::new("nonexistent"), Region::WEST_US_3];
+        let state = build_account_endpoint_state(
+            &multi_region_properties(),
+            default_endpoint(),
+            None,
+            false,
+            &preferred,
+        );
+
+        // westus3 should be first; the nonexistent region is skipped.
+        assert_eq!(
+            state.preferred_write_endpoints[0].region().unwrap(),
+            &Region::WEST_US_3
+        );
+        assert_eq!(
+            state.preferred_write_endpoints[1].region().unwrap(),
+            &Region::EAST_US
+        );
+        assert_eq!(
+            state.preferred_write_endpoints[2].region().unwrap(),
+            &Region::WEST_US_2
+        );
+    }
+
+    #[test]
+    fn empty_preferred_regions_preserves_original_order() {
+        let state = build_account_endpoint_state(
+            &multi_region_properties(),
+            default_endpoint(),
+            None,
+            false,
+            &[],
+        );
+
+        // Original account-metadata order: eastus, westus2, westus3.
+        assert_eq!(
+            state.preferred_write_endpoints[0].region().unwrap(),
+            &Region::EAST_US
+        );
+        assert_eq!(
+            state.preferred_write_endpoints[1].region().unwrap(),
+            &Region::WEST_US_2
+        );
+        assert_eq!(
+            state.preferred_write_endpoints[2].region().unwrap(),
+            &Region::WEST_US_3
+        );
+    }
 }
diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_headers.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_headers.rs
index f62d6d3188e..756bd36f7c2 100644
--- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_headers.rs
+++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_headers.rs
@@ -20,6 +20,7 @@ pub(crate) mod request_header_names {
     pub const IF_MATCH: &str = "if-match";
     pub const IF_NONE_MATCH: &str = "if-none-match";
     pub const PREFER: &str = "prefer";
+    pub const IS_UPSERT: &str = "x-ms-documentdb-is-upsert";
     pub const OFFER_THROUGHPUT: &str = "x-ms-offer-throughput";
     pub const OFFER_AUTOPILOT_SETTINGS: &str = "x-ms-cosmos-offer-autopilot-settings";
     pub const PRIORITY_LEVEL: &str = "x-ms-cosmos-priority-level";
diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs
index db1ffc0740f..a474d2061d1 100644
--- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs
+++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs
@@ -466,13 +466,16 @@ impl CosmosOperation {
 
     /// Upserts (creates or replaces) an item (document) in a container.
     ///
-    /// The `ItemReference` contains the container, partition key, and item identifier,
-    /// providing all the information needed for the operation.
+    /// The Cosmos DB REST API treats upsert as a `POST` to the collection feed
+    /// (same as create), so this takes a `ContainerReference` and `PartitionKey`
+    /// rather than an `ItemReference`.
     /// Use `with_body()` to provide the document JSON.
     /// If an item with the same ID exists, it will be replaced; otherwise, a new item is created.
-    pub fn upsert_item(item: ItemReference) -> Self {
-        let partition_key = item.partition_key().clone();
-        Self::new(OperationType::Upsert, item).with_partition_key(partition_key)
+    pub fn upsert_item(container: ContainerReference, partition_key: PartitionKey) -> Self {
+        let resource_ref: CosmosResourceReference = CosmosResourceReference::from(container)
+            .with_resource_type(ResourceType::Document)
+            .into_feed_reference();
+        Self::new(OperationType::Upsert, resource_ref).with_partition_key(partition_key)
     }
 
     /// Replaces an existing item (document) in a container.
diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/options/driver_options.rs b/sdk/cosmos/azure_data_cosmos_driver/src/options/driver_options.rs
index cd5af0d6928..de8a3aeb2ea 100644
--- a/sdk/cosmos/azure_data_cosmos_driver/src/options/driver_options.rs
+++ b/sdk/cosmos/azure_data_cosmos_driver/src/options/driver_options.rs
@@ -5,7 +5,10 @@
 
 use std::sync::Arc;
 
-use crate::{models::AccountReference, options::OperationOptions};
+use crate::{
+    models::AccountReference,
+    options::{OperationOptions, Region},
+};
 
 /// Configuration options for a Cosmos DB driver instance.
 ///
@@ -43,6 +46,12 @@ pub struct DriverOptions {
     account: AccountReference,
     /// Driver-level operation options (e.g., consistency, excluded regions, failover, session retry).
     operation_options: Arc<OperationOptions>,
+    /// Preferred regions for routing, ordered by proximity to the application.
+    ///
+    /// When non-empty, read and write endpoint lists are reordered so that
+    /// endpoints matching these regions appear first. Regions that don't match
+    /// any account endpoint are silently skipped.
+    preferred_regions: Vec<Region>,
 }
 
 impl DriverOptions {
@@ -62,6 +71,11 @@ impl DriverOptions {
     pub fn operation_options(&self) -> &Arc<OperationOptions> {
         &self.operation_options
     }
+
+    /// Returns the preferred regions for routing.
+    pub fn preferred_regions(&self) -> &[Region] {
+        &self.preferred_regions
+    }
 }
 
 /// Builder for creating [`DriverOptions`].
@@ -73,6 +87,7 @@ impl DriverOptions {
 pub struct DriverOptionsBuilder {
     account: AccountReference,
     operation_options: Option<OperationOptions>,
+    preferred_regions: Vec<Region>,
 }
 
 impl DriverOptionsBuilder {
@@ -81,6 +96,7 @@ impl DriverOptionsBuilder {
         Self {
             account,
             operation_options: None,
+            preferred_regions: Vec::new(),
         }
     }
 
@@ -90,11 +106,22 @@ impl DriverOptionsBuilder {
         self
     }
 
+    /// Sets the preferred regions for routing.
+    ///
+    /// Regions should be ordered by proximity to the application (closest first).
+    /// The driver reorders endpoint lists to prefer these regions for both reads
+    /// and writes. Regions not present in the account are silently skipped.
+    pub fn with_preferred_regions(mut self, regions: Vec<Region>) -> Self {
+        self.preferred_regions = regions;
+        self
+    }
+
     /// Builds the [`DriverOptions`].
     pub fn build(self) -> DriverOptions {
         DriverOptions {
             account: self.account,
             operation_options: Arc::new(self.operation_options.unwrap_or_default()),
+            preferred_regions: self.preferred_regions,
         }
     }
 }
@@ -170,5 +197,17 @@ mod tests {
             .operation_options()
             .read_consistency_strategy
             .is_none());
+        assert!(options.preferred_regions().is_empty());
+    }
+
+    #[test]
+    fn builder_sets_preferred_regions() {
+        let regions = vec![Region::WEST_US_2, Region::EAST_US];
+
+        let options = DriverOptionsBuilder::new(test_account())
+            .with_preferred_regions(regions.clone())
+            .build();
+
+        assert_eq!(options.preferred_regions(), &regions);
     }
 }